fix: remove graceful timeout (#22)

Author: rizkifuad

index.js

@@ -28,6 +28,11 @@ module.exports = function () {
             let serviceSubQueueRabbit = yield container.resolve("serviceSubQueueRabbit");
             service.addModule("pub-queue-rabbit", servicePubQueueRabbit);
             service.addModule("sub-queue-rabbit", serviceSubQueueRabbit);
+        },
+     
+        *onStop(container) {
+            let service = yield container.resolve("amqp");
+            yield service.stop();
         }
     };
-};
+};

lib/service_amqp.js

@@ -15,7 +15,6 @@ module.exports = class Amqp extends Component.mixin(AsyncEmitter) {
     this.serviceSub = null;
     this.serviceSubQueue = null;
     this.signals = ["SIGINT", "SIGTERM", "SIGQUIT"];
-    this.isShuttingDown = false;
   }
 
   initialize() {
@@ -28,10 +27,6 @@ module.exports = class Amqp extends Component.mixin(AsyncEmitter) {
 
   *doConnect() {
     if (this._initializing) return;
-
-    // we don't need to reconnect if there's any shutting down process
-    if (this.isShuttingDown) return;
-
     this._initializing = true;
     let {
       secure,
@@ -68,7 +63,6 @@ module.exports = class Amqp extends Component.mixin(AsyncEmitter) {
 
       this._connection.on("close", this.doConnect.bind(this));
       this._connection.on("error", this.doConnect.bind(this));
-      this.handleShutdown();
 
       this.emit("connected");
 
@@ -86,33 +80,13 @@ module.exports = class Amqp extends Component.mixin(AsyncEmitter) {
     }
   }
 
-  handleShutdown() {
-    this.signals.forEach((signal) =>
-      process.addListener(signal, () => {
-        console.log("Received signal", signal);
-        this.cleanup(signal);
-
-        console.log("Exiting process in 3sec...");
-        setTimeout(() => {
-          process.exit(0);
-        }, 3000);
-      })
-    );
-  }
-
-  cleanup(signal) {
-    if (!this.isShuttingDown) {
-      try {
-        console.log("Shutting down rabbitmq plugin...");
-
-        this.isShuttingDown = true;
-
-        this._connection.close();
-      } catch (e) {
-        console.error("Error while shutting down gracefully", e);
-
-        process.exit(1);
-      }
+  *stop() {
+    try {
+      this.logger.info("Shutting down rabbitmq plugin...");
+      this._connection.close();
+    } catch (e) {
+      this.logger.info("Error while shutting down gracefully", e);
+      process.exit(1);
     }
   }
 };

lib/service_sub_queue_rabbit.js

@@ -64,6 +64,7 @@ class ServiceSubRabbit extends ServiceSubQueue {
 
             yield channel.assertQueue(queueName, { durable: true });
 
+            const consumerTag = process.env.HOSTNAME || "";
             channel.consume(queueName, async(function* (message) {
                 try {
                     let payload = JSON.parse(message.content.toString());
@@ -74,7 +75,7 @@ class ServiceSubRabbit extends ServiceSubQueue {
                 catch (e) {
                     channel.nack(message);
                 }
-            }.bind(this)));
+            }.bind(this)), { consumerTag });
 
         }
     }
@@ -95,4 +96,4 @@ class ServiceSubRabbit extends ServiceSubQueue {
     *unmount() { }
 }
 
-module.exports = ServiceSubRabbit;
+module.exports = ServiceSubRabbit;

lib/service_sub_rabbit.js

@@ -107,6 +107,7 @@ class ServiceSubRabbit extends ServiceSub {
 
         yield channel.assertQueue(queueName, { durable: true });
 
+        const consumerTag = process.env.HOSTNAME || "";
         channel.consume(queueName, async(function* (message) {
             try {
                 let payload = JSON.parse(message.content.toString());
@@ -120,7 +121,7 @@ class ServiceSubRabbit extends ServiceSub {
             catch (e) {
                 channel.nack(message);
             }
-        }));
+        }), { consumerTag });
     }
 
     *createQueue(publisherName, event, method) {
@@ -135,6 +136,7 @@ class ServiceSubRabbit extends ServiceSub {
             yield channel.bindQueue(queueName, exchangeName, "");
             this._queues.push(queueName);
 
+            const consumerTag = process.env.HOSTNAME || "";
             channel.consume(queueName, async(function* (message) {
                 try {
                     let payload = JSON.parse(message.content.toString());
@@ -145,7 +147,7 @@ class ServiceSubRabbit extends ServiceSub {
                 catch (e) {
                     channel.nack(message);
                 }
-            }.bind(this)));
+            }.bind(this)), { consumerTag });
 
         }
     }