Repository: luin/ioredis
Pull request #292 (2016-04-30 04:26:02)

fix: reconnect when getting fatal error


Patch:
diff --git a/lib/redis.js b/lib/redis.js
index 99bfd1f..b810807 100644
--- a/lib/redis.js
+++ b/lib/redis.js
@@ -276,6 +276,12 @@ Redis.prototype.connect = function (callback) {
         stream.setTimeout(_this.options.connectTimeout, function () {
           stream.setTimeout(0);
           stream.destroy();
+
+          var err = new Error('connect ETIMEDOUT');
+          err.errorno = 'ETIMEDOUT';
+          err.code = 'ETIMEDOUT';
+          err.syscall = 'connect';
+          eventHandler.errorHandler(_this)(err);
         });
         stream.once(CONNECT_EVENT, function () {
           stream.setTimeout(0);
@@ -347,18 +353,28 @@ Redis.prototype.duplicate = function (override) {
  * Flush offline queue and command queue with error.
  *
  * @param {Error} error - The error object to send to the commands
+ * @param {object} options
  * @private
  */
-Redis.prototype.flushQueue = function (error) {
+Redis.prototype.flushQueue = function (error, options) {
+  options = _.defaults({}, options, {
+    offlineQueue: true,
+    commandQueue: true
+  });
+
   var item;
-  while (this.offlineQueue.length > 0) {
-    item = this.offlineQueue.shift();
-    item.command.reject(error);
+  if (options.offlineQueue) {
+    while (this.offlineQueue.length > 0) {
+      item = this.offlineQueue.shift();
+      item.command.reject(error);
+    }
   }
 
-  while (this.commandQueue.length > 0) {
-    item = this.commandQueue.shift();
-    item.command.reject(error);
+  if (options.commandQueue) {
+    while (this.commandQueue.length > 0) {
+      item = this.commandQueue.shift();
+      item.command.reject(error);
+    }
   }
 };
 
diff --git a/lib/redis/parser.js b/lib/redis/parser.js
index 39c3ec6..6137cc4 100644
--- a/lib/redis/parser.js
+++ b/lib/redis/parser.js
@@ -28,7 +28,9 @@ exports.initParser = function () {
       _this.returnReply(reply);
     },
     returnFatalError: function (err) {
-      _this.emit('error', new Error('Redis reply parser error: ' + err.stack));
+      _this.flushQueue(err, { offlineQueue: false });
+      _this.silentEmit('error', new Error('Redis parser fatal error: ' + err.stack));
+      _this.disconnect(true);
     }
   });
 };


Test patch:
diff --git a/test/functional/connection.js b/test/functional/connection.js
index 3d43270..2c05fa4 100644
--- a/test/functional/connection.js
+++ b/test/functional/connection.js
@@ -48,9 +48,18 @@ describe('connection', function () {
       connectTimeout: 1,
       retryStrategy: null
     });
+    var pending = 2;
+    redis.on('error', function (err) {
+      expect(err.message).to.eql('connect ETIMEDOUT');
+      if (!--pending) {
+        done();
+      }
+    });
     redis.get('foo', function (err) {
       expect(err.message).to.match(/Connection is closed/);
-      done();
+      if (!--pending) {
+        done();
+      }
     });
   });
 
diff --git a/test/functional/fatal_error.js b/test/functional/fatal_error.js
new file mode 100644
index 0000000..b2b3671
--- /dev/null
+++ b/test/functional/fatal_error.js
@@ -0,0 +1,21 @@
+'use strict';
+
+describe('fatal_error', function () {
+  it('should handle fatal error of parser', function (done) {
+    var redis = new Redis();
+    redis.once('ready', function () {
+      var execute = redis.replyParser.execute;
+      redis.replyParser.execute = function () {
+        execute.call(redis.replyParser, '&');
+      };
+      redis.get('foo', function (err) {
+        expect(err.message).to.match(/Protocol error/);
+        redis.replyParser.execute = execute;
+        redis.get('bar', function (err) {
+          expect(err).to.eql(null);
+          done();
+        });
+      });
+    });
+  });
+});
diff --git a/test/unit/redis.js b/test/unit/redis.js
index 35f2427..ed3dfe8 100644
--- a/test/unit/redis.js
+++ b/test/unit/redis.js
@@ -84,7 +84,7 @@ describe('Redis', function () {
       }
       Redis.prototype.connect.restore();
 
-      function getOption () {
+      function getOption() {
         var redis = Redis.apply(null, arguments);
         return redis.options;
       }
@@ -115,4 +115,36 @@ describe('Redis', function () {
       redis.end();
     });
   });
+
+  describe('#flushQueue', function () {
+    it('should flush all queues by default', function () {
+      var flushQueue = Redis.prototype.flushQueue;
+      var redis = {
+        offlineQueue: [{ command: { reject: function () {} } }],
+        commandQueue: [{ command: { reject: function () {} } }]
+      };
+      var offline = mock(redis.offlineQueue[0].command);
+      var command = mock(redis.commandQueue[0].command);
+      offline.expects('reject').once();
+      command.expects('reject').once();
+      flushQueue.call(redis);
+      offline.verify();
+      command.verify();
+    });
+
+    it('should be able to ignore a queue', function () {
+      var flushQueue = Redis.prototype.flushQueue;
+      var redis = {
+        offlineQueue: [{ command: { reject: function () {} } }],
+        commandQueue: [{ command: { reject: function () {} } }]
+      };
+      var offline = mock(redis.offlineQueue[0].command);
+      var command = mock(redis.commandQueue[0].command);
+      offline.expects('reject').once();
+      command.expects('reject').never();
+      flushQueue.call(redis, new Error(), { commandQueue: false });
+      offline.verify();
+      command.verify();
+    });
+  });
 });