Pulsar WebSocket API

Pulsar WebSocket API provides a simple way to interact with Pulsar using languages that do not have an official client library. You can use Pulsar WebSocket API with any WebSocket client library. See Python and Node.js examples for more details.

Through WebSocket, you can publish and consume messages and use features available on the Client Feature Matrix page.

Run the WebSocket service

The standalone variant of Pulsar that we recommend using for local development already has the WebSocket service enabled.

In non-standalone mode, there are two ways to deploy the WebSocket service:

• embedded with a Pulsar broker
• as a separate component

Embedded with a Pulsar broker

In this mode, the WebSocket service will run within the same HTTP service that’s already running in the broker. To enable this mode, set the webSocketServiceEnabled parameter in the conf/broker.conf configuration file in your installation.

webSocketServiceEnabled=true

As a separate component

In this mode, the WebSocket service will be run from a Pulsar broker as a separate service. Configuration for this mode is handled in the conf/websocket.conf configuration file. You’ll need to set at least the following parameters:

• configurationMetadataStoreUrl
• webServicePort
• clusterName

Here’s an example:

configurationMetadataStoreUrl=zk1:2181,zk2:2181,zk3:2181
webServicePort=8080
clusterName=my-cluster

Security settings

To enable TLS encryption on WebSocket service, configure the following parameters in the conf/broker.conf file.

tlsEnabled=true
tlsAllowInsecureConnection=false
tlsCertificateFilePath=/path/to/client-websocket.cert.pem
tlsKeyFilePath=/path/to/client-websocket.key-pk8.pem
tlsTrustCertsFilePath=/path/to/ca.cert.pem

To enable encryption at rest on WebSocket service, add CryptoKeyReaderFactory factory class in classpath which will create CryptoKeyReader for WebSocket and that helps to load encryption keys for producer/consumer.

cryptoKeyReaderFactoryClassName=org.apache.pulsar.MyCryptoKeyReaderFactoryClassImpl

Start the broker

When the configuration is set, you can start the service using the pulsar-daemon tool:

bin/pulsar-daemon start websocket

Release notes

For the changelog of Pulsar WebSocket APIs, see release notes.

API Reference

Pulsar’s WebSocket API offers three endpoints for producing, consuming, and reading messages.

All exchanges via the WebSocket API use JSON.

Authentication

Browser javascript WebSocket client

Use the query param token to transport the authentication token.

ws://broker-service-url:8080/path?token=token

Producer endpoint

The producer endpoint requires you to specify a tenant, namespace, and topic in the URL:

ws://broker-service-url:8080/ws/v2/producer/persistent/:tenant/:namespace/:topic

Query param

Publish a message

{
  "payload": "SGVsbG8gV29ybGQ=",
  "properties": {"key1": "value1", "key2": "value2"},
  "context": "1"
}

Example success response

{
   "result": "ok",
   "messageId": "CAAQAw==",
   "context": "1"
 }

Example failure response

 {
   "result": "send-error:3",
   "errorMsg": "Failed to de-serialize from JSON",
   "context": "1"
 }

Consumer endpoint

The consumer endpoint requires you to specify a tenant, namespace, and topic, as well as a subscription, in the URL:

ws://broker-service-url:8080/ws/v2/consumer/persistent/:tenant/:namespace/:topic/:subscription

Query param

note

These parameters (except pullMode) apply to the internal consumers of the WebSocket service. So messages will be subject to the redelivery settings as soon as they get into the receive queue, even if the client doesn’t consume on the WebSocket.

Receive messages

Server will push messages on the WebSocket session:

