Merge pull request #30 from kata-ai/fix/reconnect-rabbitmq-channel

Feat: Reconnect Channel

Author: BagusAK95

lib/service_sub_queue_rabbit.js

@@ -3,6 +3,7 @@
 const { async, utils } = require("merapi");
 const ServiceSubQueue = require("merapi-plugin-service/lib/service_sub_queue");
 const pack = require("../package");
+const sleep = require("then-sleep");
 
 class ServiceSubRabbit extends ServiceSubQueue {
 
@@ -60,6 +61,11 @@ class ServiceSubRabbit extends ServiceSubQueue {
         let channel = this._channels[event];
 
         if (channel) {
+            channel.on("close", () => {
+                console.dir(`${event} is closed`)
+                this.reconnect(event)
+            });
+            
             let queueName = `${this._namespace}.queue.${this.SERVICE_NAME}.${event}`;
 
             yield channel.assertQueue(queueName, { durable: true });
@@ -80,6 +86,25 @@ class ServiceSubRabbit extends ServiceSubQueue {
         }
     }
 
+    *reconnect(event) {
+        try {
+            let desc = this.config.default("service.queue.subscribe", {});
+            let prefetch = parseInt(this.config.default("service.rabbit.prefetch", 5));
+
+            let channel = yield this._connection.createChannel();
+            yield channel.prefetch(prefetch);
+            this._channels[event] = channel;
+    
+            let method = yield this.injector.resolveMethod(desc[event]);
+            this.createQueue(event, method);    
+        } catch (e) {
+            this.logger.warn("Failed to connect channel " + event, e);
+            yield sleep(3000);
+
+            yield this.reconnect(event);
+        }
+    }
+
     *runMethod(method, payload) {
         let ret = method(payload);
         if (utils.isPromise(ret)) yield ret;

package-lock.json

@@ -1,6 +1,6 @@
 {
     "name": "merapi-plugin-service-rabbit",
-    "version": "0.6.5",
+    "version": "0.6.6",
     "description": "Provide RabbitMQ integration interface",
     "main": "index.js",
     "scripts": {