feat(cluster): add lazy connect support (#352)

* add lazyConnect to Redis.Cluster * use _.noop * fix: handle disconnection with lazyConnect * remove unnecessary setImmediate * add comments to setImmediate
redis · Jul 28, 2016 · f1cadff · f1cadff
1 parent 9d70cc7
commit f1cadff
Show file tree

Hide file tree

Showing 5 changed files with 110 additions and 29 deletions.
diff --git a/lib/cluster/connection_pool.js b/lib/cluster/connection_pool.js
@@ -45,7 +45,7 @@ ConnectionPool.prototype.findOrCreate = function (node, readOnly) {
     redis = this.nodes.all[node.key];
     if (redis.options.readOnly !== readOnly) {
       redis.options.readOnly = readOnly;
-      redis[readOnly ? 'readonly' : 'readwrite']().catch(function () {});
+      redis[readOnly ? 'readonly' : 'readwrite']().catch(_.noop);
       if (readOnly) {
         delete this.nodes.master[node.key];
         this.nodes.slave[node.key] = redis;

diff --git a/lib/cluster/index.js b/lib/cluster/index.js
@@ -80,7 +80,11 @@ function Cluster(startupNodes, options) {
 
   this.subscriber = null;
 
-  this.connect().catch(noop);
+  if (this.options.lazyConnect) {
+    this.setStatus('wait');
+  } else {
+    this.connect().catch(_.noop);
+  }
 }
 
 /**
@@ -161,24 +165,7 @@ Cluster.prototype.connect = function () {
 
     this.once('refresh', refreshListener);
     this.once('close', closeListener);
-
-    this.once('close', function () {
-      var retryDelay;
-      if (!this.manuallyClosing && typeof this.options.clusterRetryStrategy === 'function') {
-        retryDelay = this.options.clusterRetryStrategy.call(this, ++this.retryAttempts);
-      }
-      if (typeof retryDelay === 'number') {
-        this.setStatus('reconnecting');
-        this.reconnectTimeout = setTimeout(function () {
-          this.reconnectTimeout = null;
-          debug('Cluster is disconnected. Retrying after %dms', retryDelay);
-          this.connect().catch(noop);
-        }.bind(this), retryDelay);
-      } else {
-        this.setStatus('end');
-        this.flushQueue(new Error('None of startup nodes is available'));
-      }
-    });
+    this.once('close', this._handleCloseEvent.bind(this));
 
     this.refreshSlotsCache(function (err) {
       if (err && err.message === 'Failed to refresh slots cache.') {
@@ -190,12 +177,34 @@ Cluster.prototype.connect = function () {
   }.bind(this));
 };
 
+/**
+ * Called when closed to check whether a reconnection should be made
+ */
+Cluster.prototype._handleCloseEvent = function () {
+  var retryDelay;
+  if (!this.manuallyClosing && typeof this.options.clusterRetryStrategy === 'function') {
+    retryDelay = this.options.clusterRetryStrategy.call(this, ++this.retryAttempts);
+  }
+  if (typeof retryDelay === 'number') {
+    this.setStatus('reconnecting');
+    this.reconnectTimeout = setTimeout(function () {
+      this.reconnectTimeout = null;
+      debug('Cluster is disconnected. Retrying after %dms', retryDelay);
+      this.connect().catch(_.noop);
+    }.bind(this), retryDelay);
+  } else {
+    this.setStatus('end');
+    this.flushQueue(new Error('None of startup nodes is available'));
+  }
+};
+
 /**
  * Disconnect from every node in the cluster.
  *
  * @public
  */
 Cluster.prototype.disconnect = function (reconnect) {
+  var status = this.status;
   this.setStatus('disconnecting');
 
   if (!reconnect) {
@@ -205,7 +214,13 @@ Cluster.prototype.disconnect = function (reconnect) {
     clearTimeout(this.reconnectTimeout);
     this.reconnectTimeout = null;
   }
-  this.connectionPool.reset([]);
+
+  if (status === 'wait') {
+    this.setStatus('close');
+    this._handleCloseEvent();
+  } else {
+    this.connectionPool.reset([]);
+  }
 };
 
 /**
@@ -216,6 +231,7 @@ Cluster.prototype.disconnect = function (reconnect) {
  * @public
  */
 Cluster.prototype.quit = function (callback) {
+  var status = this.status;
   this.setStatus('disconnecting');
 
   this.manuallyClosing = true;
@@ -224,6 +240,18 @@ Cluster.prototype.quit = function (callback) {
     clearTimeout(this.reconnectTimeout);
     this.reconnectTimeout = null;
   }
+  if (status === 'wait') {
+    var ret = Promise.resolve('OK').nodeify(callback);
+
+    // use setImmediate to make sure "close" event
+    // being emitted after quit() is returned
+    setImmediate(function () {
+      this.setStatus('close');
+      this._handleCloseEvent();
+    }.bind(this));
+
+    return ret;
+  }
   return Promise.all(this.nodes().map(function (node) {
     return node.quit();
   })).then(function () {
@@ -277,12 +305,12 @@ Cluster.prototype.selectSubscriber = function () {
           if (!--pending) {
             _this.lastActiveSubscriber = _this.subscriber;
           }
-        }).catch(noop);
+        }).catch(_.noop);
       }
     });
   } else {
     if (this.subscriber.status === 'wait') {
-      this.subscriber.connect().catch(noop);
+      this.subscriber.connect().catch(_.noop);
     }
     this.lastActiveSubscriber = this.subscriber;
   }
@@ -389,6 +417,9 @@ Cluster.prototype.executeOfflineCommands = function () {
 };
 
 Cluster.prototype.sendCommand = function (command, stream, node) {
+  if (this.status === 'wait') {
+    this.connect().catch(_.noop);
+  }
   if (this.status === 'end') {
     command.reject(new Error(utils.CONNECTION_CLOSED_ERROR_MSG));
     return command.promise;
@@ -627,6 +658,4 @@ Cluster.prototype._readyCheck = function (callback) {
 
 require('../transaction').addTransactionSupport(Cluster.prototype);
 
-function noop() {}
-
 module.exports = Cluster;
diff --git a/lib/redis.js b/lib/redis.js
@@ -127,7 +127,7 @@ function Redis() {
   if (this.options.lazyConnect) {
     this.setStatus('wait');
   } else {
-    this.connect().catch(function () {});
+    this.connect().catch(_.noop);
   }
 }
 
@@ -544,7 +544,7 @@ require('./transaction').addTransactionSupport(Redis.prototype);
  */
 Redis.prototype.sendCommand = function (command, stream) {
   if (this.status === 'wait') {
-    this.connect().catch(function () {});
+    this.connect().catch(_.noop);
   }
   if (this.status === 'end') {
     command.reject(new Error(utils.CONNECTION_CLOSED_ERROR_MSG));

diff --git a/lib/redis/event_handler.js b/lib/redis/event_handler.js
@@ -3,6 +3,7 @@
 var debug = require('debug')('ioredis:connection');
 var Command = require('../command');
 var utils = require('../utils');
+var _ = require('lodash');
 
 exports.connectHandler = function (self) {
   return function () {
@@ -91,7 +92,7 @@ exports.closeHandler = function (self) {
     self.setStatus('reconnecting', retryDelay);
     self.reconnectTimeout = setTimeout(function () {
       self.reconnectTimeout = null;
-      self.connect().catch(function () {});
+      self.connect().catch(_.noop);
     }, retryDelay);
   };
 
@@ -145,7 +146,7 @@ exports.readyHandler = function (self) {
 
     if (self.options.readOnly) {
       debug('set the connection to readonly mode');
-      self.readonly().catch(function () {});
+      self.readonly().catch(_.noop);
     }
 
     if (self.prevCondition) {

diff --git a/test/functional/lazy_connect.js b/test/functional/lazy_connect.js
@@ -34,4 +34,55 @@ describe('lazy connect', function () {
     });
     redis.disconnect();
   });
+
+  describe('Cluster', function () {
+    it('should not call `connect` when init', function () {
+      stub(Redis.Cluster.prototype, 'connect').throws(new Error('`connect` should not be called'));
+      new Redis.Cluster([], { lazyConnect: true });
+      Redis.Cluster.prototype.connect.restore();
+    });
+
+    it('should quit before "close" being emited', function (done) {
+      stub(Redis.Cluster.prototype, 'connect').throws(new Error('`connect` should not be called'));
+      var cluster = new Redis.Cluster([], { lazyConnect: true });
+      cluster.quit(function () {
+        cluster.once('close', function () {
+          cluster.once('end', function () {
+            Redis.Cluster.prototype.connect.restore();
+            done();
+          });
+        });
+      });
+    });
+
+    it('should disconnect before "close" being emited', function (done) {
+      stub(Redis.Cluster.prototype, 'connect').throws(new Error('`connect` should not be called'));
+      var cluster = new Redis.Cluster([], { lazyConnect: true });
+      cluster.disconnect();
+      cluster.once('close', function () {
+        cluster.once('end', function () {
+          Redis.Cluster.prototype.connect.restore();
+          done();
+        });
+      });
+    });
+
+    it('should support disconnecting with reconnect', function (done) {
+      stub(Redis.Cluster.prototype, 'connect').throws(new Error('`connect` should not be called'));
+      var cluster = new Redis.Cluster([], {
+        lazyConnect: true,
+        clusterRetryStrategy: function () {
+          return 1;
+        }
+      });
+      cluster.disconnect(true);
+      cluster.once('close', function () {
+        Redis.Cluster.prototype.connect.restore();
+        stub(Redis.Cluster.prototype, 'connect', function () {
+          Redis.Cluster.prototype.connect.restore();
+          done();
+        });
+      });
+    });
+  });
 });