Merge pull request #60 from nodejitsu/v0.7.x

v0.7.x

Author: jcrugzz

README.md

@@ -51,12 +51,11 @@ Here is a simple example of a [Reactor](#reactors) server that will send an emai
       })
     ],
     //
-    // Add Reconnect logic that uses node-backoff
+    // Add Reconnect logic that uses `back`
     //
     reconnect: {
-      type: 'exponential',
-      maxTries: 2,
-      initialDelay: 100,
+      retries: 2,
+      minDelay: 100,
       maxDelay: 300
     }
   }).connect(1337);

lib/godot/net/client.js

@@ -7,9 +7,11 @@
 
 var dgram = require('dgram'),
     net = require('net'),
-    util = require('util'),
-    backoff = require('backoff'),
-    EventEmitter = require('events').EventEmitter;
+    utile = require('utile'),
+    clone = utile.clone,
+    back = require('back'),
+    EventEmitter = require('events').EventEmitter,
+    noop = function () {};
 
 //
 // ### function Server (options)
@@ -29,13 +31,19 @@ var Client = module.exports = function Client(options) {
     throw new Error('Cannot create client without type: udp, tcp, unix');
   }
 
+  if(typeof options.reconnect !== 'undefined'
+     && typeof options.reconnect!== 'object') {
+    throw new Error('Reconnect must be a defined object if used')
+  }
+
   var self = this;
 
   this.type      = options.type;
   this.host      = options.host;
   this.port      = options.port;
   this.path      = options.path;
   this.reconnect = options.reconnect;
