feat(Monitor): 当consumer下线时，自动将其未完成的消息转移回主队列

Author: leizongmin

lib/monitor.js

@@ -89,10 +89,10 @@ class Monitor extends EventEmitter {
     this._previousConsumers = [];
     this._autoCheckTid = setInterval(() => this.check(), this._interval * 1000);
     this._emitCounter = 0;
-
-    this.on('consumerDown', name => {
-      this.withdrawnProcessingQueue(name);
+    this.on('consumerDown', info => {
+      this.withdrawnProcessingQueue(info);
     });
+    this.check();
 
     this._debug = utils.debug('monitor:' + this.name);
     this._debug('created: interval=%s, redis=%j', this._interval, options.redis);
@@ -131,7 +131,7 @@ class Monitor extends EventEmitter {
       }
 
       const emit = (event, list) => {
-        //if (this._emitCounter < 1) return this._debug('fist time emit, skip');
+        if (this._emitCounter < 1) return this._debug('fist time emit, skip');
         for (const item of list) {
           this.emit(event, item);
         }
@@ -255,11 +255,21 @@ class Monitor extends EventEmitter {
 
   }
 
-  withdrawnProcessingQueue(queue) {
+  withdrawnProcessingQueue(info, callback) {
+
+    const processingKey = utils.getProcessingQueueKey(this._redisPrefix, info.queue, info.name);
+    const waitingKey = utils.getQueueKey(this._redisPrefix, info.queue);
+    this._debug('withdrawnProcessingQueue: %s -> %s', processingKey, waitingKey);
 
-    const key = utils.getProcessingQueueKey(this._redisPrefix, queue);
+    callback = callback || () => {};
+    const next = (err, ret) => {
+      if (err) return callback(err);
+      if (!ret) return callback(null);
+      if (ret !== true) this._debug('withdrawnProcessingQueue: %s -> %s %s', processingKey, waitingKey, ret);
+      this._redis.rpoplpush(processingKey, waitingKey, next);
+    };
+    next(null, true);
 
-    this._debug('withdrawnProcessingQueue');
   }
 
   /**