Listening for messages

[mhaagens]

How can I listen to incoming messages? The listener property on consumer doesn't seem to do anything.

[david-streamlio]

@mhaagens In order to receive messages, you will want to use the subscribe method as shown in the following example: https://github.com/apache/pulsar-client-node/blob/master/examples/consumer.js

[mhaagens]

Thanks for the quick reply! I am using subscribe, but as far I can see from the example you have to manually call receive on the subscriber in a loop to get messages?

[david-streamlio]

@mhaagens You are correct. The subscribe command is used to register your consumer on the Pulsar broker, so it will know that you want to receive the messages. After you had subscribed, you have to make a call as shown on line 39 of the example code; const msg = await consumer.receive(); to get the message. It is a blocking call that will deliver one message (if and when it is available).

HTH

[mhaagens]

Okay, so am I understanding it correctly that there’s no way to set up a listener that processes messages as they come in? Similar to this in kafka-Node;

consumer.on('message', function (message) { console.log(message); });

[david-streamlio]

@mhaagens Not yet, but there appears to be a PR in process now to add this capability. #56

[mhaagens]

Awesome, thanks!

[mhaagens]

Would this be a solution in the meantime?

while (true) {
   const msg = await consumer.receive();
   console.log(msg.getData().toString());
   consumer.acknowledge(msg);
}

[david-streamlio]

@mhaagens Yes.

[yosiat]

Hi!

I tried the listener implementation with the following code, that listens forever to messages and once any key received we close the consumer and client (in real-life: instead of listening to any key, we will listen to shutdown signals) -

const Pulsar = require('pulsar-client');

const readline = require('readline');

const rl = new readline.createInterface({
  input: process.stdin,
  output: process.stdout
})

function prompt() {
  return new Promise((resolve) => {
    rl.question(`press any key to stop consuming`, () => resolve());
  })
}

(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,
    listener: function(msg, consumer) {
      console.log(msg.getData().toString());
      consumer.acknowledge(msg);
    }
  });

  await prompt();
  await consumer.close();
  await client.close();

})();

I have producer which emits at regular intervals (after some time, fails on this - #78 and I restart it).

The consumer after some random time, just closes the connection. I added std::cout to the code and I see Consumer deconstructor is called for every message (which makes sense, since we create for every message a consumer) and the Client deconstructor is called after a while.

I did some changes in order to solve this and it works now, except the listener don't get the consumer instance.

Changes are here - master...yosiat:consumer-listener-bug

If my changes make sense to you, I am happy to open a PR and move the discussion over there.

[sijie]

@yosiat yes. this approach looks good to me. Can you create a pull request for your proposed change?

[yosiat]

@sijie created a PR, here: #83