+  this.attempt = null;
   this.producers = {};
   this.handlers  = {
     data: {},
@@ -52,7 +60,7 @@ var Client = module.exports = function Client(options) {
 //
 // Inherit from EventEmitter
 //
-util.inherits(Client, EventEmitter);
+utile.inherits(Client, EventEmitter);
 
 //
 // ### function add (producer)
@@ -126,28 +134,7 @@ Client.prototype.write = function (data) {
 // Opens the underlying network connection for this client.
 //
 Client.prototype.connect = function (port, host, callback) {
-  var self = this,
-      connectBackoff, backoffType;
-
-  if (this.reconnect) {
-    if (typeof this.reconnect === 'object') {
-      backoffType = this.reconnect.type || 'exponential';
-      connectBackoff = backoff[backoffType](this.reconnect);
-      connectBackoff.failAfter(this.reconnect.maxTries || 10);
-    }
-    else {
-      connectBackoff = backoff.exponential();
-      connectBackoff.failAfter(10);
-    }
-
-    connectBackoff.on('fail', function (err) {
-      self.emit('error', err);
-    });
-
-    connectBackoff.on('ready', function () {
-      connect();
-    });
-  }
+  var self = this;
 
   //
   // Do some fancy arguments parsing to support everything
@@ -173,8 +160,41 @@ Client.prototype.connect = function (port, host, callback) {
       : self.emit('error', err) ;
   }
 
+  function reconnect(err) {
+    self.attempt = self.attempt || clone(self.reconnect);
+    //
+    // Remark: Terminate the backoff when we have hit our fail condition with
+    // a noop to avoid an if statement
+    //
+    // TODO: Make this less coupled (I feel like i want this contained in
+    // `back` but eh)
+    //
+    return self.terminate
+      ? noop()
+      : back(function (fail, backoff) {
+        //
+        // Remark: We are done here, emit error and set termination
+        //
+        if (fail) {
+          self.terminate = true;
+          self.attempt = null;
+          return self.emit('error', err);
+        }
+        //
+        // So we can listen on when reconnect events are about to fire
+        //
+        self.emit('reconnect');
+        //
+        // Attempt a CONNECT!
+        //
+        return connect();
+      }, self.attempt);
+  }
+
   function onError(err) {
-    return connectBackoff ? connectBackoff.backoff(err) : self.emit('error', err);
+    return self.reconnect
+      ? reconnect(err)
+      : self.emit('error', err);
   }
 
   function connect() {
@@ -193,9 +213,10 @@ Client.prototype.connect = function (port, host, callback) {
 
     self.socket.on('error', onError);
     self.socket.on('connect', function () {
-      if (connectBackoff) {
-        connectBackoff.reset();
-      }
+      //
+      // Remark: We have successfully connected so reset the terminate variable
+      //
+      self.terminate = false;
       self.emit('connect');
     });
   }

lib/godot/net/server.js

@@ -71,6 +71,7 @@ utile.inherits(Server, events.EventEmitter);
 Server.prototype.add = function (reactor) {
   this.emit('add', reactor);
   reactor.on('error', this.emit.bind(this, 'error'));
+  reactor.on('reactor:error', this.emit.bind(this, 'reactor:error'));
   this.reactors[reactor.id] = reactor;
 };
 

lib/godot/producer/producer.js

@@ -8,10 +8,11 @@
 var stream = require('stream'),
     ip = require('ip'),
     utile = require('utile'),
-    uuid = require('node-uuid');
+    uuid = require('node-uuid'),
+    tick = typeof setImmediate == 'undefined'
+      ? process.nextTick
+      : setImmediate;
 
-//
-// ### function Producer (options)
 // #### @options {Object} Options for this producer.
 // Constructor function for the Producer object responsible
 // for creating events to process.
@@ -107,7 +108,7 @@ Object.keys(Producer.prototype.types).forEach(function (key) {
         //
         if (value === 0) {
           return (function tickProduce() {
-            process.nextTick(function () {
+            tick(function () {
               self.produce();
               tickProduce();
             });

lib/godot/reactor/email.js

@@ -76,7 +76,7 @@ Email.prototype.write = function (data) {
     self._last = new Date();
 
     return err
-      ? self.emit('error', err)
+      ? self.emit('reactor:error', err)
       : self.emit('data', data);
   });
 };

lib/godot/reactor/graphite.js

@@ -82,7 +82,7 @@ Graphite.prototype.write = function (data) {
   metrics[metricName] = this.meta ? data.meta[this.meta] : data.metric;
   this.client.write(metrics, data.time, function (err) {
     self._last = now;
-    if (err) { return self.emit('error', err) }
+    if (err) { return self.emit('reactor:error', err) }
   });
 
   self.emit('data', data);

lib/godot/reactor/map.js

@@ -55,13 +55,13 @@ Map.prototype.write = function (data) {
   if (!this.passThrough) {
     return this.mapFn(data, function (err, data) {
       return err
-        ? self.emit('error', err)
+        ? self.emit('reactor:error', err)
         : self.emit('data', data);
     });
   }
 
   this.mapFn(data, function (err) {
-    if (err) { self.emit('error', err) }
+    if (err) { self.emit('reactor:error', err) }
   });
 
   this.emit('data', data);

lib/godot/reactor/reactor.js

@@ -75,6 +75,7 @@ Reactor.prototype.createStream = function (source) {
   return this.reactors.reduce(function (last, nextOptions) {
     var stream = wrapStream(nextOptions.Factory, nextOptions.args || []);
     stream.on('error', self.emit.bind(self, 'error'));
+    stream.on('reactor:error', self.emit.bind(self, 'reactor:error'));
     return last.pipe(stream);
   }, source);
 };

lib/godot/reactor/redis.js

@@ -34,7 +34,7 @@ var Redis = module.exports = function Redis(options, redisFn) {
   if (!this.client) {
     this.client = redis.createClient(this.port, this.host, this.redisOptions);
 
-    this.client.on('error', this.emit.bind(this, 'error'));
+    this.client.on('error', this.emit.bind(this, 'reactor:error'));
     if (this.password) {
       this.client.auth(this.password, function () {
         // Remark: What if data is sent before we are authenticated?
@@ -56,7 +56,7 @@ utile.inherits(Redis, ReadWriteStream);
 Redis.prototype.write = function (data) {
   var self = this;
   this.redisFn(this.client, data, function (err, data) {
-    if (err) { return self.emit('error', err) }
+    if (err) { return self.emit('reactor:error', err) }
   });
 
   this.emit('data', data);

lib/godot/reactor/sms.js

@@ -80,7 +80,7 @@ Sms.prototype.write = function (data) {
     self._last = new Date();
 
     return err
-      ? self.emit('error', err)
+      ? self.emit('reactor:error', err)
       : self.emit('data', data);
   });
-};
+};