add different handling for on close and on error

Author: goFrendiAsgard

.gitignore

@@ -5,3 +5,4 @@ coverage
 .vscode
 .nyc_output
 test/test.env
+labs/

README.md

@@ -203,6 +203,9 @@ export RABBIT_HOST=0.0.0.0
 export RABBIT_PORT=5672
 export RABBIT_USERNAME=root
 export RABBIT_PASSWORD=toor
+# these commands are for reconnecting test
+export START_RABBIT_COMMAND="docker start rabbitmq"
+export STOP_RABBIT_COMMAND="docker stop rabbitmq"
 ```
 
 and perform:

lib/service_amqp.js

@@ -1,5 +1,6 @@
 "use strict";
 
+const amqpConnectionManager = require('amqp-connection-manager');
 const amqp = require("amqplib");
 const sleep = require("then-sleep");
 const { Component, AsyncEmitter } = require("merapi");
@@ -42,20 +43,32 @@ module.exports = class Amqp extends Component.mixin(AsyncEmitter) {
 
         try {
             this._connection = yield amqp.connect(connectionString);
+            // this._connection = yield amqpConnectionManager.connect([connectionString]);
 
             this.logger.info("Connected to rmq.");
 
-            this._connection.on("close", this.doConnect.bind(this));
-            this._connection.on("error", this.doConnect.bind(this));
+            this._connection.on("close", (e) => {
+                this.logger.error(e);
+                this.emit("disconnected");
+                this._connection.close();
+                this.doConnect.bind(this)();
+            });
+            this._connection.on("error", (e) => {
+                this.logger.error(e);
+                this.emit("disconnected");
+                this._connection = null;
+                this.doConnect.bind(this)();
+            });
 
             this.emit("connected");
 
             this._initializing = false;
         } catch (e) {
+            this.logger.error(e);
             this._initializing = false;
 
             this.logger.warn("Failed to connect to rmq.", e);
-            yield sleep(3000);
+            yield sleep(30000);
             this.logger.info("Attempting to reconnect to rmq.");
 
             yield this.doConnect();

lib/service_sub_queue_rabbit.js

@@ -94,4 +94,4 @@ class ServiceSubRabbit extends ServiceSubQueue {
     *unmount() { }
 }
 
-module.exports = ServiceSubRabbit;
+module.exports = ServiceSubRabbit;

package-lock.json

@@ -26,7 +26,8 @@
     "license": "ISC",
     "homepage": "https://github.com/kata-ai/merapi-plugin-service-rabbit#readme",
     "dependencies": {
-        "amqplib": "^0.4.2",
+        "amqp-connection-manager": "^2.3.0",
+        "amqplib": "^0.5.3",
         "merapi": "^0.16.0",
         "merapi-plugin-service": "^0.3.5",
         "requestretry": "^1.12.0",

package.json

@@ -1,3 +1,5 @@
+const { exec } = require("child_process");
+
 const rabbitConnection = {
     host: process.env.RABBIT_HOST || "localhost",
     port: process.env.RABBIT_PORT || 5672,
@@ -9,8 +11,12 @@ const rabbitConnection = {
 };
 
 const rabbitUrl = `amqp://${rabbitConnection.user}:${rabbitConnection.password}@${rabbitConnection.host}:${rabbitConnection.port}`;
+const startRabbitCommand = process.env.START_RABBIT_COMMAND;
+const stopRabbitCommand = process.env.STOP_RABBIT_COMMAND;
 
 module.exports = {
     rabbitConnection,
     rabbitUrl,
+    startRabbitCommand,
+    stopRabbitCommand,
 };