Merge pull request #17 from bitpay/promisify

EventBus with promises

.jshintrc

@@ -41,5 +41,8 @@
         "it",
         "module",
         "require"
-    ]
+    ],
+    "globals" : {
+      "Promise"   : true
+    }
 }

lib/databases/transaction.js

@@ -45,20 +45,7 @@ var db = imports.db || levelup(config.leveldb + '/txs', {
   maxOpenFiles: MAX_OPEN_FILES
 });
 var PoolMatch = imports.poolMatch || require('soop').load('./PoolMatch', config);
-// This is 0.1.2 = > c++ version of base58-native
+var Address = require('bitcore').Address;
-var base58 = require('base58-native').base58Check;
-var encodedData = require('soop').load('bitcore/util/EncodedData', {
-  base58: base58
-});
-var versionedData = require('soop').load('bitcore/util/VersionedData', {
-  parent: encodedData
-});
-
-var Address = require('soop').load('bitcore/lib/Address', {
-  parent: versionedData
-});
-
-
 
 var TransactionDb = function() {
   TransactionDb.super(this, arguments);

lib/eventbus.js

@@ -0,0 +1,58 @@
+'use strict';
+
+
+var bitcore = require('bitcore');
+var Promise = require('bluebird');
+var $ = bitcore.util.preconditions;
+var _ = bitcore.deps._;
+var EventEmitter = require('events').EventEmitter;
+var util = require('util');
+
+function EventBus() {
+  this.handlers = {};
+}
+util.inherits(EventBus, EventEmitter);
+
+EventBus.prototype.process = function(e) {
+  $.checkArgument(_.isObject(e));
+
+  var self = this;
+  var done = [];
+
+  var processEvent = function(event) {
+    done = done.concat(event);
+    var handlers = self.handlers[event.constructor.name] || [];
+    var whenHandlersResolve = Promise.all(handlers.map(function(handler) {
+      return handler(event);
+    }));
+    return whenHandlersResolve.each(function(events) {
+      if (_.isUndefined(events)) {
+        events = [];
+      }
+      if (!_.isArray(events)) {
+        events = [events];
+      }
+      return Promise.all(
+        events.map(processEvent)
+      );
+    });
+  };
+  var eventsEmitted = processEvent(e)
+    .then(function() {
+      done.forEach(function(event) {
+        self.emit(event.name || event.constructor.name, event);
+      });
+    });
+  return eventsEmitted;
+};
+
+
+EventBus.prototype.register = function(clazz, handler) {
+  $.checkArgument(_.isFunction(handler));
+  var name = clazz.name;
+  this.handlers[name] = this.handlers[name] || [];
+  this.handlers[name].push(handler);
+};
+
+
+module.exports = EventBus;

lib/funnel.js

@@ -1,45 +0,0 @@
-'use strict';
-
-
-var bitcore = require('bitcore');
-var $ = bitcore.util.preconditions;
-var _ = bitcore.deps._;
-var EventEmitter = require('events').EventEmitter;
-var util = require('util');
-
-function Funnel() {
-  this.handlers = {};
-}
-util.inherits(Funnel, EventEmitter);
-
-Funnel.prototype.process = function(e) {
-  var queue = [];
-  var done = [];
-  queue.push(e);
-  while (queue.length !== 0) {
-    var event = queue.shift();
-    var handlers = this.handlers[event.constructor.name] || [];
-    handlers.forEach(function(handler) {
-      var responses = handler(event);
-      if (responses && responses.length > 0) {
-        queue = queue.concat(responses);
-      }
-    });
-    done.push(event);
-  }
-  done.forEach(function(event) {
-    //that.emit(event.name, event);
-  });
-
-};
-
-
-Funnel.prototype.register = function(clazz, handler) {
-  $.checkArgument(_.isFunction(handler));
-  var name = clazz.name;
-  this.handlers[name] = this.handlers[name] || [];
-  this.handlers[name].push(handler);
-};
-
-
-module.exports = Funnel;

package.json

@@ -43,15 +43,13 @@
     "start": "node node_modules/grunt-cli/bin/grunt"
   },
   "dependencies": {
-    "async": "*",
+    "async": "0.9.0",
-    "base58-native": "0.1.2",
     "bignum": "*",
-    "bitauth": "^0.1.1",
+    "bitcore": "0.10.4",
-    "bitcore": "git://github.com/bitpay/bitcore.git#aa41c70cff2583d810664c073a324376c39c8b36",
+    "bluebird": "^2.9.12",
     "bufferput": "git://github.com/bitpay/node-bufferput.git",
     "buffertools": "*",
     "commander": "^2.3.0",
-    "connect-ratelimit": "git://github.com/dharmafly/connect-ratelimit.git#0550eff209c54f35078f46445000797fa942ab97",
     "cron": "^1.0.4",
     "express": "~3.4.7",
     "glob": "*",
@@ -61,7 +59,6 @@
     "microtime": "^0.6.0",
     "mkdirp": "^0.5.0",
     "moment": "~2.5.0",
-    "nodemailer": "^1.3.0",
     "preconditions": "^1.0.7",
     "request": "^2.48.0",
     "socket.io": "1.0.6",
@@ -71,17 +68,9 @@
     "xmlhttprequest": "~1.6.0"
   },
   "devDependencies": {
-    "gulp": "^3.8.10",
+    "bitcore-build": "bitpay/bitcore-build",
-    "gulp-bump": "^0.1.11",
-    "gulp-coveralls": "^0.1.3",
-    "gulp-jshint": "^1.9.0",
-    "gulp-mocha": "^2.0.0",
-    "gulp-shell": "^0.2.10",
-    "istanbul": "^0.3.5",
-    "mocha": "^2.0.1",
-    "plato": "^1.3.0",
     "chai": "*",
-    "memdown": "^0.10.2",
+    "gulp": "^3.8.10",
     "should": "^2.1.1",
     "sinon": "^1.10.3"
   }

