feat(Consumer): add method msg.checkTimeout(callback)

Author: leizongmin

README.md

@@ -96,6 +96,7 @@ c.listen(msg => {
   // msg.expire = 消息过期秒时间戳
   // msg.reject(err) 消息处理出错
   // msg.resolve(result) 消息处理成功
+  // msg.checkTimeout(callback) 检查执行是否超时，如果在expire之后的时间还没有响应，则自动响应一个MessageProcessingTimeoutError，并执行回调函数
 });
 
 // 退出

lib/consumer.js

@@ -156,17 +156,33 @@ class Consumer extends EventEmitter {
         this.expire = expire;
         this.data = data;
         this.originData = originData;
+        this._checkTimeoutTid = null;
+        this._isDone = false;
       }
 
       reject(err) {
-        reply(this.producerName, this.msgId, 'e', (err && err.message) || err.toString(), this.originData);
+        if (this._isDone) return;
+        clearTimeout(this._checkTimeoutTid);
+        this._isDone = true;
+        reply(this.producerName, this.msgId, 'e', ((err && err.message) || err).toString(), this.originData);
       }
 
       resolve(data) {
+        if (this._isDone) return;
+        clearTimeout(this._checkTimeoutTid);
+        this._isDone = true;
         if (typeof data !== 'string') throw new Error('`data` must be string');
         reply(this.producerName, this.msgId, 's', data, this.originData);
       }
 
+      checkTimeout(callback) {
+        if (this._isDone) return;
+        setTimeout(() => {
+          this.reject(new utils.MessageProcessingTimeoutError());
+          callback && callback(null);
+        }, Date.now() - this.expire * 1000);
+      }
+
     }
 
     pull();

lib/utils.js

@@ -109,6 +109,15 @@ function MessageExpiredError(message) {
 MessageExpiredError.prototype = Error.prototype;
 utils.MessageExpiredError = MessageExpiredError;
 
+function MessageProcessingTimeoutError(message) {
+  Error.captureStackTrace(this, MessageProcessingTimeoutError);
+  this.name = 'MessageProcessingTimeoutError';
+  this.message = (message || '');
+  this.code = 'msg_timeout';
+}
+MessageProcessingTimeoutError.prototype = Error.prototype;
+utils.MessageProcessingTimeoutError = MessageProcessingTimeoutError;
+
 utils.classMethodWrapPromise = function (obj, methods) {
   for (const method of methods) {
     obj.prototype['_' + method + 'Callback'] = obj.prototype[method];

test/test_normal.js

@@ -17,7 +17,7 @@ function generateOptions(options) {
   options.queue = utils.date('Ymd_His_') + utils.randomString(10);
   options.redis = {
     host: '127.0.0.1',
-    port: 6378,
+    port: 6379,
     db: 15,
     prefix: 'TEST:',
   };