Pulsar WebSocket API

推荐使用单机模式的 Pulsar 进行开发，在环境中启用 WebSocket 服务。

在非单机模式下，有两种方法可以部署 WebSocket 服务：

• 嵌入 Pulsar Broker
• 作为一个的组件

在这种模式下，WebSocket 服务会使用已经在 broker 中运行的 HTTP 服务。 要启用此模式，需在安装目录下的 文件中设置 参数。

作为一个独立的组件

在这种模式下，WebSocket 会作为单独的服务在 Pulsar 上运行。 运行此模式，需在 conf/websocket.conf 文件中进行配置。 You’ll need to set at least the following parameters:

• webServicePort

下面是一个示例：

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

安全设置

To enable TLS encryption on WebSocket service:

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

配置完成后，你可以使用 命令来启动服务：

$ bin/pulsar-daemon start websocket

Pulsar 的 WebSocket API 提供三个端点，用于生产消息、和阅读消息。

所有通过 WebSocket API 的数据都使用 JSON 进行交互。

认证

Broswer javascript WebSocket client

Use the query param token transport the authentication token.

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

Producer 端

Producer 端需要在 URL 中指定租户、命名空间和 topic，例如：

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

查询参数

发布消息

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

响应成功示例

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

响应失败示例

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

查询参数

注意：以上参数（pullMode 除外）适用于 WebSocket 服务的内部 consumer。 因此，即使客户端没有在 WebSocket 上消费，只要消息进入接收队列，就会受到传递设置的约束。

接收消息

Server will push messages on the WebSocket session:

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

ACK 确认消息

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

{
  "messageId": "CAAQAw=="
}

Negatively acknowledging messages

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

流量控制

推送模式

默认情况下（pullMode=false），consumer 端使用 receiverQueueSize 参数设置内部接收队列的大小，并限制传递到 WebSocket 客户端的未确认消息数。 在这种模式下，如果不发送消息确认，发送到 WebSocket 客户端的消息达到 receiverQueueSize时，Pulsar WebSocket 将停止发送消息。

拉取模式

如果设置 pullMode 为 true，则 WebSocket 客户端需要使用 permit 命令允许 Pulsar WebSocket 服务发送更多消息。

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

注意：在这种模式下，可以在不同的连接中确认消息。

Check if reach end of topic

Consumer can check if it has reached end of topic by sending isEndOfTopic request.

Request

{
  "type": "isEndOfTopic"
}

Response

{
   "endOfTopic": "true/false"
 }

Reader 端

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

查询参数

接收消息

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
}

ACK 确认消息

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 acknowledgements, Pulsar WebSocket service will stop sending messages after reaching the pendingMessages limit.

Check if reach end of topic

Request

{
  "type": "isEndOfTopic"
}

Response

{
   "endOfTopic": "true/false"
 }

错误代码

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

应用程序负责在后台重新建立 WebSocket 连接。

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

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

$ pip install websocket-client

You can also download it from .

Python producer

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

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)
# Send one message as JSON
ws.send(json.dumps({
    'payload' : base64.b64encode('Hello World'),
    'properties': {
        'key1' : 'value1',
        'key2' : 'value2'
    },
    'context' : 5
}))
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'
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 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

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));