Circuit breaker

.gitignore

@@ -1,2 +1,5 @@
 *.swp
 /node_modules
+
+#webstorm
+.idea

ShardedRedisClient.js

@@ -254,7 +254,7 @@ class ShardedRedisClient extends EventEmitter {
     });
   }
 
-};
+}
 
 SHARDABLE.forEach((cmd) => {
 
@@ -280,59 +280,32 @@ SHARDABLE.forEach((cmd) => {
 
     mainCb = _.once(mainCb);
 
-    const isReadCmd = READ_ONLY.indexOf(cmd) !== -1;
-    const timeout = isReadCmd ? this._readTimeout : this._writeTimeout;
-
     const _this = this;
-    let timeoutHandler = null;
     let breaker = client._breaker;
 
-    const timeoutCb = args[args.length - 1] = function (err) {
-      if (timeoutHandler)
-        clearTimeout(timeoutHandler);
-
-      if (!err) {
-        if (breaker)
-          breaker.pass();
-
-        return mainCb.apply(this, arguments);
-      }
-
-      // For now, both emit and log. Eventually, logging will be removed
-      _this.emit('err', new Error(`sharded-redis-client [${client.address}] err: ${err}`));
-      console.error(new Date().toISOString(), `sharded-redis-client [${client.address }] err: ${err}`);
-
-      if (breaker && err.message !== 'breaker open')
-        breaker.fail();
-
-      if (!client._isMaster) {
-        client = wrappedClient.slaves.next(client);
+    execute();
 
-        if (client._rrindex == startIndex)
-          client = findMasterClient(shardKey, _this._wrappedClients);
+    function execute() {
+      breaker.exec(cmd, args)
+        .then(result => mainCb(null, result))
+        .catch((err) => {
+          _this.emit('err', new Error(`sharded-redis-client [${client.address}] err: ${err}`));
+          console.error(new Date().toISOString(), `sharded-redis-client [${client.address }] err: ${err}`);
 
-        breaker = client._breaker;
+          if(!client._isMaster) {
+            client = wrappedClient.slaves.next(client);
 
-        return wrappedCmd(client, args);
-      }
-
-      mainCb.apply(this, arguments);
-    };
-
-    wrappedCmd(client, args);
+            if (client._rrindex == startIndex)
+              client = findMasterClient(shardKey, _this._wrappedClients);
 
-    function wrappedCmd(client, args) {
-      if (!breaker || breaker.closed()) {
-        // Intentionally don't do this if timeout was set to 0
-        if (timeout)
-          timeoutHandler = setTimeout(timeoutCb, timeout, new Error('Redis call timed out'));
+            breaker = client._breaker;
+            return execute();
+          }
 
-        return client[cmd].apply(client, args);
-      }
-
-      timeoutCb(new Error('breaker open'));
+          mainCb(err);
+        });
     }
-  }
+  };
 
   ShardedRedisClient.prototype[cmd] = function () {
     const args = arguments;

lib/WrappedClient.js

@@ -19,8 +19,9 @@
 const _ = require('lodash');
 const redis = require('redis');
 const EventEmitter = require('events').EventEmitter;
-const CircuitBreaker = require('circuit-breaker');
 const RoundRobinSet = require('./RoundRobinSet');
+const Brakes = require('brakes');
+const Promise = require('bluebird');
 
 function createClient(port, host, options) {
   const client = redis.createClient(port, host, options.redisOptions);
@@ -30,8 +31,31 @@ function createClient(port, host, options) {
   if (options.usePing)
     setTimeout(() => setInterval(() => client.ping(_.noop), interval), timeout);
 
-  if (options.breakerConfig)
-    client._breaker = new CircuitBreaker(options.breakerConfig);
+  if (options.breakerConfig) {
+    options.breakerConfig = _.defaults(options.breakerConfig, {
+      statInterval: 2500,
+      threshold: 0.5,
+      circuitDuration: 15000,
+      timeout: 5000
+    });
+
+    const brake = new Brakes((cmd, args) => new Promise((resolve, reject) => {
+      args = _.cloneDeep(args);
+      args[length - 1] = (err, result) => {
+        if(err)
+          return reject(err);
+        return resolve(result);
+      };
+
+      return client[cmd].apply(client, args);
+    }), options.breakerConfig);
+
+    brake.on('snapshot', (snapshot) => console.log(`Snapshot ${snapshot} received`));
+    brake.on('circuitOpen', () => console.log(`Circuit Opened at ${new Date().toISOString()}`));
+    brake.on('circuitClosed', () => console.log(`Circuit Closed at ${new Date().toISOString()}`));
+
+    client._breaker = brake;
+  }
 
   client.on('error', (e) => console.log(`Redis Error [${host}:${port}]: ${e} : ${new Date().toISOString()}`));
   client.on('end', (e) => console.log(`Redis End [${host}:${port}]: ${e} : ${new Date().toISOString()}`));

package.json

@@ -21,6 +21,8 @@
   "homepage": "https://github.com/Tinder/sharded-redis-client",
   "dependencies": {
     "async": "2.6.0",
+    "bluebird": "3.5.1",
+    "brakes": "2.6.0",
     "circuit-breaker": "git://github.com/Tinder/circuit-breaker",
     "hiredis": "0.5.0",
     "lodash": "4.17.4",