ayeo-flex-org / pulsar-flex Goto Github PK

const { Producer, Consumer, logLevel } = require('pulsar-flex')

const producer = new Producer({
  topic: "persistent://public/default/my-topic",
  discoveryServers: ['pulsar-host:6650'],
  //If you dont provide any jwt token it will use no auth
  jwt: process.env.JWT_TOKEN,
  producerAccessMode: Producer.ACCESS_MODES.SHARED,
  logLevel: logLevel.INFO
  // you can also provide logCreator function
})

const consumer = new Consumer({
  topic: "persistent://public/default/my-topic",
  subscription: "my-subscription",
  discoveryServers: ['pulsar-host:6650'],
  jwt: process.env.JWT_TOKEN,
  subType: Consumer.SUB_TYPES.EXCLUSIVE,
  consumerName: 'Consumer name',
  receiveQueueSize: 1000,
  logLevel: logLevel.INFO,
  // you can also provide logCreator function
})

const run = async () => {
  await producer.create();
  // you can also send single message using sendMessage function
  await producer.sendBatch({messages: [
    {
      properties: {pulsar: "flex"}, 
      payload: 'Ayeo' 
    },
    {
      properties: {pulsar: "flex"},
      payload: 'Ayeo'
    }
  ]});

  await consumer.subscribe();

  consumer.onStateChange(({previousState, newState}) => {
      console.log(`Consumer state has changed from ${previousState} to ${newState}.`);
    };
  );

  await consumer.run({
    onMessage: async ({ ack, message, properties, redeliveryCount }) => {
      await ack(); // Default is individual ack
      // await ack({type: Consumer.ACK_TYPES.CUMULATIVE});
      console.log({
        message,
        properties,
        redeliveryCount,
      })
    }, autoAck: false, // specify true in order to use automaticAck
  });
}

run().catch(console.error)