FKDEV-363: mq reconnect logic

src/ConnectionPool.js

@@ -101,6 +101,17 @@ class ConnectionPool {
       await connection.connect()
     }
   }
+
+  /**
+   * @return {Promise}
+   */
+  async reconnect () {
+    const connections = [...this.connections.values()]
+
+    for (const connection of connections) {
+      await connection.reconnect()
+    }
+  }
 }
 
 module.exports = ConnectionPool

src/QueueConnection.js

@@ -29,27 +29,6 @@ class QueueConnection {
     this._logger = logger
   }
 
-  close () {
-    if (this._connection) {
-      this._connection.close((err) => {
-        this._logger.error('RabbitMQ close connection failed', err)
-        throw err
-      })
-    }
-  }
-
-  reconnect () {
-    if (this._connection) {
-      this._connection.off('close', this._onClose)
-      this.close()
-    }
-
-    this._connection = null
-    this._connectionPromise = null
-
-    return this.connect()
-  }
-
   /**
    * @return Promise<amqplib.Connection>
    * */
@@ -139,6 +118,25 @@ class QueueConnection {
     }
   }
 
+  /**
+   * @return Promise
+   * */
+  close (invokeCloseEvent = false) {
+    if (this._connection) {
+      if (!invokeCloseEvent) {
+        this._connection.off('close', this._onClose)
+      }
+
+      this._connection.close((err) => {
+        this._logger.error('RabbitMQ close connection failed', err)
+        throw err
+      })
+    }
+
+    this._connection = null
+    this._connectionPromise = null
+  }
+
   onConnection (event, callback) {
     if (this._connection) {
       this._connection.on(event, callback)

src/QueueManager.js

@@ -90,6 +90,16 @@ class QueueManager {
     }
   }
 
+  async reconnect () {
+    try {
+      await this.connection.close()
+      await this.connect()
+    } catch (err) {
+      this._logger.error('Failed to reconnect to queue server', err)
+      throw err
+    }
+  }
+
   setLogger (logger) {
     this._logger = logger
     this.connection.setLogger(logger)