How to use pulsar-client - 8 common examples
To help you get started, we’ve selected a few pulsar-client examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
apache / pulsar-client-node / examples / consumer.js View on Github 
(async () => {
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar://localhost:6650',
operationTimeoutSeconds: 30,
});
// Create a consumer
const consumer = await client.subscribe({
topic: 'persistent://public/default/my-topic',
subscription: 'sub1',
subscriptionType: 'Shared',
ackTimeoutMs: 10000,
});
// Receive messages
for (let i = 0; i < 10; i += 1) {
const msg = await consumer.receive();
console.log(msg.getData().toString());(async () => {
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar://localhost:6650',
operationTimeoutSeconds: 30,
});
// Create a reader
const reader = await client.createReader({
topic: 'persistent://public/default/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());
}(async () => {
const auth = new Pulsar.AuthenticationTls({
certificatePath: '/path/to/client.crt',
privateKeyPath: '/path/to/client.key',
});
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar+ssl://localhost:6651',
authentication: auth,
tlsTrustCertsFilePath: '/path/to/server.crt',
});
// Create a consumer
const consumer = await client.subscribe({
topic: 'persistent://public/default/my-topic',
subscription: 'sub1',
subscriptionType: 'Shared',
});
// Receive messages
for (let i = 0; i < 10; i += 1) {
const msg = await consumer.receive();
console.log(msg.getData().toString());(async () => {
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar://localhost:6650',
operationTimeoutSeconds: 30,
});
// Create a reader
const reader = await client.createReader({
topic: 'persistent://public/default/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();
})();(async () => {
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar://localhost:6650',
operationTimeoutSeconds: 30,
});
// Create a producer
const producer = await client.createProducer({
topic: 'persistent://public/default/my-topic',
sendTimeoutMs: 30000,
batchingEnabled: true,
});
// Send messages
for (let i = 0; i < 10; i += 1) {
const msg = `my-message-${i}`;
producer.send({
data: Buffer.from(msg),(async () => {
const auth = new Pulsar.AuthenticationTls({
certificatePath: '/path/to/client.crt',
privateKeyPath: '/path/to/client.key',
});
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar+ssl://localhost:6651',
authentication: auth,
tlsTrustCertsFilePath: '/path/to/server.crt',
});
// Create a producer
const producer = await client.createProducer({
topic: 'persistent://public/default/my-topic',
});
// Send messages
for (let i = 0; i < 10; i += 1) {
const msg = `my-message-${i}`;
producer.send({
data: Buffer.from(msg),
});(async () => {
const auth = new Pulsar.AuthenticationTls({
certificatePath: '/path/to/client.crt',
privateKeyPath: '/path/to/client.key',
});
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar+ssl://localhost:6651',
authentication: auth,
tlsTrustCertsFilePath: '/path/to/server.crt',
});
// Create a producer
const producer = await client.createProducer({
topic: 'persistent://public/default/my-topic',
});(async () => {
const auth = new Pulsar.AuthenticationTls({
certificatePath: '/path/to/client.crt',
privateKeyPath: '/path/to/client.key',
});
// Create a client
const client = new Pulsar.Client({
serviceUrl: 'pulsar+ssl://localhost:6651',
authentication: auth,
tlsTrustCertsFilePath: '/path/to/server.crt',
});
// Create a consumer
const consumer = await client.subscribe({
topic: 'persistent://public/default/my-topic',
subscription: 'sub1',
subscriptionType: 'Shared',
