Autorebalance on startup

lib/server/index.js

@@ -40,6 +40,11 @@ class Server extends http.Server {
     this.loaded = false;
     this._node = opts.node || new Node();
     this._router = new Router(this._node);
+    this._node.on('ringchange', (evt) => {
+      timer.rebalance(evt, this._node, (data) => {
+        this.proxy(data);
+      });
+    });
   }
 
   /**
@@ -67,46 +72,49 @@ class Server extends http.Server {
   }
 
   /**
-   * Joins the node to the configured ring and starts the http server
-   * @method module:skyring/lib/server#listen
-   * @param {Number} port Port number to listen on
-   * @param {String} [host=localhost] host or ip address to listen on 
+   * Joins the node to the configured ring and starts the http server
+   * @method module:skyring/lib/server#listen
+   * @param {Number} port Port number to listen on
+   * @param {String} [host=localhost] host or ip address to listen on 
    * @param {Number} [backlog]
    * @param {Function} [callback] Callback function to call when the server is running
-   * @return {module:skyring/lib/server}
-   **/
+   * @return {module:skyring/lib/server}
+   **/
   listen(port, host, backlog, callback) {
     this._node.join(null, (err) => {
       if (err) return callback(err)
       this._node.handle(( req, res ) => {
         this._router.handle( req, res );
       })
       timer.watch('skyring', (err, data) => {
-        debug('fabricating request', data)
-        const opts = {
-          url: '/timer'
-        , method: 'POST'
-        , headers: {
-            "x-timer-id": data.id
-          }
-        , payload: JSON.stringify(data)
-        }
-        const res = new mock.Response();
-        const req = new mock.Request( opts );
-
-        this._router.handle( req, res );
+        this.proxy(data)
       });
       super.listen(port, host, backlog, callback)
     })
     return this;
   }
 
+  proxy(data){
+    debug('fabricating request', data)
+    const opts = {
+      url: '/timer'
+    , method: 'POST'
+    , headers: {
+        "x-timer-id": data.id
+      }
+    , payload: JSON.stringify(data)
+    }
+    const res = new mock.Response();
+    const req = new mock.Request( opts );
+
+    this._router.handle( req, res );
+  }
   /**
-   * Removes a server from the ring, closes the http server and redistributes
+   * Removes a server from the ring, closes the http server and redistributes
    * any pending timers
-   * @method module:skyring/lib/server#close
-   * @param {Function} callback A callback to be called when the server is completely shut down
-   **/
+   * @method module:skyring/lib/server#close
+   * @param {Function} callback A callback to be called when the server is completely shut down
+   **/
   close( cb ){
 
     super.close(()=>{

lib/server/node.js

@@ -88,6 +88,13 @@ if (err) throw err
           })
       });
       this._ring.setupChannel();
+      this._ring.on('ringChanged', ( evt ) => {
+        const added = evt.added;
+        if(!added.length) return;
+        if(added.length === 1 && added.indexOf(`${host}:${this._port}`) !== -1) return;
+        debug('node added', added)
+        this.emit('ringchange', evt);
+      })
       this._tchannel.listen( this._port, host, (er)=> {
         if(er) return cb(er)
         debug('tchannel listening on ', host, this._port);

lib/timer.js

@@ -15,6 +15,7 @@ const transports = require('./transports')
     , json       = require('./json')
     , debug      = require('debug')('skyring:timer')
     , rebalance  = require('debug')('skyring:rebalance')
+    , noop       = function(){}
     , cache      = new Map()
     ;
 
@@ -100,6 +101,31 @@ exports.delete = function(id, cb) {
   cb && setImmediate(cb, null);
 }
 
+
+exports.rebalance = function(opts, node, cb) {
+  const callback = cb || noop
+      , size = cache.size
+      , client = nats.client
+      ;
+
+  if( !size ) return;
+  const records = cache.values();
+  const run = ( obj ) => {
+    if (node.owns(obj.id)) return;
+    clearTimeout( obj.timer );
+    cache.delete( obj.id );
+    const data = Object.assign({}, obj.payload, {
+      id: obj.id
+    , created: obj.created
+    });
+    rebalance('no longer the owner of %s', obj.id);
+    cb(data)
+  }
+
+  for( var record of records ) {
+    run( record );
+  }
+}
 /**
  * Updates a timer inplace
  * @method module:skyring/lib/timer#update
@@ -127,8 +153,6 @@ exports.update = function update(id, body, cb){
  * @param {Nodeback} callback Node style callback to execute when the function is complete
  **/
 exports.shutdown = function shutdown(cb) {
-  BAIL = true;
-  const exit = false;
   const size = cache.size;
   const client = nats.client;
   let sent = 0;
@@ -140,10 +164,10 @@ exports.shutdown = function shutdown(cb) {
   NATS_SID = null;
 
   const run = ( obj ) => {
-    clearTimeout( record.timer );
-    const data = Object.assign({}, record.payload, {
-      id: record.id
-    , created: record.created
+    clearTimeout( obj.timer );
+    const data = Object.assign({}, obj.payload, {
+      id: obj.id
+    , created: obj.created
     , count: ++sent
     });