http: add chain watching and scanning.

lib/bcoin/http/server.js

@@ -189,7 +189,7 @@ HTTPServer.prototype._init = function _init() {
       assert(typeof params.hash === 'string', 'Hash must be a string.');
       if (params.hash.length !== 64) {
         options.height = Number(params.hash);
-        assert(utils.isUInt32(options.height), 'Hash must be a number.');
+        assert(utils.isUInt32(options.height), 'Height must be a number.');
       } else {
         options.hash = utils.revHex(params.hash);
       }
@@ -247,8 +247,6 @@ HTTPServer.prototype._init = function _init() {
       if (typeof params.subtractFee === 'number') {
         options.subtractFee = params.subtractFee;
         assert(utils.isUInt32(options.subtractFee), 'subtractFee must be a number.');
-      } else if (params.subtractFee === 'true') {
-        options.subtractFee = true;
       } else {
         assert(typeof options.subtractFee === 'boolean', 'subtractFee must be a boolean.');
         options.subtractFee = params.subtractFee;
@@ -917,6 +915,10 @@ HTTPServer.prototype._initIO = function _initIO() {
       self.emit('error', err);
     });
 
+    socket.on('disconnect', function() {
+      socket.destroy();
+    });
+
     socket.once('auth', function(apiKey, callback) {
       if (typeof callback !== 'function')
         return socket.destroy();
@@ -990,6 +992,50 @@ HTTPServer.prototype._initIO = function _initIO() {
 
       return callback();
     });
+
+    socket.on('watch chain', function() {
+      socket.watchChain();
+    });
+
+    socket.on('unwatch chain', function() {
+      socket.unwatchChain();
+    });
+
+    socket.on('watch address', function(addresses) {
+      if (!Array.isArray(addresses))
+        return socket.destroy();
+
+      try {
+        socket.addFilter(addresses);
+      } catch (e) {
+        return socket.destroy();
+      }
+    });
+
+    socket.on('unwatch address', function(addresses) {
+      if (!Array.isArray(addresses))
+        return socket.destroy();
+
+      try {
+        socket.removeFilter(addresses);
+      } catch (e) {
+        return socket.destroy();
+      }
+    });
+
+    socket.on('scan chain', function(start, callback) {
+      if (typeof callback !== 'function')
+        return socket.destroy();
+
+      if (typeof start !== 'string')
+        return callback({ error: 'Invalid parameter.' });
+
+      try {
+        socket.scan(start, callback);
+      } catch (e) {
+        return callback({ error: e.message });
+      }
+    });
   });
 
   this.walletdb.on('tx', function(id, tx, info) {
@@ -1138,6 +1184,20 @@ function ClientSocket(server, socket) {
   this.host = socket.conn.remoteAddress;
   this.timeout = null;
   this.auth = false;
+  this.filter = {};
+  this.filterCount = 0;
+
+  this.node = this.server.node;
+  this.network = this.server.network;
+  this.chain = this.server.chain;
+  this.mempool = this.server.mempool;
+  this.pool = this.server.pool;
+  this.fees = this.server.fees;
+  this.miner = this.server.miner;
+  this.wallet = this.server.wallet;
+  this.walletdb = this.server.walletdb;
+  this.logger = this.server.logger;
+  this.events = [];
 
   this._init();
 }
@@ -1149,12 +1209,162 @@ ClientSocket.prototype._init = function _init() {
   var socket = this.socket;
   var emit = EventEmitter.prototype.emit;
   var onevent = socket.onevent.bind(socket);
+
   socket.onevent = function(packet) {
     var result = onevent(packet);
     var args = packet.data || [];
     emit.apply(self, args);
     return result;
   };
+
+  socket.on('error', function(err) {
+    emit.call(self, 'error', err);
+  });
+
+  socket.on('disconnect', function() {
+    emit.call(self, 'disconnect');
+  });
+};
+
+ClientSocket.prototype.addFilter = function addFilter(addresses) {
+  var i, hash;
+
+  for (i = 0; i < addresses.length; i++) {
+    hash = bcoin.address.getHash(addresses[i], 'hex');
+
+    if (!hash)
+      throw new Error('Bad address.');
+
+    if (!this.filter[hash]) {
+      this.filter[hash] = true;
+      this.filterCount++;
+    }
+  }
+};
+
+ClientSocket.prototype.removeFilter = function removeFilter(addresses) {
+  var i, hash;
+
+  for (i = 0; i < addresses.length; i++) {
+    hash = bcoin.address.getHash(addresses[i], 'hex');
+
+    if (!hash)
+      throw new Error('Bad address.');
+
+    if (this.filter[hash]) {
+      delete this.filter[hash];
+      this.filterCount--;
+    }
+  }
+};
+
+ClientSocket.prototype.bind = function bind(obj, event, listener) {
+  this.events.push([obj, event, listener]);
+  obj.on(event, listener);
+};
+
+ClientSocket.prototype.unbind = function unbind(obj, event) {
+  var i, event;
+
+  for (i = this.events.length - 1; i >= 0; i--) {
+    event = this.events[i]
+    if (event[0] === obj && event[1] === event) {
+      obj.removeListener(event, event[2]);
+      this.events.splice(i, 1);
+    }
+  }
+};
+
+ClientSocket.prototype.unbindAll = function unbindAll() {
+  var i, event;
+
+  for (i = 0; i < this.events.length; i++) {
+    event = this.events[i];
+    event[0].removeListener(event[1], event[2]);
+  }
+
+  this.events.length = 0;
+};
+
+ClientSocket.prototype.watchChain = function watchChain() {
+  var self = this;
+
+  this.bind(this.chain, 'connect', function(entry, block) {
+    var txs;
+
+    self.emit('block connect', entry.toJSON());
+
+    txs = self.testBlock(block);
+
+    if (txs)
+      self.emit('block tx', txs, entry.toJSON());
+  });
+
+  this.bind(this.chain, 'disconnect', function(entry, block) {
+    self.emit('block disconnect', entry.toJSON());
+  });
+
+  this.bind(this.mempool, 'tx', function(tx) {
+    if (self.testFilter(tx))
+      self.emit('mempool tx', tx.toJSON());
+  });
+};
+
+ClientSocket.prototype.unwatchChain = function unwatchChain() {
+  this.unbind(this.chain, 'connect');
+  this.unbind(this.chain, 'disconnect');
+  this.unbind(this.mempool, 'tx');
+};
+
+ClientSocket.prototype.testBlock = function testBlock(tx) {
+  var txs = [];
+  var i, tx;
+
+  if (this.filterCount === 0)
+    return;
+
+  for (i = 0; i < block.txs.length; i++) {
+    tx = block.txs[i];
+    if (this.testFilter(tx))
+      txs.push(tx.toJSON());
+  }
+
+  if (txs.length === 0)
+    return;
+
+  return txs;
+};
+
+ClientSocket.prototype.testFilter = function testFilter(tx) {
+  var i, hashes, hash;
+
+  if (this.filterCount === 0)
+    return;
+
+  hashes = tx.getHashes('hex');
+
+  for (i = 0; i < hashes.length; i++) {
+    hash = hashes[i];
+    if (this.filter[hash])
+      return true;
+  }
+};
+
+ClientSocket.prototype.scan = function scan(start, callback) {
+  var self = this;
+  var hash;
+
+  hashes = Object.keys(this.filter);
+  start = utils.revHex(start);
+
+  this.chain.db.scan(start, hashes, function(tx, entry, next) {
+    self.emit('block tx', [tx.toJSON()], entry.toJSON());
+    next();
+  }, function(err) {
+    if (err)
+      return callback({ error: err.message });
+    callback();
+  });
 };
 
 ClientSocket.prototype.join = function join(id) {
@@ -1186,6 +1396,7 @@ ClientSocket.prototype.stop = function stop() {
 };
 
 ClientSocket.prototype.destroy = function() {
+  this.unbindAll();
   this.stop();
   this.socket.disconnect();
 };