Merge pull request #69 from ayeo-flex-org/bug/fixProducerReconnect

Bug/fix producer handling reconnect & topic unload

package.json

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

src/errors/PulsarFlexProducerAlreadyCreatedError.js

@@ -0,0 +1,9 @@
+class PulsarFlexProducerAlreadyCreatedError extends Error {
+  constructor({ message }) {
+    super();
+    this.message = message;
+    this.name = 'PulsarFlexProducerAlreadyCreatedError';
+  }
+}
+
+module.exports = PulsarFlexProducerAlreadyCreatedError;

src/errors/index.js

@@ -1,5 +1,6 @@
 const PulsarFlexBrokerTimeoutError = require('./PulsarFlexBrokerTimeoutError');
 const PulsarFlexProducerCreationError = require('./PulsarFlexProducerCreationError');
+const PulsarFlexProducerAlreadyCreatedError = require('./PulsarFlexProducerAlreadyCreatedError');
 const PulsarFlexNoPayloadError = require('./PulsarFlexNoPayloadError');
 const PulsarFlexProducerCloseError = require('./PulsarFlexProducerCloseError');
 const PulsarFlexConnectionError = require('./PulsarFlexConnectionError');
@@ -16,6 +17,7 @@ const PulsarFlexConsumerCloseError = require('./PulsarFlexConsumerCloseError');
 module.exports = {
   PulsarFlexBrokerTimeoutError,
   PulsarFlexProducerCreationError,
+  PulsarFlexProducerAlreadyCreatedError,
   PulsarFlexNoPayloadError,
   PulsarFlexProducerCloseError,
   PulsarFlexConnectionError,

src/producer/index.js

@@ -56,7 +56,7 @@ class Producer {
     this._logger.info(`Creating producer to topic: ${this._topic}`);
 
     if (this._connected)
-      throw new errors.PulsarFlexProducerCreationError({
+      throw new errors.PulsarFlexProducerAlreadyCreatedError({
         message: 'Already connected, please close before trying again',
       });
     this._logger.info(`Creating client connection for producer to topic: ${this._topic}`);
@@ -65,7 +65,10 @@ class Producer {
     await this._client.getCnx().addCleanUpListener(() => {
       this._logger.warn(`Starting reconnection because socket ended unexpectedly`);
       this._connected = false;
-      this._created && services.reconnect(this.create).then(() => (this._connected = true));
+      this._created &&
+        services.reconnect(this.create).then(() => {
+          this._connected = true;
+        });
     });
 
     this._logger.info(
@@ -119,6 +122,10 @@ class Producer {
       });
     if (utils.isNil(payload)) throw new errors.PulsarFlexNoPayloadError();
     try {
+      if (!this._connected)
+        throw new errors.PulsarFlexProducerCreationError({
+          message: 'Cannot send messages over not connected producer',
+        });
       const { command } = await services.sendMessage({
         producerId: this._producerId,
         producerName: this._producerName,
@@ -135,6 +142,8 @@ class Producer {
         this._pendingMessageQueue.push({
           func: () =>
             services.sendMessage({
+              connected: this._connected,
+              isResend: true,
               producerId: this._producerId,
               producerName: this._producerName,
               client: this._client,
@@ -169,6 +178,10 @@ class Producer {
         message: 'Pending messages queue size has been exceeded',
       });
     try {
+      if (!this._connected)
+        throw new errors.PulsarFlexProducerCreationError({
+          message: 'Cannot send batch over not connected producer',
+        });
       const { command } = await services.sendBatch({
         producerId: this._producerId,
         producerName: this._producerName,

src/producer/services/producerClose.js

@@ -1,4 +1,3 @@
-const reconnect = require('./reconnect');
 const errors = require('../../errors');
 
 const producerClose = ({ client, create, setConnected, sendResponseMediator }) => {
@@ -7,8 +6,6 @@ const producerClose = ({ client, create, setConnected, sendResponseMediator }) =
     setConnected(false);
     sendResponseMediator.purgeRequests({ error: errors.PulsarFlexProducerCloseError });
     client.getCnx().close();
-    await reconnect(create, setConnected);
-    setConnected(true);
   });
 };
 

src/producer/services/reconnect.js

@@ -1,7 +1,7 @@
 const utils = require('../../utils');
 
 const reconnect = (create) =>
-  create().catch(async () => {
+  create().catch(async (e) => {
     await utils.sleep(5000);
     await reconnect(create);
   });

src/producer/services/resendMessages.js

@@ -1,6 +1,6 @@
 const resendMessages = (client, messageQueue, logger) => {
   client.getResponseEvents().on('producerSuccess', async () => {
-    logger.info('Starting resend message progress');
+    logger.info('Starting resend messages');
 
     while (messageQueue.length > 0) {
       logger.info(`De queueing message from messageQueue current length ${messageQueue.length}`);

src/producer/services/sendMessage.js

@@ -8,6 +8,8 @@ const sendMessage = async ({
   responseMediator,
   payload,
   properties,
+  connected,
+  isResend = false,
 }) => {
   const { sendPayloadCommandRequest } = client.getCnx();
   const messageMetadata = commands.messageMetadata({

test/producer/index-spec.js

@@ -6,6 +6,23 @@ const utils = require('../utils');
 const { jwt, discoveryServers, topic, containerName } = config;
 
 describe('Producer tests', function () {
+  const producer = new Producer({
+    discoveryServers,
+    jwt,
+    topic,
+  });
+  beforeEach(async function () {
+    if (producer._connected) {
+      await producer.close();
+    }
+  });
+
+  afterEach(async function () {
+    if (producer._connected) {
+      await producer.close();
+    }
+  });
+
   describe('on creating & closing ', function () {
     it('should not throw exception', async function () {
       try {
@@ -191,6 +208,46 @@ describe('Producer tests', function () {
       });
       await producer.close();
     });
+    it('should resend messages after reconnect', async function () {
+      await producer.create();
+      let msgsSent = 0;
+      await new Promise(async (resolve, reject) => {
+        while (msgsSent <= 2) {
+          await producer
+            .sendMessage({ payload: 'sinai', properties: { k: 'v' } })
+            .then(() => {
+              msgsSent++;
+            })
+            .catch(reject);
+          if (msgsSent >= 2) {
+            resolve();
+            break;
+          }
+          producer._client.getCnx().close();
+        }
+      });
+    });
+    it('should resend messages after unload', async function () {
+      await producer.create();
+      let msgsSent = 0;
+      await new Promise(async (resolve, reject) => {
+        while (msgsSent <= 5) {
+          await producer
+            .sendMessage({ payload: 'sinai', properties: { k: 'v' } })
+            .then(() => {
+              msgsSent++;
+            })
+            .catch(reject);
+          if (msgsSent === 3) {
+            await utils.unloadTopic();
+          }
+          if (msgsSent >= 5) {
+            resolve();
+            break;
+          }
+        }
+      });
+    });
   });
   describe('on connection exception should resend batch', function () {
     it('should not throw exception', async function () {