fix address streamed data

app/controllers/socket.js

@@ -41,7 +41,6 @@ module.exports.broadcastBlock = function(block) {
 };
 
 module.exports.broadcastAddressTx = function(address, tx) {
-  console.log('bcatx = '+address+' '+tx);
   if (ios) ios.sockets. in (address).emit(address, tx);
 };
 

lib/BlockDb.js

@@ -5,6 +5,7 @@ require('classtool');
 
 function spec(b) {
 
+  var superclass = b.superclass || require('events').EventEmitter;
   var TIMESTAMP_PREFIX  = 'bts-';     // b-ts-<ts> => <hash>
   var PREV_PREFIX       = 'bpr-';     // b-prev-<hash> => <prev_hash> 
   var NEXT_PREFIX       = 'bne-';     // b-next-<hash> => <next_hash> 
@@ -22,7 +23,9 @@ function spec(b) {
   var Rpc = b.rpc || require('./Rpc').class();
 
   var BlockDb = function() {
+    BlockDb.super(this, arguments);
   };
+  BlockDb.superclass = superclass;
 
   BlockDb.prototype.close = function(cb) {
     db.close(cb);
@@ -42,6 +45,7 @@ function spec(b) {
   // the block prev to the new block, nor TIP pointer
   //
   BlockDb.prototype.add = function(b, cb) {
+    var self = this;
     var time_key = TIMESTAMP_PREFIX +
       ( b.time || Math.round(new Date().getTime() / 1000) );
 
@@ -49,7 +53,12 @@ function spec(b) {
       .put(time_key, b.hash)
       .put(MAIN_PREFIX + b.hash, 1)
       .put(PREV_PREFIX + b.hash, b.previousblockhash)
-      .write(cb);
+      .write(function(err){
+        if (!err) {
+          self.emit('new_block', {blockid: b.hash});
+        }
+        cb(err);
+      });
   };
 
   BlockDb.prototype.getTip = function(cb) {

lib/PeerSync.js

@@ -6,7 +6,6 @@ function spec() {
   var CoinConst = require('bitcore/const');
   var coinUtil = require('bitcore/util/util');
   var Sync = require('./Sync').class();
-  var Script = require('bitcore/Script').class();
   var Peer = require('bitcore/Peer').class();
   var config = require('../config/config');
   var networks = require('bitcore/networks');
@@ -60,24 +59,12 @@ function spec() {
   };
 
   PeerSync.prototype.handleTx = function(info) {
-    var self =this;
     var tx = info.message.tx.getStandardizedObject();
+    tx.outs = info.message.tx.outs;
+    tx.ins = info.message.tx.ins;
     console.log('[p2p_sync] Handle tx: ' + tx.hash);
     tx.time = tx.time || Math.round(new Date().getTime() / 1000);
 
-    var to=0;
-    info.message.tx.outs.forEach( function(o) {
-      var s = new Script(o.s);
-      var addrs = self.sync.getAddrStr(s);
-
-      // support only for p2pubkey p2pubkeyhash and p2sh
-      if (addrs.length === 1) {
-        tx.out[to].addrStr = addrs[0];
-        tx.out[to].n = to;
-      }
-      to++;
-    });
-
     this.sync.storeTxs([tx], function(err) {
       if (err) {
         console.log('[p2p_sync] Error in handle TX: ' + JSON.stringify(err));

lib/Sync.js

@@ -6,28 +6,24 @@ require('classtool');
 function spec() {
   var sockets = require('../app/controllers/socket.js');
   var BlockDb = require('./BlockDb').class();
-  var bitutil = require('bitcore/util/util');
 
-  // This is 0.1.2 => c++ version of base57-native
-  var base58        = require('base58-native');
-  var encodedData   = require('bitcore/util/EncodedData').class({base58: base58});
-  var versionedData = require('bitcore/util/VersionedData').class({superclass: encodedData});
-  var Address = require('bitcore/Address').class({superclass: versionedData});
   var TransactionDb = require('./TransactionDb').class();
   var config = require('../config/config');
   var networks = require('bitcore/networks');
-  var Script = require('bitcore/Script').class();
   var async = require('async');
 
 
-  function Sync() {
+  function Sync() {}
-  }
 
   Sync.prototype.init = function(opts, cb) {
     var self = this;
     self.opts = opts;
     this.bDb = new BlockDb(opts);
     this.txDb = new TransactionDb(opts);
+    this.txDb.on('tx_for_address', this.handleTxForAddress.bind(this));
+    this.txDb.on('new_tx', this.handleNewTx.bind(this));
+    this.bDb.on('new_block', this.handleNewBlock.bind(this));
+
     this.network = config.network === 'testnet' ? networks.testnet : networks.livenet;
     return cb();
   };
@@ -43,8 +39,13 @@ function spec() {
   Sync.prototype.destroy = function(next) {
     var self = this;
     async.series([
-      function(b) { self.bDb.drop(b); },
+
-      function(b) { self.txDb.drop(b); },
+      function(b) {
+        self.bDb.drop(b);
+      },
+      function(b) {
+        self.txDb.drop(b);
+      },
     ], next);
   };
 
@@ -88,9 +89,9 @@ function spec() {
     var self = this;
     var oldTip, oldNext, needReorg = false;
     var newPrev = b.previousblockhash;
-    var updatedAddrs;
 
     async.series([
+
         function(c) {
           self.bDb.has(b.hash, function(err, val) {
             return c(err ||
@@ -107,8 +108,7 @@ function spec() {
           });
         },
         function(c) {
-        self.txDb.createFromBlock(b, function(err, addrs) {
+          self.txDb.createFromBlock(b, function(err) {
-          updatedAddrs = addrs;
             return c(err);
           });
         },
@@ -144,9 +144,9 @@ function spec() {
               return c(err);
             });
           });
-      }],
+        }
+      ],
       function(err) {
-        if (!err) self._handleBroadcast(b.hash, null, updatedAddrs);
         if (err && err.toString().match(/WARN/)) {
           err = null;
         }
@@ -162,6 +162,7 @@ function spec() {
     var orphanizeFrom;
 
     async.series([
+
         function(c) {
           self.bDb.isMain(newPrev, function(err, val) {
             if (!val) return c();
@@ -211,7 +212,9 @@ function spec() {
       hashInterator = fromHash;
 
     async.whilst(
-      function() { return hashInterator; },
+      function() {
+        return hashInterator;
+      },
       function(c) {
         self.setBlockMain(hashInterator, false, function(err) {
           if (err) return cb(err);
@@ -244,7 +247,9 @@ function spec() {
           });
         });
       },
-      function() { return hashInterator && !isMain; },
+      function() {
+        return hashInterator && !isMain;
+      },
       function(err) {
         console.log('\tFound yBlock:', hashInterator);
         return cb(err, hashInterator, lastHash);
@@ -252,79 +257,39 @@ function spec() {
     );
   };
 
-  Sync.prototype._handleBroadcast = function(hash, updatedTxs, updatedAddrs) {
+  Sync.prototype._handleBroadcast = function() {
     var self = this;
+    console.log('broadcast:' + self.opts.shouldBroadcast);
+  };
 
-    if (self.opts.shouldBroadcast) {
-      if (hash) {
-        sockets.broadcastBlock(hash);
-      }
 
-      if (updatedTxs) {
+  Sync.prototype.handleTxForAddress = function(data) {
-        updatedTxs.forEach(function(tx) {
+    if (this.opts.shouldBroadcast) {
-          sockets.broadcastTx(tx);
+      sockets.broadcastAddressTx(data.address, data.txid);
-        });
     }
+  };
 
-      if (updatedAddrs ) {
+  Sync.prototype.handleNewTx = function(data) {
-        updatedAddrs.forEach(function(addr, txs){
+    if (this.opts.shouldBroadcast) {
-          txs.forEach(function(addr, t){
+      sockets.broadcastTx(data.txid);
-            sockets.broadcastAddressTx(addr, t);
-          });
-        });
     }
+  };
+
+  Sync.prototype.handleNewBlock = function(data) {
+    if (this.opts.shouldBroadcast) {
+      sockets.broadcastBlock(data.blockid);
     }
   };
 
   Sync.prototype.storeTxs = function(txs, cb) {
     var self = this;
-
+    self.txDb.createFromArray(txs, null, function(err) {
-    self.txDb.createFromArray(txs, null, function(err, updatedAddrs) {
       if (err) return cb(err);
-
-      self._handleBroadcast(null, txs, updatedAddrs);
       return cb(err);
     });
   };
 
 
-  // TODO. replace with 
-  // Script.prototype.getAddrStrs if that one get merged in bitcore
-  Sync.prototype.getAddrStr = function(s) {
-    var self = this;
-
-    var addrStrs = [];
-    var type = s.classify();
-    var addr;
-
-    switch(type) {
-      case Script.TX_PUBKEY:
-        var chunk = s.captureOne();
-        addr = new Address(self.network.addressPubkey, bitutil.sha256ripe160(chunk));
-        addrStrs = [ addr.toString() ];
-        break;
-      case  Script.TX_PUBKEYHASH:
-        addr = new Address(self.network.addressPubkey, s.captureOne());
-        addrStrs = [ addr.toString() ];
-        break;
-      case Script.TX_SCRIPTHASH:
-        addr = new Address(self.network.addressScript, s.captureOne());
-        addrStrs = [ addr.toString() ];
-        break;
-      case Script.TX_MULTISIG:
-        var chunks = s.capture();
-        chunks.forEach(function(chunk) {
-          var a = new Address(self.network.addressPubkey,  bitutil.sha256ripe160(chunk));
-          addrStrs.push(a.toString());
-        });
-        break;
-      case Script.TX_UNKNOWN:
-        break;
-    }
-
-    return addrStrs;
-  };
   return Sync;
 }
 module.defineClass(spec);
-

lib/TransactionDb.js

@@ -5,6 +5,7 @@ require('classtool');
 
 function spec(b) {
 
+  var superclass = b.superclass || require('events').EventEmitter;
   // blockHash -> txid mapping 
   var IN_BLK_PREFIX = 'txb-'; //txb-<txid>-<block> => 1/0 (connected or not)
 
@@ -32,9 +33,27 @@ function spec(b) {
     config = require('../config/config'),
     assert = require('assert');
   var db = b.db || levelup(config.leveldb + '/txs');
+  var Script = require('bitcore/Script').class();
+  // This is 0.1.2 => c++ version of base57-native
+  var base58 = require('base58-native');
+  var encodedData = require('bitcore/util/EncodedData').class({
+    // TODO: check why c++ implementation differs
+    //base58: base58
+  });
+  var versionedData = require('bitcore/util/VersionedData').class({
+    superclass: encodedData
+  });
+  var Address = require('bitcore/Address').class({
+    superclass: versionedData
+  });
+  var bitutil = require('bitcore/util/util');
+  var networks = require('bitcore/networks');
 
   var TransactionDb = function() {
+    TransactionDb.super(this, arguments);
+    this.network = config.network === 'testnet' ? networks.testnet : networks.livenet;
   };
+  TransactionDb.superclass = superclass;
 
   TransactionDb.prototype.close = function(cb) {
     db.close(cb);
@@ -81,8 +100,7 @@ function spec(b) {
         txid: txid,
         index: parseInt(index),
       });
-    }
+    } else {
-    else {
       r.spendTxId = txid;
       r.spendIndex = parseInt(index);
       r.spendTs = parseInt(ts);
@@ -99,7 +117,10 @@ function spec(b) {
     var i = 0;
 
     // outs.
-    db.createReadStream({start: k, end: k + '~'})
+    db.createReadStream({
+      start: k,
+      end: k + '~'
+    })
       .on('data', function(data) {
         var k = data.key.split('-');
         var v = data.value.split(':');
@@ -116,7 +137,10 @@ function spec(b) {
       .on('end', function() {
 
         var k = SPEND_PREFIX + txid;
-        db.createReadStream({start: k, end: k + '~'})
+        db.createReadStream({
+          start: k,
+          end: k + '~'
+        })
           .on('data', function(data) {
             var k = data.key.split('-');
             var j = idx[parseInt(k[2])];
@@ -142,7 +166,10 @@ function spec(b) {
     if (!info) return cb();
 
     var k = SPEND_PREFIX + info.txid;
-    db.createReadStream({start: k, end: k + '~'})
+    db.createReadStream({
+      start: k,
+      end: k + '~'
+    })
       .on('data', function(data) {
         var k = data.key.split('-');
         self._addSpendInfo(info.vout[k[2]], k[3], k[4], data.value);
@@ -181,8 +208,7 @@ function spec(b) {
           i.value = ret.valueSat / util.COIN;
 
           // Double spend?
-        if ( ret.multipleSpendAttempt ||
+          if (ret.multipleSpendAttempt || !ret.spendTxId ||
-            !ret.spendTxId ||
             (ret.spendTxId && ret.spendTxId !== info.txid)
           ) {
             if (ret.multipleSpendAttempts) {
@@ -192,16 +218,13 @@ function spec(b) {
                   i.doubleSpendIndex = ret.spendIndex;
                 }
               });
-          }
+            } else if (!ret.spendTxId) {
-          else if (!ret.spendTxId) {
               i.dbError = 'Input spend not registered';
-          }
+            } else {
-          else {
               i.doubleSpendTxID = ret.spendTxId;
               i.doubleSpendIndex = ret.spendIndex;
             }
-        }
+          } else {
-        else {
             i.doubleSpendTxID = null;
           }
 
@@ -213,8 +236,7 @@ function spec(b) {
         if (!incompleteInputs) {
           info.valueIn = valueIn / util.COIN;
           info.fees = (valueIn - parseInt(info.valueOut * util.COIN)) / util.COIN;
-      }
+        } else {
-      else {
           info.incompleteInputs = 1;
         }
         return cb();
@@ -242,7 +264,10 @@ function spec(b) {
     self._getInfo(txid, function(err, info) {
       if (err) return cb(err);
       if (!info) return cb();
-      return cb(err, {txid: txid, info: info} );
+      return cb(err, {
+        txid: txid,
+        info: info
+      });
     });
   };
 
@@ -252,7 +277,9 @@ function spec(b) {
 
     db.get(k, function(err, val) {
       if (!val || (err && err.notFound)) {
-        return cb(null, { unconfirmedInput: 1} );
+        return cb(null, {
+          unconfirmedInput: 1
+        });
       }
 
       var a = val.split(':');
@@ -263,7 +290,10 @@ function spec(b) {
 
       // Spend?
       var k = SPEND_PREFIX + txid + '-' + n;
-      db.createReadStream({start: k, end: k + '~'})
+      db.createReadStream({
+        start: k,
+        end: k + '~'
+      })
         .on('data', function(data) {
           var k = data.key.split('-');
           self._addSpendInfo(ret, k[3], k[4], data.value);
@@ -300,8 +330,7 @@ function spec(b) {
               return e_c();
             });
           }, cb);
-      }
+      } else {
-      else {
         self.isConfirmed(o.spendTxId, function(err, is) {
           if (err) return cb(err);
           o.spendIsConfirmed = is;
@@ -317,7 +346,10 @@ function spec(b) {
     var k = ADDR_PREFIX + addr;
     var ret = [];
 
-    db.createReadStream({start: k, end: k + '~'})
+    db.createReadStream({
+      start: k,
+      end: k + '~'
+    })
       .on('data', function(data) {
         var k = data.key.split('-');
         var v = data.value.split(':');
@@ -335,7 +367,10 @@ function spec(b) {
 
         async.each(ret, function(o, e_c) {
             var k = SPEND_PREFIX + o.txid + '-' + o.index;
-          db.createReadStream({start: k, end: k + '~'})
+            db.createReadStream({
+              start: k,
+              end: k + '~'
+            })
               .on('data', function(data) {
                 var k = data.key.split('-');
                 self._addSpendInfo(o, k[3], k[4], data.value);
@@ -362,12 +397,15 @@ function spec(b) {
   TransactionDb.prototype.removeFromTxId = function(txid, cb) {
 
     async.series([
+
         function(c) {
           db.createReadStream({
             start: OUTS_PREFIX + txid,
             end: OUTS_PREFIX + txid + '~',
           }).pipe(
-            db.createWriteStream({type:'del'})
+            db.createWriteStream({
+              type: 'del'
+            })
           ).on('close', c);
         },
         function(c) {
@@ -376,9 +414,12 @@ function spec(b) {
             end: SPEND_PREFIX + txid + '~'
           })
             .pipe(
-            db.createWriteStream({type:'del'})
+              db.createWriteStream({
+                type: 'del'
+              })
           ).on('close', c);
-      }],
+        }
+      ],
       function(err) {
         cb(err);
       });
@@ -386,20 +427,70 @@ function spec(b) {
   };
 
 
+  // TODO. replace with 
+  // Script.prototype.getAddrStrs if that one get merged in bitcore
+  TransactionDb.prototype.getAddrStr = function(s) {
+    var self = this;
+
+    var addrStrs = [];
+    var type = s.classify();
+    var addr;
+
+    switch (type) {
+      case Script.TX_PUBKEY:
+        var chunk = s.captureOne();
+        addr = new Address(self.network.addressPubkey, bitutil.sha256ripe160(chunk));
+        addrStrs.push(addr.toString());
+        break;
+      case Script.TX_PUBKEYHASH:
+        addr = new Address(self.network.addressPubkey, s.captureOne());
+        addrStrs.push(addr.toString());
+        break;
+      case Script.TX_SCRIPTHASH:
+        addr = new Address(self.network.addressScript, s.captureOne());
+        addrStrs.push(addr.toString());
+        break;
+      case Script.TX_MULTISIG:
+        var chunks = s.capture();
+        chunks.forEach(function(chunk) {
+          var a = new Address(self.network.addressPubkey, bitutil.sha256ripe160(chunk));
+          addrStrs.push(a.toString());
+        });
+        break;
+      case Script.TX_UNKNOWN:
+        break;
+    }
+
+    return addrStrs;
+  };
 
   TransactionDb.prototype.adaptTxObject = function(txInfo) {
-
+    var self = this;
     // adapt bitcore TX object to bitcoind JSON response
     txInfo.txid = txInfo.hash;
 
+    
+    var to=0;
+    var tx = txInfo;
+    tx.outs.forEach( function(o) {
+      var s = new Script(o.s);
+      var addrs = self.getAddrStr(s);
+
+      // support only for p2pubkey p2pubkeyhash and p2sh
+      if (addrs.length === 1) {
+        tx.out[to].addrStr = addrs[0];
+        tx.out[to].n = to;
+      }
+      to++;
+    });
+
     var count = 0;
     txInfo.vin = txInfo.in.map(function(txin) {
       var i = {};
 
       if (txin.coinbase) {
         txInfo.isCoinBase = true;
-      }
+      } else {
-      else {
         i.txid = txin.prev_out.hash;
         i.vout = txin.prev_out.n;
       }
@@ -434,7 +525,7 @@ function spec(b) {
     var ts = tx.time;
 
     async.series([
-      // Input Outpoints (mark them as spended)
+      // Input Outpoints (mark them as spent)
       function(p_c) {
         if (tx.isCoinBase) return p_c();
         async.forEachLimit(tx.vin, CONCURRENCY,
@@ -454,17 +545,15 @@ function spec(b) {
           function(o, next_out) {
             if (o.value && o.scriptPubKey &&
               o.scriptPubKey.addresses &&
-              o.scriptPubKey.addresses[0] &&
+              o.scriptPubKey.addresses[0] && !o.scriptPubKey.addresses[1] // TODO : not supported
-              ! o.scriptPubKey.addresses[1] // TODO : not supported
             ) {
-              // This is only to broadcast (WIP)
-              if (addrs.indexOf(o.scriptPubKey.addresses[0]) === -1) {
-                addrs.push(o.scriptPubKey.addresses[0]);
-              }
-
               var addr = o.scriptPubKey.addresses[0];
               var sat = Math.round(o.value * util.COIN);
 
+              if (addrs.indexOf(addr) === -1) {
+                addrs.push(addr);
+              }
+
               // existed?
               var k = OUTS_PREFIX + tx.txid + '-' + o.n;
               db.get(k, function(err) {
@@ -473,14 +562,11 @@ function spec(b) {
                     .put(k, addr + ':' + sat)
                     .put(ADDR_PREFIX + addr + '-' + tx.txid + '-' + o.n, sat + ':' + ts)
                     .write(next_out);
-                }
+                } else {
-                else {
                   return next_out();
                 }
               });
-            }
+            } else {
-            else {
-              //console.log ('WARN in TX: %s could not parse OUTPUT %d', tx.txid, o.n);
               return next_out();
             }
           },
@@ -493,11 +579,26 @@ function spec(b) {
           });
       },
       function(p_c) {
-        if (!blockhash) return p_c();
+        if (!blockhash) {
+          return p_c();
+        }
         return self.setConfirmation(tx.txid, blockhash, true, p_c);
       },
     ], function(err) {
-        return cb(err, addrs);
+      if (addrs.length > 0 && !blockhash) {
+        // only emit if we are processing a single tx (not from a block)
+        addrs.forEach(function(addr) {
+          self.emit('tx_for_address', {
+            address: addr,
+            txid: tx.txid
+          });
+        });
+      }
+      self.emit('new_tx', {
+        txid: tx.txid
+      });
+
+      return cb(err);
     });
   };
 
@@ -520,7 +621,10 @@ function spec(b) {
     var k = IN_BLK_PREFIX + txId;
     var ret = false;
 
-    db.createReadStream({start: k, end: k + '~'})
+    db.createReadStream({
+      start: k,
+      end: k + '~'
+    })
       .on('data', function(data) {
         if (data.value === '1') ret = true;
       })
@@ -539,7 +643,10 @@ function spec(b) {
     var k = FROM_BLK_PREFIX + hash;
     var k2 = IN_BLK_PREFIX;
     // This is slow, but prevent us to create a new block->tx index.
-    db.createReadStream({start: k, end: k + '~'})
+    db.createReadStream({
+      start: k,
+      end: k + '~'
+    })
       .on('data', function(data) {
         var ks = data.key.split('-');
         toChange.push({
@@ -561,15 +668,10 @@ function spec(b) {
   // txs can be a [hashes] or [txObjects]
   TransactionDb.prototype.createFromArray = function(txs, blockHash, next) {
     var self = this;
-
     if (!txs) return next();
 
-    var updatedAddrs = []; // TODO
-
     async.forEachLimit(txs, CONCURRENCY, function(t, each_cb) {
         if (typeof t === 'string') {
-
-        // Is it from genesis block? (testnet==livenet)
           // TODO: parse it from networks.genesisTX?
           if (t === genesisTXID) return each_cb();
 
@@ -578,13 +680,12 @@ function spec(b) {
 
             return self.add(inInfo, blockHash, each_cb);
           });
-      }
+        } else {
-      else {
           return self.add(t, blockHash, each_cb);
         }
       },
       function(err) {
-      return next(err, updatedAddrs);
+        return next(err);
       });
   };