Do not emit the broken mode twice if exec is called

Add more tests

index.js

@@ -137,10 +137,8 @@ RedisClient.prototype.unref = function () {
 };
 
 // flush offline_queue and command_queue, erroring any items with a callback first
-RedisClient.prototype.flush_and_error = function (message) {
-    var command_obj, error;
-
-    error = new Error(message);
+RedisClient.prototype.flush_and_error = function (error) {
+    var command_obj;
 
     while (command_obj = this.offline_queue.shift()) {
         if (typeof command_obj.callback === "function") {
@@ -438,17 +436,19 @@ RedisClient.prototype.connection_gone = function (why) {
     // If this is a requested shutdown, then don't retry
     if (this.closing) {
         debug("connection ended from quit command, not retrying.");
-        this.flush_and_error("Redis connection gone from " + why + " event.");
+        this.flush_and_error(new Error("Redis connection gone from " + why + " event."));
         return;
     }
 
     if (this.max_attempts !== 0 && this.attempts >= this.max_attempts || this.retry_totaltime >= this.connect_timeout) {
-        this.flush_and_error("Redis connection gone from " + why + " event.");
-        this.end();
         var message = this.retry_totaltime >= this.connect_timeout ?
             'connection timeout exceeded.' :
             'maximum connection attempts exceeded.';
-        this.emit('error', new Error("Redis connection in broken state: " + message));
+        var error = new Error("Redis connection in broken state: " + message);
+        error.code = 'CONNECTION_BROKEN';
+        this.flush_and_error(error);
+        this.emit('error', error);
+        this.end();
         return;
     }
 
@@ -1035,7 +1035,7 @@ Multi.prototype.exec = Multi.prototype.EXEC = function (callback) {
 
     // TODO - make this callback part of Multi.prototype instead of creating it each time
     return this._client.send_command("exec", [], function (err, replies) {
-        if (err) {
+        if (err && !err.code) {
             if (callback) {
                 errors.push(err);
                 callback(errors);
@@ -1071,6 +1071,9 @@ Multi.prototype.exec = Multi.prototype.EXEC = function (callback) {
 
         if (callback) {
             callback(null, replies);
+        } else if (err && err.code !== 'CONNECTION_BROKEN') {
+            // Exclude CONNECTION_BROKEN so that error won't be emitted twice
+            self._client.emit('error', err);
         }
     });
 };

test/connection.spec.js

@@ -19,9 +19,7 @@ describe("on lost connection", function () {
                 var calls = 0;
 
                 client.once('ready', function() {
-                    // Pretend that redis can't reconnect
-                    client.on_connect = client.on_error;
-                    client.stream.destroy();
+                    helper.killConnection(client);
                 });
 
                 client.on("reconnecting", function (params) {
@@ -40,16 +38,14 @@ describe("on lost connection", function () {
 
             it("emit an error after max retry timeout and do not try to reconnect afterwards", function (done) {
                 var connect_timeout = 1000; // in ms
-                client = redis.createClient({
+                var client = redis.createClient({
                     parser: parser,
                     connect_timeout: connect_timeout
                 });
                 var time = 0;
 
                 client.once('ready', function() {
-                    // Pretend that redis can't reconnect
-                    client.on_connect = client.on_error;
-                    client.stream.destroy();
+                    helper.killConnection(client);
                 });
 
                 client.on("reconnecting", function (params) {
@@ -66,6 +62,23 @@ describe("on lost connection", function () {
                 });
             });
 
+            it("end connection while retry is still ongoing", function (done) {
+                var connect_timeout = 1000; // in ms
+                var client = redis.createClient({
+                    parser: parser,
+                    connect_timeout: connect_timeout
+                });
+
+                client.once('ready', function() {
+                    helper.killConnection(client);
+                });
+
+                client.on("reconnecting", function (params) {
+                    client.end();
+                    setTimeout(done, 100);
+                });
+            });
+
         });
     });
 });

test/helper.js

@@ -145,5 +145,15 @@ module.exports = {
                 func();
             }
         };
+    },
+    killConnection: function (client) {
+        // Change the connection option to a non existing one and destroy the stream
+        client.connectionOption = {
+            port: 6370,
+            host: '127.0.0.2',
+            family: 4
+        };
+        client.address = '127.0.0.2:6370';
+        client.stream.destroy();
     }
 };

test/node_redis.spec.js

@@ -729,9 +729,11 @@ describe("The node_redis client", function () {
                         client.on('reconnecting', function(params) {
                             i++;
                             assert.equal(params.attempt, i);
-                            assert.strictEqual(client.offline_queue.length, 1);
+                            assert.strictEqual(client.offline_queue.length, 2);
                         });
 
+                        // Should work with either a callback or without
+                        client.set('baz', 13);
                         client.set('foo', 'bar', function(err, result) {
                             assert(i, 3);
                             assert('Redis connection gone from error event', err.message);
@@ -764,6 +766,40 @@ describe("The node_redis client", function () {
                             });
                         });
                     });
+
+                    it("flushes the command queue connection if in broken connection mode", function (done) {
+                        var client = redis.createClient({
+                            parser: parser,
+                            max_attempts: 2,
+                            enable_offline_queue: false
+                        });
+
+                        client.once('ready', function() {
+                            var multi = client.multi();
+                            multi.config("bar");
+                            var cb = function(err, reply) {
+                                assert.equal(err.code, 'CONNECTION_BROKEN');
+                            };
+                            for (var i = 0; i < 10; i += 2) {
+                                multi.set("foo" + i, "bar" + i);
+                                multi.set("foo" + (i + 1), "bar" + (i + 1), cb);
+                            }
+                            multi.exec();
+                            assert.equal(client.command_queue.length, 13);
+                            helper.killConnection(client);
+                        });
+
+                        client.on("reconnecting", function (params) {
+                            assert.equal(client.command_queue.length, 13);
+                        });
+
+                        client.on('error', function(err) {
+                            if (/Redis connection in broken state:/.test(err.message)) {
+                                assert.equal(client.command_queue.length, 0);
+                                done();
+                            }
+                        });
+                    });
                 });
             });