Merge pull request #1 from taina0407/develop

v1.1.3
taina0407 · May 1, 2021 · ce3582f · ce3582f
2 parents 7764b27 + c61aec0
commit ce3582f
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 7 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+## v1.1.3
+- [Fix]
+  - Fix bug graceful shutdown not cancel consumer
+
 ## v1.1.2
 - [Fix]
   - Fix bug using merge-deep instead of defaults-deep

diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "moleculer-rabbitmq",
-  "version": "1.1.2",
+  "version": "1.1.3",
   "description": "Moleculer RabbitMQ queue plugin",
   "main": "index.js",
   "scripts": {

diff --git a/src/index.js b/src/index.js
@@ -1,4 +1,3 @@
-const Crypto = require("crypto");
 const Amqplib = require("amqplib");
 const DefaultDeep = require("defaults-deep");
 const Deep = require("deep-get-set");
@@ -59,9 +58,10 @@ const ACTION_OPTIONS_VALIDATOR = {
 const gracefulShutdown = async function () {
   isShuttingDown = true;
 
-  Object.keys(this.$amqpQueues).forEach((queueName) => {
-    const { channel, options } = this.$amqpQueues[queueName] || {};
-    channel && channel.cancel(Deep(options, "amqp.consume.consumerTag"));
+  Object.keys(this.$amqpQueues).forEach(async (queueName) => {
+    const { channel, consumerTag } = this.$amqpQueues[queueName] || {};
+
+    channel && consumerTag && await channel.cancel(consumerTag);
   });
 };
 
@@ -80,7 +80,6 @@ const initAMQPQueues = function (schema) {
       const queueName = `amqp.${schema.version ? `v${schema.version}.` : ""}${schema.name}.${originActionName}`;
 
       const queueOption = DefaultDeep({}, schema.actions[originActionName].queue, DEFAULT_QUEUE_OPTIONS);
-      Deep(queueOption, "amqp.consume.consumerTag", Crypto.randomBytes(16).toString("hex"));
 
       this.$amqpQueues[queueName] = {
         options: queueOption,
@@ -267,7 +266,10 @@ module.exports = (options) => ({
         } = queue;
 
         const amqpChannel = await this.assertAMQPQueue(queueName);
-        amqpChannel.consume(queueName, consumeHandler.bind(this, amqpChannel), consumeOptions);
+        const {
+          consumerTag,
+        } = await amqpChannel.consume(queueName, consumeHandler.bind(this, amqpChannel), consumeOptions);
+        this.$amqpQueues[queueName].consumerTag = consumerTag;
       });
     },
   },