Merge pull request #76 from ayeo-flex-org/task/fixSequenceId

moved the sequenceId to increase before sending, so it always increases

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pulsar-flex",
-  "version": "1.0.1-beta.6",
+  "version": "1.0.1-beta.7",
   "description": "A package that natively supports pulsar api",
   "main": "src/index.js",
   "scripts": {

src/producer/index.js

@@ -126,6 +126,7 @@ class Producer {
         throw new errors.PulsarFlexProducerCreationError({
           message: 'Cannot send messages over not connected producer',
         });
+      this._sequenceId++;
       const { command } = await services.sendMessage({
         producerId: this._producerId,
         producerName: this._producerName,
@@ -160,7 +161,6 @@ class Producer {
         );
       });
     }
-    this._sequenceId++;
     return true;
   };
 
@@ -182,6 +182,7 @@ class Producer {
         throw new errors.PulsarFlexProducerCreationError({
           message: 'Cannot send batch over not connected producer',
         });
+      this._sequenceId++;
       const { command } = await services.sendBatch({
         producerId: this._producerId,
         producerName: this._producerName,
@@ -212,7 +213,6 @@ class Producer {
         );
       });
     }
-    this._sequenceId++;
     return true;
   };
 

test/producer/index-spec.js

@@ -281,6 +281,36 @@ describe('Producer tests', function () {
       await producer.close();
     });
   });
+  describe('on sending messages check that sequenceId increases', function () {
+    it('should increase the sequenceId, per message on send message', async function () {
+      const expectedSequenceId = 50;
+      const producer = new Producer({
+        discoveryServers,
+        jwt,
+        topic,
+      });
+      await producer.create();
+      for (let i = 0; i < expectedSequenceId; i++) {
+        await producer.sendMessage({ payload: 'galrose', properties: { sinai: 'noob' } });
+      }
+      assert.deepStrictEqual(producer._sequenceId, expectedSequenceId);
+    });
+    it('should increase the sequenceId, per batch on send batch', async function () {
+      const expectedSequenceId = 50;
+      const producer = new Producer({
+        discoveryServers,
+        jwt,
+        topic,
+      });
+      await producer.create();
+      for (let i = 0; i < expectedSequenceId; i++) {
+        await producer.sendBatch({
+          messages: [{ payload: 'galrose', properties: { sinai: 'noob' } }],
+        });
+      }
+      assert.deepStrictEqual(producer._sequenceId, expectedSequenceId);
+    });
+  });
   describe('on sending message should contain payload and properties', function () {
     it('should not throw exception', async function () {
       const topic = 'public/default/testSendMessage';