introduce message broker through rebase

lib/blockchainmonitor.js

@@ -10,15 +10,11 @@ var BlockchainExplorer = require('./blockchainexplorer');
 var Storage = require('./storage');
 var MessageBroker = require('./messagebroker');
 
-<<<<<<< HEAD
 var Notification = require('./model/notification');
 
 function BlockchainMonitor() {};
 
 BlockchainMonitor.prototype.start = function(opts, cb) {
-=======
-function BlockchainMonitor(opts) {
->>>>>>> refactor notification dispatching
   opts = opts || {};
   $.checkArgument(opts.blockchainExplorerOpts);
   $.checkArgument(opts.storageOpts);
@@ -69,14 +65,8 @@ BlockchainMonitor.prototype._initExplorer = function(provider, network, url) {
   });
   socket.on('tx', _.bind(self._handleIncommingTx, self));
 
-<<<<<<< HEAD
   return explorer;
 };
-=======
-BlockchainMonitor.prototype.subscribeAddresses = function(walletService, addresses) {
-  $.checkArgument(walletService);
-  $.checkArgument(walletService.walletId);
->>>>>>> refactor notification dispatching
 
 BlockchainMonitor.prototype._handleIncommingTx = function(data) {
   var self = this;
@@ -84,27 +74,15 @@ BlockchainMonitor.prototype._handleIncommingTx = function(data) {
 
   if (!data || !data.vout) return;
 
-<<<<<<< HEAD
   var outs = _.compact(_.map(data.vout, function(v) {
     var addr = _.keys(v)[0];
     var startingChar = addr.charAt(0);
     if (startingChar != '2' && startingChar != '3') return;
-=======
-  function handlerFor(address, txid) {
-    var data = {
-      walletId: this.walletId,
-      address: address,
-      txid: txid,
-    };
-    self.emit('NewIncomingTx', data, this);
-  };
->>>>>>> refactor notification dispatching
 
     return {
       address: addr,
       amount: +v[addr]
     };
-<<<<<<< HEAD
   }));
   if (_.isEmpty(outs)) return;
 
@@ -122,23 +100,12 @@ BlockchainMonitor.prototype._handleIncommingTx = function(data) {
     });
   }, function(err) {
     return;
-=======
-  };
-
-  var addresses = [].concat(addresses);
-  var network = Bitcore.Address.fromString(addresses[0]).network.name;
-  var subscriber = self.subscriber[network];
-  _.each(addresses, function(address) {
-    self.subscriptions[walletId].addresses.push(address);
-    subscriber.subscribe(address, _.bind(handlerFor, walletService, address));
->>>>>>> refactor notification dispatching
   });
 };
 
 BlockchainMonitor.prototype._createNotification = function(walletId, txid, address, amount, cb) {
   var self = this;
 
-<<<<<<< HEAD
   var n = Notification.create({
     type: 'NewIncomingTx',
     data: {
@@ -150,17 +117,6 @@ BlockchainMonitor.prototype._createNotification = function(walletId, txid, addre
   });
   self.storage.storeNotification(walletId, n, function() {
     self.messageBroker.send(n)
-=======
-  var walletId = walletService.walletId;
-  if (self.subscriptions[walletId]) return;
-
-  walletService.getMainAddresses({}, function(err, addresses) {
-    if (err) {
-      delete self.subscriptions[walletId];
-      return cb(new Error('Could not subscribe to addresses for wallet ' + walletId));
-    }
-    self.subscribeAddresses(walletService, _.pluck(addresses, 'address'));
->>>>>>> refactor notification dispatching
     return cb();
   });
 };

lib/server.js

@@ -117,7 +117,6 @@ WalletService.initialize = function(opts, cb) {
       log.error('Could not initialize', err);
       throw err;
     }
-  ], function() {
     initialized = true;
     return cb();
   });

test/integration/server.js

@@ -3103,9 +3103,6 @@ describe('Wallet service', function() {
         }, done);
       });
     });
-    afterEach(function() {
-      NotificationBroadcaster.removeAllListeners();
-    });
 
     it('should notify copayers a new tx proposal has been created', function(done) {
       helpers.stubUtxos(server, wallet, [1, 1], function() {