Support get the producer name of a message

index.d.ts

@@ -170,6 +170,7 @@ export class Message {
   getRedeliveryCount(): number;
   getPartitionKey(): string;
   getOrderingKey(): string;
+  getProducerName(): string;
 }
 
 export class MessageId {

src/Message.cc

@@ -46,7 +46,8 @@ Napi::Object Message::Init(Napi::Env env, Napi::Object exports) {
        InstanceMethod("getEventTimestamp", &Message::GetEventTimestamp),
        InstanceMethod("getRedeliveryCount", &Message::GetRedeliveryCount),
        InstanceMethod("getPartitionKey", &Message::GetPartitionKey),
-       InstanceMethod("getOrderingKey", &Message::GetOrderingKey)});
+       InstanceMethod("getOrderingKey", &Message::GetOrderingKey),
+       InstanceMethod("getProducerName", &Message::GetProducerName)});
 
   constructor = Napi::Persistent(func);
   constructor.SuppressDestruct();
@@ -147,6 +148,14 @@ Napi::Value Message::GetOrderingKey(const Napi::CallbackInfo &info) {
   return Napi::String::New(env, pulsar_message_get_orderingKey(this->cMessage.get()));
 }
 
+Napi::Value Message::GetProducerName(const Napi::CallbackInfo &info) {
+  Napi::Env env = info.Env();
+  if (!ValidateCMessage(env)) {
+    return env.Null();
+  }
+  return Napi::String::New(env, pulsar_message_get_producer_name(this->cMessage.get()));
+}
+
 bool Message::ValidateCMessage(Napi::Env env) {
   if (this->cMessage.get()) {
     return true;

src/Message.h

@@ -45,6 +45,7 @@ class Message : public Napi::ObjectWrap<Message> {
   Napi::Value GetEventTimestamp(const Napi::CallbackInfo &info);
   Napi::Value GetPartitionKey(const Napi::CallbackInfo &info);
   Napi::Value GetOrderingKey(const Napi::CallbackInfo &info);
+  Napi::Value GetProducerName(const Napi::CallbackInfo &info);
   Napi::Value GetRedeliveryCount(const Napi::CallbackInfo &info);
   bool ValidateCMessage(Napi::Env env);
 

tests/end_to_end.test.js

@@ -40,11 +40,13 @@
         } : {}),
       });
 
+      const producerName = 'test-producer';
       const topic = 'persistent://public/default/produce-consume';
       const producer = await client.createProducer({
         topic,
         sendTimeoutMs: 30000,
         batchingEnabled: true,
+        producerName,
       });
       expect(producer).not.toBeNull();
 
@@ -70,6 +72,7 @@
       for (let i = 0; i < 10; i += 1) {
         const msg = await consumer.receive();
         consumer.acknowledge(msg);
+        expect(msg.getProducerName()).toBe(producerName);
         results.push(msg.getData().toString());
       }
       expect(lodash.difference(messages, results)).toEqual([]);
@@ -315,7 +318,7 @@
       });
 
       let consumer2Recv = 0;
       while (true) {
         try {
           const msg = await consumer2.receive(3000);
           await new Promise((resolve) => setTimeout(resolve, 10));
@@ -360,7 +363,7 @@
         topic,
         startMessageId: Pulsar.MessageId.earliest(),
         receiverQueueSize: 10,
         listener: async (message, reader) => {
           await new Promise((resolve) => setTimeout(resolve, 10));
           reader1Recv += 1;
         },
@@ -392,7 +395,7 @@
       await client.close();
     });
 
     test('Message Listener error handling', async () => {
       const client = new Pulsar.Client({
         serviceUrl: 'pulsar://localhost:6650',
       });
@@ -428,7 +431,7 @@
         subscription: 'sync',
         subscriptionType: 'Shared',
         subscriptionInitialPosition: 'Earliest',
         listener: (message, messageConsumer) => {
           throw new Error('consumer1 callback expected error');
         },
       });
@@ -438,7 +441,7 @@
         subscription: 'async',
         subscriptionType: 'Shared',
         subscriptionInitialPosition: 'Earliest',
         listener: async (message, messageConsumer) => {
           throw new Error('consumer2 callback expected error');
         },
       });