Merge pull request #153 from maraoz/improve/broker

Improve broker for development

app/controllers/socket.js

@@ -7,16 +7,24 @@ var mdb = require('../../lib/MessageDb').default();
 var microtime = require('microtime');
 var enableMessageBroker;
 
+var verbose = false;
+var log = function() {
+  if (verbose) {
+    console.log(arguments);
+  }
+}
+
 module.exports.init = function(io_ext, config) {
   enableMessageBroker = config ? config.enableMessageBroker : false;
   ios = io_ext;
   if (ios) {
     // when a new socket connects
     ios.sockets.on('connection', function(socket) {
+      log('New connection from ' + socket.id);
       // when it subscribes, make it join the according room
       socket.on('subscribe', function(topic) {
         if (socket.rooms.length === 1) {
-          console.log('subscribe to ' + topic);
+          log('subscribe to ' + topic);
           socket.join(topic);
         }
       });
@@ -24,6 +32,8 @@ module.exports.init = function(io_ext, config) {
       if (enableMessageBroker) {
         // when it requests sync, send him all pending messages
         socket.on('sync', function(ts) {
+          log('Sync requested by ' + socket.id);
+          log('    from timestamp '+ts);
           var rooms = socket.rooms;
           if (rooms.length !== 2) {
             socket.emit('insight-error', 'Must subscribe with public key before syncing');
@@ -31,10 +41,12 @@ module.exports.init = function(io_ext, config) {
           }
           var to = rooms[1];
           var upper_ts = Math.round(microtime.now());
+          log('    to timestamp '+upper_ts);
           mdb.getMessages(to, ts, upper_ts, function(err, messages) {
             if (err) {
               throw new Error('Couldn\'t get messages on sync request: ' + err);
             }
+            log('\tFound ' + messages.length + ' message' + (messages.length !== 1 ? 's' : ''));
             for (var i = 0; i < messages.length; i++) {
               broadcastMessage(messages[i], socket);
             }
@@ -43,12 +55,19 @@ module.exports.init = function(io_ext, config) {
 
         // when it sends a message, add it to db
         socket.on('message', function(m) {
+          log('Message sent from ' + m.pubkey + ' to ' + m.to);
           mdb.addMessage(m, function(err) {
             if (err) {
               throw new Error('Couldn\'t add message to database: ' + err);
             }
           });
         });
+
+
+        // disconnect handler
+        socket.on('disconnect', function() {
+          log('disconnected ' + socket.id);
+        });
       }
     });
     if (enableMessageBroker)
@@ -103,6 +122,7 @@ module.exports.broadcastSyncInfo = function(historicSync) {
 var broadcastMessage = module.exports.broadcastMessage = function(message, socket) {
   if (ios) {
     var s = socket || ios.sockets.in(message.to);
+    log('sending message to ' + message.to);
     s.emit('message', message);
   }
 

examples/node-client.js

@@ -0,0 +1,34 @@
+var io = require('socket.io-client');
+
+var bitcore = require('bitcore');
+var util = bitcore.util;
+var Key = bitcore.Key;
+var AuthMessage = bitcore.AuthMessage;
+var Buffer = bitcore.Buffer;
+
+var socket = io.connect('http://localhost:3001', {
+  reconnection: false
+});
+
+var pk = Key.generateSync();
+var pubkey = pk.public.toString('hex');
+socket.emit('subscribe', pubkey);
+socket.emit('sync');
+
+
+
+socket.on('connect', function() {
+  console.log('connected as ' + pubkey);
+});
+
+socket.on('message', function(m) {
+  var data = AuthMessage.decode(pk, m);
+  console.log('message received ' + data.payload);
+  var echo = AuthMessage.encode(m.pubkey, pk, data.payload);
+  socket.emit('message', echo);
+});
+
+
+socket.on('error', function(err) {
+  console.log(err);
+});

lib/PeerSync.js

@@ -23,8 +23,13 @@ function PeerSync(opts) {
   this.peerman = new PeerManager(pmConfig);
   this.load_peers();
   this.sync = new Sync(opts);
+  this.verbose = opts.verbose || false;
 }
 
+PeerSync.prototype.log = function() {
+  if (this.verbose) console.log(arguments);
+};
+
 PeerSync.prototype.load_peers = function() {
   this.peerdb = [{
     ipv4: config.bitcoind.p2pHost,
@@ -59,12 +64,12 @@ PeerSync.prototype._broadcastAddr = function(txid, addrs) {
 PeerSync.prototype.handleTx = function(info) {
   var self =this;
   var tx = this.sync.txDb.getStandardizedTx(info.message.tx);
-  console.log('[p2p_sync] Handle tx: ' + tx.txid);
+  self.log('[p2p_sync] Handle tx: ' + tx.txid);
   tx.time = tx.time || Math.round(new Date().getTime() / 1000);
 
   this.sync.storeTx(tx, function(err, relatedAddrs) {
     if (err) {
-      console.log('[p2p_sync] Error in handle TX: ' + JSON.stringify(err));
+      self.log('[p2p_sync] Error in handle TX: ' + JSON.stringify(err));
     }
     else if (self.shouldBroadcast) {
       sockets.broadcastTx(tx);
@@ -78,7 +83,7 @@ PeerSync.prototype.handleBlock = function(info) {
   var self = this;
   var block = info.message.block;
   var blockHash = bitcoreUtil.formatHashFull(block.calcHash());
-  console.log('[p2p_sync] Handle block: %s (allowReorgs: %s)', blockHash, self.allowReorgs);
+  self.log('[p2p_sync] Handle block: %s (allowReorgs: %s)', blockHash, self.allowReorgs);
 
   var tx_hashes = block.txs.map(function(tx) {
     return bitcoreUtil.formatHashFull(tx.hash);
@@ -90,13 +95,13 @@ PeerSync.prototype.handleBlock = function(info) {
     'previousblockhash': bitcoreUtil.formatHashFull(block.prev_hash),
   }, self.allowReorgs, function(err, height) {
     if (err && err.message.match(/NEED_SYNC/) && self.historicSync) {
-      console.log('[p2p_sync] Orphan block received. Triggering sync');
+      self.log('[p2p_sync] Orphan block received. Triggering sync');
       self.historicSync.start({forceRPC:1}, function(){
-        console.log('[p2p_sync] Done resync.');
+        self.log('[p2p_sync] Done resync.');
       });
     }
     else if (err) {
-      console.log('[p2p_sync] Error in handle Block: ', err);
+      self.log('[p2p_sync] Error in handle Block: ', err);
     }
     else {
       if (self.shouldBroadcast) {
@@ -111,7 +116,7 @@ PeerSync.prototype.handleBlock = function(info) {
 PeerSync.prototype.handleConnected = function(data) {
   var peerman = data.pm;
   var peers_n = peerman.peers.length;
-  console.log('[p2p_sync] Connected to ' + peers_n + ' peer' + (peers_n !== 1 ? 's' : ''));
+  this.log('[p2p_sync] Connected to ' + peers_n + ' peer' + (peers_n !== 1 ? 's' : ''));
 };
 
 PeerSync.prototype.run = function() {

package.json

@@ -66,10 +66,11 @@
     "preconditions": "^1.0.7",
     "should": "~2.1.1",
     "sinon": "~1.7.3",
-    "socket.io": "~1.0.4",
     "soop": "=0.1.5",
     "winston": "*",
-    "xmlhttprequest": "~1.6.0"
+    "xmlhttprequest": "~1.6.0",
+    "socket.io": "1.0.6",
+    "socket.io-client": "1.0.6"
   },
   "devDependencies": {
     "chai": "*",