{
  "messageId": "CAMQADAA",
  "payload": "hvXcJvHW7kOSrUn17P2q71RA5SdiXwZBqw==",
  "properties": {},
  "publishTime": "2021-10-29T16:01:38.967-07:00",
  "redeliveryCount": 0,
  "encryptionContext": {
    "keys": {
      "client-rsa.pem": {
        "keyValue": "jEuwS+PeUzmCo7IfLNxqoj4h7txbLjCQjkwpaw5AWJfZ2xoIdMkOuWDkOsqgFmWwxiecakS6GOZHs94x3sxzKHQx9Oe1jpwBg2e7L4fd26pp+WmAiLm/ArZJo6JotTeFSvKO3u/yQtGTZojDDQxiqFOQ1ZbMdtMZA8DpSMuq+Zx7PqLo43UdW1+krjQfE5WD+y+qE3LJQfwyVDnXxoRtqWLpVsAROlN2LxaMbaftv5HckoejJoB4xpf/dPOUqhnRstwQHf6klKT5iNhjsY4usACt78uILT0pEPd14h8wEBidBz/vAlC/zVMEqiDVzgNS7dqEYS4iHbf7cnWVCn3Hxw==",
        "metadata": {}
      }
    },
    "param": "Tfu1PxVm6S9D3+Hk",
    "compressionType": "NONE",
    "uncompressedMessageSize": 0,
    "batchSize": {
      "empty": false,
      "present": true
    }
  }
}

Below are the parameters in the WebSocket consumer response.

• General parameters

  Key	Type	Required?	Explanation
  messageId	string	yes	Message ID
  payload	string	yes	Base-64 encoded payload
  publishTime	string	yes	Publish timestamp
  redeliveryCount	number	yes	Number of times this message was already delivered
  properties	key-value pairs	no	Application-defined properties
  key	string	no	Original routing key set by producer
  encryptionContext	EncryptionContext	no	Encryption context that consumers can use to decrypt received messages
  param	string	no	Initialization vector for cipher (Base64 encoding)
  batchSize	string	no	Number of entries in a message (if it is a batch message)
  uncompressedMessageSize	string	no	Message size before compression
  compressionType	string	no	Algorithm used to compress the message payload

• encryptionContext related parameter

  Key	Type	Required?	Explanation
  keys	key-EncryptionKey pairs	yes	Key in key-EncryptionKey pairs is an encryption key name. Value in key-EncryptionKey pairs is an encryption key object.

• encryptionKey related parameters

  Key	Type	Required?	Explanation
  keyValue	string	yes	Encryption key (Base64 encoding)
  metadata	key-value pairs	no	Application-defined metadata

Acknowledge the message

Consumer needs to acknowledge the successful processing of the message to have the Pulsar broker delete it.

{
  "messageId": "CAAQAw=="
}

Negatively acknowledge messages

{
  "type": "negativeAcknowledge",
  "messageId": "CAAQAw=="
}

Flow control

Push Mode

By default (pullMode=false), the consumer endpoint will use the receiverQueueSize parameter both to size its internal receive queue and to limit the number of unacknowledged messages that are passed to the WebSocket client. In this mode, if you don’t send acknowledgments, the Pulsar WebSocket service will stop sending messages after reaching receiverQueueSize unacked messages sent to the WebSocket client.

Pull Mode

If you set pullMode to true, the WebSocket client will need to send permit commands to permit the Pulsar WebSocket service to send more messages.

{
  "type": "permit",
  "permitMessages": 100
}

In this mode it’s possible to acknowledge messages in a different connection.

Check if reach the end of topic

Consumers can check if it has reached the end of a topic by sending the isEndOfTopic request.

Request

{
  "type": "isEndOfTopic"
}

Response

{
   "endOfTopic": "true/false"
 }

Reader endpoint

The reader endpoint requires you to specify a tenant, namespace, and topic in the URL:

ws://broker-service-url:8080/ws/v2/reader/persistent/:tenant/:namespace/:topic

Query param

Receiving messages

Server will push messages on the WebSocket session:

{
  "messageId": "CAAQAw==",
  "payload": "SGVsbG8gV29ybGQ=",
  "properties": {"key1": "value1", "key2": "value2"},
  "publishTime": "2016-08-30 16:45:57.785",
  "redeliveryCount": 4
}

Acknowledging the message

In WebSocket, Reader needs to acknowledge the successful processing of the message to have the Pulsar WebSocket service update the number of pending messages. If you don’t send acknowledgments, Pulsar WebSocket service will stop sending messages after reaching the pendingMessages limit.

{
  "messageId": "CAAQAw=="
}

