add socket connection to block explorer

lib/blockexplorer.js

@@ -7,6 +7,7 @@ log.debug = log.verbose;
 
 var Explorers = require('bitcore-explorers');
 var request = require('request');
+var io = require('socket.io-client');
 
 
 function BlockExplorer(opts) {
@@ -15,7 +16,6 @@ function BlockExplorer(opts) {
   var network = opts.network || 'livenet';
 
   var url;
-
   switch (provider) {
     case 'insight':
       switch (network) {
@@ -29,7 +29,8 @@ function BlockExplorer(opts) {
       }
       var bc = new Explorers.Insight(url, network);
       bc.getTransactions = _.bind(getTransactionsInsight, bc, url);
-      this.blockExplorer = bc;
+      bc.initSocket = _.bind(initSocketInsight, bc, url);
+      return bc;
       break;
     default:
       throw new Error('Provider ' + provider + ' not supported');
@@ -51,4 +52,9 @@ function getTransactionsInsight(url, addresses, cb) {
   });
 };
 
+function initSocketInsight(url) {
+  var socket = io.connect(url, {});
+  return socket;
+};
+
 module.exports = BlockExplorer;

lib/model/notification.js

@@ -12,8 +12,8 @@ var Uuid = require('uuid');
  * txProposalFinallyRejected - txProposalId
  * txProposalFinallyAccepted - txProposalId
  *
- * newIncommingTx (amount)
- * newOutgoingTx - (txProposalId, txid)
+ * NewIncommingTx (address, txid)
+ * NewOutgoingTx - (txProposalId, txid)
  *
  * data Examples:
  * { amount: 'xxx', address: 'xxx'}

lib/server.js

@@ -337,7 +337,7 @@ WalletService.prototype.verifyMessageSignature = function(opts, cb) {
 
 WalletService.prototype._getBlockExplorer = function(provider, network) {
   if (!this.blockExplorer) {
-    this.blockexplorer = new BlockExplorer({
+    this.blockExplorer = new BlockExplorer({
       provider: provider,
       network: network,
     });

lib/wsapp.js

@@ -9,12 +9,22 @@ var querystring = require('querystring');
 var bodyParser = require('body-parser')
 var Uuid = require('uuid');
 
+var WalletUtils = require('bitcore-wallet-utils');
+var Bitcore = WalletUtils.Bitcore;
 var WalletService = require('./server');
+var BlockExplorer = require('./blockexplorer');
+
+var Notification = require('./model/notification');
 
 log.debug = log.verbose;
 log.level = 'debug';
 
-var subscriptions = {};
+var io;
+var blockExplorerSockets = {};
+var subscriptions = {
+  wallets: {},
+  addresses: {},
+};
 
 var WsApp = function() {};
 
@@ -24,37 +34,71 @@ WsApp._unauthorized = function() {
 };
 
 WsApp.subscribeAddresses = function(walletId, addresses) {
-  console.log('*** [wsapp.js ln27] subscribing:', addresses, walletId); // TODO
+  if (!addresses || addresses.length == 0) return;
 
-  _.each([].concat(addresses), function(address) {
+  function handlerFor(walletId, address, txid) {
+    var notification = Notification.create({
+      walletId: walletId,
+      type: 'NewIncommingTx',
+      data: {
+        address: address,
+        txid: txid,
+      },
+    });
+    WsApp._sendNotification(notification);
+  };
+
+  var addresses = [].concat(addresses);
+  var network = Bitcore.Address.fromString(addresses[0]).network;
+  var socket = blockExplorerSockets[network];
+  _.each(addresses, function(address) {
     subscriptions[address] = walletId;
+    socket.emit('subscribe', address);
+    socket.on(address, _.bind(handlerFor, null, walletId, address));
   });
 };
 
 WsApp.subscribeWallet = function(serviceInstance) {
-  // TODO: optimize!
+  var walletId = serviceInstance.walletId;
+  if (subscriptions.wallets[walletId]) return;
+
+  subscriptions.wallets[walletId] = true;
   serviceInstance.getMainAddresses({}, function(err, addresses) {
     if (err) {
+      delete subscriptions.wallets[walletId];
       log.warn('Could not subscribe to addresses for wallet ' + serviceInstance.walletId);
       return;
     }
-    WsApp.subscribeAddress(_.pluck(addresses, 'address'), serviceInstance.walletId);
+    WsApp.subscribeAddresses(serviceInstance.walletId, _.pluck(addresses, 'address'));
   });
 };
 
+WsApp._sendNotification = function(notification) {
+  if (notification.type == 'NewAddress') {
+    WsApp.subscribeAddresses(notification.walletId, notification.data.address);
+  }
+  io.to(notification.walletId).emit('notification', notification);
+};
+
+WsApp._initBlockExplorerSocket = function(provider, network) {
+  var explorer = new BlockExplorer({
+    provider: provider,
+    network: network,
+  });
+
+  blockExplorerSockets[network] = explorer.initSocket();
+};
+
 WsApp.start = function(server) {
-  var self = this;
+  io = require('socket.io')(server);
 
-  var io = require('socket.io')(server);
+  WsApp._initBlockExplorerSocket('insight', 'testnet');
+  WsApp._initBlockExplorerSocket('insight', 'livenet');
 
-  WalletService.onNotification(function(serviceInstance, args) {
-    var walletId = serviceInstance.walletId || args.walletId;
-    if (!walletId) return;
+  WalletService.onNotification(function(serviceInstance, notification) {
+    if (!notification.walletId) return;
 
-    if (args.type == 'NewAddress') {
-      WsApp.subscribeAddress(walletId, args.address);
-    }
-    io.to(walletId).emit('notification', args);
+    WsApp._sendNotification(notification);
   });
 
   io.on('connection', function(socket) {
@@ -64,13 +108,13 @@ WsApp.start = function(server) {
     socket.on('authorize', function(data) {
       if (data.message != socket.nonce) return WsApp.unauthorized();
 
-      WalletService.getInstanceWithAuth(data, function(err, res) {
+      WalletService.getInstanceWithAuth(data, function(err, service) {
         if (err) return WsApp.unauthorized();
 
-        socket.join(res.walletId);
+        socket.join(service.walletId);
         socket.emit('authorized');
 
-        WsApp.subscribeWallet(res);
+        WsApp.subscribeWallet(service);
       });
     });
   });