The Pulsar Node.js client

You can use a Pulsar Node.js client to create producers, consumers, and readers. For Pulsar features that Python clients support, see Client Feature Matrix.

For 1.3.0 or later versions, type definitions used in TypeScript are available.

Installation

Install the pulsar-client library via npm:

npm install pulsar-client

For more information, see README.

note

This library only works in Node.js 10.x or later versions because it uses the node-addon-api module.

Connection URLs

To connect to Pulsar using client libraries, you need to specify a Pulsar protocol URL.

You can assign Pulsar protocol URLs to specific clusters and use the pulsar scheme. The following is an example of localhost with the default port 6650:

pulsar://localhost:6650

If you have multiple brokers, separate IP:port by commas:

pulsar://localhost:6550,localhost:6651,localhost:6652

If you use mTLS authentication, add +ssl in the scheme:

pulsar+ssl://pulsar.us-west.example.com:6651

Release notes

For the changelog of Pulsar Node.js clients, see release notes.

Create a client

To interact with Pulsar, you first need a client object. You can create a client instance using a new operator and the Client method, passing in a client options object (more on configuration below).

Here is an example:

const Pulsar = require('pulsar-client');
(async () => {
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });
  await client.close();
})();

Client configuration

The following configurable parameters are available for Pulsar clients:

Producers

Pulsar producers publish messages to Pulsar topics. You can configure Node.js producers using a producer configuration object.

Here is an example:

const producer = await client.createProducer({
  topic: 'my-topic', // or 'my-tenant/my-namespace/my-topic' to specify topic's tenant and namespace
});
await producer.send({
  data: Buffer.from("Hello, Pulsar"),
});
await producer.close();

Promise operation

When you create a new Pulsar producer, the operation returns Promise object and get producer instance or an error through executor function. In this example, use await operator instead of executor function.

Producer operations

Pulsar Node.js producers have the following methods available:

Producer configuration

Producer example

This example creates a Node.js producer for the my-topic topic and sends 10 messages to that topic:

const Pulsar = require('pulsar-client');
(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });
  // Create a producer
  const producer = await client.createProducer({
    topic: 'my-topic',
  });
  // Send messages
  for (let i = 0; i < 10; i += 1) {
    const msg = `my-message-${i}`;
    producer.send({
      data: Buffer.from(msg),
    });
    console.log(`Sent message: ${msg}`);
  }
  await producer.flush();
  await producer.close();
  await client.close();
})();

Consumers

Pulsar consumers subscribe to one or more Pulsar topics and listen for incoming messages produced on that topic/those topics. You can configure Node.js consumers using a consumer configuration object.

Here is an example:

const consumer = await client.subscribe({
  topic: 'my-topic',
  subscription: 'my-subscription',
});
const msg = await consumer.receive();
console.log(msg.getData().toString());
consumer.acknowledge(msg);
await consumer.close();

Promise operation

When you create a new Pulsar consumer, the operation returns Promise object and get consumer instance or an error through executor function. In this example, use await operator instead of executor function.

Consumer operations

Pulsar Node.js consumers have the following methods available:

Consumer configuration

Consumer example

This example creates a Node.js consumer with the my-subscription subscription on the my-topic topic, receives messages, prints the content that arrive, and acknowledges each message to the Pulsar broker for 10 times:

const Pulsar = require('pulsar-client');
(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });
  // Create a consumer
  const consumer = await client.subscribe({
    topic: 'my-topic',
    subscription: 'my-subscription',
    subscriptionType: 'Exclusive',
  });
  // Receive messages
  for (let i = 0; i < 10; i += 1) {
    const msg = await consumer.receive();
    console.log(msg.getData().toString());
    consumer.acknowledge(msg);
  }
  await consumer.close();
  await client.close();
})();

Instead, a consumer can be created with listener to process messages.

// Create a consumer
const consumer = await client.subscribe({
  topic: 'my-topic',
  subscription: 'my-subscription',
  subscriptionType: 'Exclusive',
  listener: (msg, msgConsumer) => {
    console.log(msg.getData().toString());
    msgConsumer.acknowledge(msg);
  },
});

note

Pulsar Node.js client uses AsyncWorker. Asynchronous operations such as creating consumers/producers and receiving/sending messages are performed in worker threads. Until completion of these operations, worker threads are blocked. Since there are only 4 worker threads by default, a called method may never be complete. To avoid this situation, you can set UV_THREADPOOL_SIZE to increase the number of worker threads, or define listener instead of calling receive() many times.

Readers

Pulsar readers process messages from Pulsar topics. Readers are different from consumers because with readers you need to explicitly specify which message in the stream you want to begin with (consumers, on the other hand, automatically begin with the most recently unacked message). You can configure Node.js readers using a reader configuration object.

Here is an example:

const reader = await client.createReader({
  topic: 'my-topic',
  startMessageId: Pulsar.MessageId.earliest(),
});
const msg = await reader.readNext();
console.log(msg.getData().toString());
await reader.close();

Reader operations

Pulsar Node.js readers have the following methods available:

Reader configuration

Reader example

This example creates a Node.js reader with the my-topic topic, reads messages, and prints the content that arrive for 10 times:

const Pulsar = require('pulsar-client');
(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
    operationTimeoutSeconds: 30,
  });
  // Create a reader
  const reader = await client.createReader({
    topic: 'my-topic',
    startMessageId: Pulsar.MessageId.earliest(),
  });
  // read messages
  for (let i = 0; i < 10; i += 1) {
    const msg = await reader.readNext();
    console.log(msg.getData().toString());
  }
  await reader.close();
  await client.close();
})();

Messages

In Pulsar Node.js client, you have to construct producer message objects for producers.

Here is an example of a message:

const msg = {
  data: Buffer.from('Hello, Pulsar'),
  partitionKey: 'key1',
  properties: {
    'foo': 'bar',
  },
  eventTimestamp: Date.now(),
  replicationClusters: [
    'cluster1',
    'cluster2',
  ],
}
await producer.send(msg);

The following keys are available for producer message objects:

Message object operations

In Pulsar Node.js client, you can receive (or read) message objects as consumers (or readers).

The message object has the following methods available:

Message ID object operations

In Pulsar Node.js client, you can get message id objects from message objects.

The message id object has the following methods available:

The client has a static method of message id object. You can access it as Pulsar.MessageId.someStaticMethod.

The following static methods are available for the message id object:

End-to-end encryption

Pulsar encryption allows applications to encrypt messages at producers and decrypt messages at consumers. See Get started for more details.