test/eventbus.js

@@ -0,0 +1,120 @@
+'use strict';
+
+var chai = require('chai');
+var should = chai.should();
+var sinon = require('sinon');
+
+var Promise = require('bluebird');
+var EventBus = require('../lib/eventbus');
+Promise.longStackTraces();
+
+describe('EventBus', function() {
+
+  it('instantiate', function() {
+    var bus = new EventBus();
+    should.exist(bus);
+  });
+
+  describe('process', function() {
+    function FooEvent() {}
+
+    function BarEvent() {}
+    var foo = new FooEvent();
+    var bar = new BarEvent();
+    foo.x = 2;
+    bar.y = 3;
+
+    it('no handlers registered', function() {
+      var bus = new EventBus();
+      bus.process.bind(bus, foo).should.not.throw();
+    });
+    it('simple handler gets called', function(cb) {
+      var bus = new EventBus();
+      bus.register(FooEvent, function(e) {
+        e.x.should.equal(foo.x);
+        cb();
+      });
+      bus.process(foo);
+    });
+    it('other event does not get called', function() {
+      var bus = new EventBus();
+      var spy = sinon.spy();
+      bus.register(FooEvent, spy);
+      bus.process(bar);
+      spy.callCount.should.equal(0);
+    });
+    it('foo returns bar', function(cb) {
+      var bus = new EventBus();
+      bus.register(FooEvent, function(e) {
+        var b = new BarEvent();
+        b.y = e.x;
+        return b;
+      });
+      bus.register(BarEvent, function(e) {
+        e.y.should.equal(foo.x);
+        cb();
+      });
+      bus.process(foo);
+    });
+    var b1 = new BarEvent();
+    b1.x = 42;
+    var b2 = new BarEvent();
+    b2.x = 69;
+    it('foo returns two bars', function() {
+      var bus = new EventBus();
+      var spy = sinon.spy();
+      bus.register(FooEvent, function() {
+        return [b1, b2];
+      });
+      bus.register(BarEvent, spy);
+      bus.process(foo);
+      spy.callCount.should.equal(2);
+    });
+    it('foo returns two bars and emits external events', function(cb) {
+      var bus = new EventBus();
+      var spy = sinon.spy(bus, 'emit');
+      bus.register(FooEvent, function() {
+        return [b1, b2];
+      });
+      bus.process(foo)
+        .then(function() {
+          spy.calledWith('BarEvent', b1).should.equal(true);
+          spy.calledWith('BarEvent', b2).should.equal(true);
+        })
+        .then(cb);
+    });
+    it('foo returns two async bars', function(cb) {
+      var bus = new EventBus();
+      var spy = sinon.spy();
+      bus.register(FooEvent, function() {
+        return Promise.resolve([b1, b2]).delay(1);
+      });
+      bus.register(BarEvent, spy);
+      bus.process(foo)
+        .then(function() {
+          spy.callCount.should.equal(2);
+        })
+        .then(cb);
+    });
+    it('events are not externalized when async processing fails', function(cb) {
+      var bus = new EventBus();
+      var spy = sinon.spy(bus, 'emit');
+      var err = new Error();
+      bus.register(FooEvent, function() {
+        return Promise.resolve([b1, b2]).delay(1);
+      });
+      bus.register(BarEvent, function(e) {
+        if (e.x === b1.x) {
+          throw err;
+        }
+      });
+      bus.process(foo)
+        .catch(function(reason) {
+          reason.should.equal(err);
+          spy.callCount.should.equal(0);
+          cb();
+        });
+    });
+  });
+
+});

test/funnel.js

@@ -1,70 +0,0 @@
-'use strict';
-
-var chai = require('chai');
-var should = chai.should();
-var sinon = require('sinon');
-
-var Funnel = require('../lib/funnel');
-
-describe('Funnel', function() {
-
-  it('instantiate', function() {
-    var f = new Funnel();
-    should.exist(f);
-  });
-
-  describe('process', function() {
-    function FooEvent() {}
-    function BarEvent() {}
-    var foo = new FooEvent();
-    var bar = new BarEvent();
-    foo.x = 2;
-    bar.y = 3;
-
-    it('no handlers registered', function() {
-      var f = new Funnel();
-      f.process.bind(f, foo).should.not.throw();
-    });
-    it('simple handler gets called', function(cb) {
-      var f = new Funnel();
-      f.register(FooEvent, function(e) {
-        e.x.should.equal(foo.x);
-        cb();
-      });
-      f.process(foo);
-    });
-    it('other event does not get called', function() {
-      var f = new Funnel();
-      var spy = sinon.spy();
-      f.register(FooEvent, spy);
-      f.process(bar);
-      spy.callCount.should.equal(0);
-    });
-    it('foo returns bar', function(cb) {
-      var f = new Funnel();
-      f.register(FooEvent, function(e) {
-        var b = new BarEvent();
-        b.y = e.x;
-        return [b];
-      });
-      f.register(BarEvent, function(e) {
-        e.y.should.equal(foo.x);
-        cb();
-      });
-      f.process(foo);
-    });
-    it('foo returns two bars', function() {
-      var f = new Funnel();
-      var spy = sinon.spy();
-      f.register(FooEvent, function() {
-        var b1 = new BarEvent();
-        var b2 = new BarEvent();
-        return [b1, b2];
-      });
-      f.register(BarEvent, spy);
-      f.process(foo);
-      spy.callCount.should.equal(2);
-    });
-  });
-
-});