Usee Node.js client
Create a producer​
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();
When you create a new Pulsar producer, the operation returns Promise
object and get producer instance or an error through executor function. In the above example, use await
operator instead of executor function.
Producer operations​
Pulsar Node.js producers have the following methods available:
Method | Description | Return type |
---|---|---|
send(Object) | Publishes a message to the producer's topic. When the message is successfully acknowledged by the Pulsar broker, or an error is thrown, the Promise object whose result is the message ID runs executor function. | Promise<Object> |
flush() | Sends message from send queue to Pulsar broker. When the message is successfully acknowledged by the Pulsar broker, or an error is thrown, the Promise object runs executor function. | Promise<null> |
close() | Closes the producer and releases all resources allocated to it. Once close() is called, no more messages are accepted from the publisher. This method returns a Promise object. It runs the executor function when all pending publish requests are persisted by Pulsar. If an error is thrown, no pending writes are retried. | Promise<null> |
getProducerName() | Getter method of the producer name. | string |
getTopic() | Getter method of the name of the topic. | string |
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();
})();
Create a consumer​
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();
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:
Method | Description | Return type |
---|---|---|
receive() | Receives a single message from the topic. When the message is available, the Promise object run executor function and get message object. | Promise<Object> |
receive(Number) | Receives a single message from the topic with specific timeout in milliseconds. | Promise<Object> |
acknowledge(Object) | Acknowledges a message to the Pulsar broker by message object. | void |
acknowledgeId(Object) | Acknowledges a message to the Pulsar broker by message ID object. | void |
acknowledgeCumulative(Object) | Acknowledges all the messages in the stream, up to and including the specified message. The acknowledgeCumulative method returns void, and send the ack to the broker asynchronously. After that, the messages are not redelivered to the consumer. Cumulative acking can not be used with a shared subscription type. | void |
acknowledgeCumulativeId(Object) | Acknowledges all the messages in the stream, up to and including the specified message ID. | void |
negativeAcknowledge(Message) | Negatively acknowledges a message to the Pulsar broker by message object. | void |
negativeAcknowledgeId(MessageId) | Negatively acknowledges a message to the Pulsar broker by message ID object. | void |
close() | Closes the consumer, disabling its ability to receive messages from the broker. | Promise<null> |
unsubscribe() | Unsubscribes the subscription. | Promise<null> |
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);
},
});
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.
Create a reader​
Pulsar 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:
Method | Description | Return type |
---|---|---|
readNext() | Receives the next message on the topic (analogous to the receive method for consumers). When the message is available, the Promise object run executor function and get message object. | Promise<Object> |
readNext(Number) | Receives a single message from the topic with specific timeout in milliseconds. | Promise<Object> |
hasNext() | Return whether the broker has next message in target topic. | Boolean |
close() | Closes the reader, disabling its ability to receive messages from the broker. | Promise<null> |
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();
})();