fix: reconnect when getting fatal error (#292)

* fix: reconnect when getting fatal error

* test: add test for Redis#flushQueue

* test: add test for fatal error

Author: luin

lib/redis.js

@@ -353,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);
+    }
   }
 };
 

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/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();
+        });
+      });
+    });
+  });
+});

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();
+    });
+  });
 });