Check if reach the end of topic

Consumers can check if it has reached the end of a topic by sending the isEndOfTopic request.

Request

{
  "type": "isEndOfTopic"
}

Response

{
   "endOfTopic": "true/false"
 }

Error codes

In case of error the server will close the WebSocket session using the following error codes:

The application is responsible for re-establishing a new WebSocket session after a backoff period.

Client examples

Below you’ll find code examples for the Pulsar WebSocket API in Python and Node.js.

Python

This example uses the websocket-client package. You can install it using pip:

pip install websocket-client

You can also download it from PyPI.

Python producer

Here’s an example Python producer that sends a simple message to a Pulsar topic:

import websocket, base64, json
# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'
TOPIC = scheme + '://localhost:8080/ws/v2/producer/persistent/public/default/my-topic'
ws = websocket.create_connection(TOPIC)
# encode message
s = "Hello World"
firstEncoded = s.encode("UTF-8")
binaryEncoded = base64.b64encode(firstEncoded)
payloadString = binaryEncoded.decode('UTF-8')
# Send one message as JSON
ws.send(json.dumps({
    'payload' : payloadString,
    'properties': {
        'key1' : 'value1',
        'key2' : 'value2'
    },
    'context' : 5
}))
response =  json.loads(ws.recv())
if response['result'] == 'ok':
    print( 'Message published successfully')
else:
    print('Failed to publish message:', response)
ws.close()

Python consumer

Here’s an example Python consumer that listens on a Pulsar topic and prints the message ID whenever a message arrives:

import websocket, base64, json
# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'
TOPIC = scheme + '://localhost:8080/ws/v2/consumer/persistent/public/default/my-topic/my-sub'
ws = websocket.create_connection(TOPIC)
while True:
    msg = json.loads(ws.recv())
    if not msg: break
    print( "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload'])))
    # Acknowledge successful processing
    ws.send(json.dumps({'messageId' : msg['messageId']}))
ws.close()

Python reader

Here’s an example Python reader that listens on a Pulsar topic and prints the message ID whenever a message arrives:

import websocket, base64, json
# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'
TOPIC = scheme + '://localhost:8080/ws/v2/reader/persistent/public/default/my-topic'
ws = websocket.create_connection(TOPIC)
while True:
    msg = json.loads(ws.recv())
    if not msg: break
    print ( "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload'])))
    # Acknowledge successful processing
    ws.send(json.dumps({'messageId' : msg['messageId']}))
ws.close()

Node.js

This example uses the ws package. You can install it using npm:

npm install ws

Node.js producer

Here’s an example Node.js producer that sends a simple message to a Pulsar topic:

const WebSocket = require('ws');
// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/producer/persistent/public/default/my-topic`;
const ws = new WebSocket(topic);
var message = {
  "payload" : new Buffer("Hello World").toString('base64'),
  "properties": {
    "key1" : "value1",
    "key2" : "value2"
  },
  "context" : "1"
};
ws.on('open', function() {
  // Send one message
  ws.send(JSON.stringify(message));
});
ws.on('message', function(message) {
  console.log('received ack: %s', message);
});

Node.js consumer

Here’s an example Node.js consumer that listens on the same topic used by the producer above:

const WebSocket = require('ws');
// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/consumer/persistent/public/default/my-topic/my-sub`;
const ws = new WebSocket(topic);
ws.on('message', function(message) {
    var receiveMsg = JSON.parse(message);
    console.log('Received: %s - payload: %s', message, new Buffer(receiveMsg.payload, 'base64').toString());
    var ackMsg = {"messageId" : receiveMsg.messageId};
    ws.send(JSON.stringify(ackMsg));
});

NodeJS reader

const WebSocket = require('ws');
// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/reader/persistent/public/default/my-topic`;
const ws = new WebSocket(topic);
ws.on('message', function(message) {
    var receiveMsg = JSON.parse(message);
    console.log('Received: %s - payload: %s', message, new Buffer(receiveMsg.payload, 'base64').toString());
    var ackMsg = {"messageId" : receiveMsg.messageId};
    ws.send(JSON.stringify(ackMsg));
});