Added block regtests.

lib/service.js

@@ -167,10 +167,12 @@ Service.prototype._loadTip = function(callback) {
 
     if (!tipBuf) {
       self.tip = self._getGenesisBlock();
+      self._lastHeight = 0;
       return callback();
     }
 
     self.tip = self._decodeTipData(tipData);
+    self._lastHeight = self.tip.height;
 
     log.info('Loaded tip for: ' + self.name + ' service, hash: ' +
       self.tip.hash + ' height: ' + self.tip.height);

lib/services/block/index.js

@@ -1,7 +1,6 @@
 'use strict';
 
 var BaseService = require('../../service');
-var levelup = require('levelup');
 var bitcore = require('bitcore-lib');
 var inherits = require('util').inherits;
 var Encoding = require('./encoding');
@@ -10,7 +9,6 @@ var log = index.log;
 var BufferUtil = bitcore.util.buffer;
 var Reorg = require('./reorg');
 var async = require('async');
-var BlockHandler = require('./block_handler');
 var LRU = require('lru-cache');
 var utils = require('../../utils');
 var _ = require('lodash');
@@ -22,17 +20,11 @@ var BlockService = function(options) {
   this.db = this.node.services.db;
   this.subscriptions = {};
   this.subscriptions.block = [];
-  this._blockHandler = new BlockHandler(this.node, this);
-  this._lockTimes = [];
   this.tip = null;
-  this.genesis = null;
-  this.dbOptions = {
-    keyEncoding: 'string',
-    valueEncoding: 'binary'
-  };
   this._blockHeaderQueue = LRU(50); //hash -> header, height -> header,
   this._blockQueue = LRU(10); // keep 10 blocks in the cache in case of reorg's
   this._lastReportedTime = Date.now();
+  this._lastHeight;
 };
 
 inherits(BlockService, BaseService);
@@ -96,13 +88,6 @@ BlockService.prototype.getPublishEvents = function() {
   ];
 };
 
-BlockService.prototype._sync = function() {
-
-  var self = this;
-  self._startSubscriptions();
-
-};
-
 BlockService.prototype.printTipInfo = function(prependedMessage) {
 
   log.info(
@@ -310,14 +295,15 @@ BlockService.prototype._startSubscriptions = function() {
 };
 
 BlockService.prototype._onHeaders = function(headers) {
-  log.debug('New header received: ' + block.hash);
+  log.info('New header received: ' + block.hash);
   this._cacheHeaders(headers);
 };
 
 BlockService.prototype._onBlock = function(block) {
 
-  log.debug('New block received: ' + block.hash);
+  log.info('New block received: ' + block.hash);
 
+  block.height =  ++this._lastHeight;
   this._cacheBlock(block);
   this._broadcast(this.subscriptions.block, 'block/block', block);
 };
@@ -333,6 +319,18 @@ BlockService.prototype._cacheBlock = function(block) {
   log.debug('Setting block: ' + block.hash + ' in the block cache.');
   this._blockQueue.set(block.hash, block);
 
+  var operations = this._getBlockOperations(block);
+
+  this.db.batch(operations, function(err) {
+
+    if(err) {
+      log.error('There was an error attempting to save block hash: ' + block.hash);
+      return;
+    }
+
+    log.debug('Success saving block hash ' + block.hash);
+  });
+
 };
 
 BlockService.prototype._getHeader = function(block) {
@@ -360,8 +358,7 @@ BlockService.prototype._setHandlers = function() {
   self.p2p.once('bestHeight', function(height) {
 
     self._bestHeight = height;
-    console.log(self._bestHeight);
+    self._loadTip(self._startSubscriptions.bind(self));
-    self._loadTip(self._sync.bind(self));
 
   });
 
@@ -434,14 +431,14 @@ BlockService.prototype._getBlockValue = function(hashOrHeight, callback) {
     valueFn = self.encoding.decodeBlockHeightValue.bind(self.encoding);
   }
 
+console.trace('hello');
   self.db.get(key, function(err, buf) {
-    if (err instanceof levelup.errors.NotFoundError) {
+
-      return callback();
-    }
     if (err) {
       return callback(err);
     }
     callback(null, valueFn(buf));
+
   });
 
 };

lib/services/p2p/index.js

@@ -237,7 +237,7 @@ P2P.prototype._setResourceFilter = function(filter, resource) {
   // startHash is usually the last block or header you have, endHash is a hash that comes after that hash,
   // or a block at a greater height
   if (resource === 'headers' || resource === 'blocks') {
-    assert(filter.startHash, 'A "startHash" field is required to retrieve headers or blocks');
+    assert(filter && filter.startHash, 'A "startHash" field is required to retrieve headers or blocks');
     if (!filter.endHash) {
       filter.endHash = 0;
     }

regtest/block.js

@@ -6,6 +6,8 @@ var async = require('async');
 var BitcoinRPC = require('bitcoind-rpc');
 var path = require('path');
 var Utils = require('./utils');
+var constants = require('../lib/constants');
+var zmq = require('zmq');
 
 var debug = true;
 var extraDebug = true;
@@ -60,7 +62,7 @@ var bitcore = {
           ]
         },
         'block-test': {
-          requirePath: path.resolve(__dirname + '/test_web.js')
+          requirePath: path.resolve(__dirname + '/test_bus.js')
         }
       }
     }
@@ -101,6 +103,38 @@ var opts = {
 
 var utils = new Utils(opts);
 
+var subSocket;
+var blocks = [];
+
+function processMessages(topic, message) {
+  var topicStr = topic.toString();
+  if (topicStr === 'block/block') {
+    return blocks.push(message);
+  }
+}
+
+function setupZmqSubscriber(callback) {
+
+  subSocket = zmq.socket('sub');
+  subSocket.on('connect', function(fd, endPoint) {
+    if (debug) {
+      console.log('ZMQ connected to:', endPoint);
+    }
+  });
+
+  subSocket.on('disconnect', function(fd, endPoint) {
+    if (debug) {
+      console.log('ZMQ disconnect:', endPoint);
+    }
+  });
+
+  subSocket.monitor(100, 0);
+  subSocket.connect('tcp://127.0.0.1:38332');
+  subSocket.subscribe('block');
+  subSocket.on('message', processMessages);
+  callback();
+}
+
 describe('Block Operations', function() {
 
   this.timeout(60000);
@@ -118,36 +152,56 @@ describe('Block Operations', function() {
         utils.startBitcoind.bind(utils),
         utils.waitForBitcoinReady.bind(utils),
         utils.startBitcoreNode.bind(utils),
-        utils.waitForBitcoreNode.bind(utils)
+        utils.waitForBitcoreNode.bind(utils),
+        setupZmqSubscriber
       ], done);
     });
 
-    it('should sync block hashes as keys and heights as values', function(done) {
+    it.only('should be able to get historical blocks from the network', function(done) {
-
+      var filter = { startHash: constants.BITCOIN_GENESIS_HASH.regtest };
-      async.timesLimit(opts.initialHeight, 12, function(n, next) {
+      utils.queryBitcoreNode(Object.assign({
-        utils.queryBitcoreNode(Object.assign({
+        path: '/test/p2p/blocks?filter=' + JSON.stringify(filter),
-          path: '/test/block/hash/' + n
+      }, bitcore.httpOpts), function(err) {
-        }, bitcore.httpOpts), function(err, res) {
-
-          if(err) {
-            return done(err);
-          }
-          res = JSON.parse(res);
-          expect(res.height).to.equal(n);
-          expect(res.hash.length).to.equal(64);
-          next(null, res.hash);
-        });
-      }, function(err, hashes) {
 
         if(err) {
           return done(err);
         }
-        self.hashes = hashes;
+
-        done();
+        setTimeout(function() {
+          expect(blocks.length).to.equal(150);
+          done();
+        }, 2000);
+
 
       });
     });
 
+    it('should sync block hashes as keys and heights as values', function(done) {
+
+      //async.timesLimit(opts.initialHeight, 12, function(n, next) {
+      //  utils.queryBitcoreNode(Object.assign({
+      //    path: '/test/block/hash/' + n
+      //  }, bitcore.httpOpts), function(err, res) {
+
+      //    if(err) {
+      //      return done(err);
+      //    }
+      //    res = JSON.parse(res);
+      //    expect(res.height).to.equal(n);
+      //    expect(res.hash.length).to.equal(64);
+      //    next(null, res.hash);
+      //  });
+      //}, function(err, hashes) {
+
+      //  if(err) {
+      //    return done(err);
+      //  }
+      //  self.hashes = hashes;
+      //  done();
+
+      //});
+    });
+
     it('should sync block heights as keys and hashes as values', function(done) {
       async.timesLimit(opts.initialHeight, 12, function(n, next) {
         utils.queryBitcoreNode(Object.assign({

regtest/test_bus.js

@@ -13,7 +13,7 @@ var TestBusService = function(options) {
 
 inherits(TestBusService, BaseService);
 
-TestBusService.dependencies = ['p2p', 'web'];
+TestBusService.dependencies = ['p2p', 'web', 'block', 'timestamp'];
 
 TestBusService.prototype.start = function(callback) {
 
@@ -45,7 +45,7 @@ TestBusService.prototype.start = function(callback) {
     if (self._ready) {
       while(self._cache.block.length > 0) {
         var blk = self._cache.block.shift();
-        self.pubSocket.send([ 'block', blk.toBuffer() ]);
+        self.pubSocket.send([ 'p2p/block', blk.toBuffer() ]);
       }
     }
   });
@@ -63,9 +63,20 @@ TestBusService.prototype.start = function(callback) {
     }
   });
 
+  self.bus.on('block/block', function(block) {
+    self._cache.block.push(block);
+    if (self._ready) {
+      while(self._cache.block.length > 0) {
+        var blk = self._cache.block.shift();
+        self.pubSocket.send([ 'block/block', blk.toBuffer() ]);
+      }
+    }
+  });
+
   self.bus.subscribe('p2p/transaction');
   self.bus.subscribe('p2p/block');
   self.bus.subscribe('p2p/headers');
+  self.bus.subscribe('block/block');
 
   self.node.on('ready', function() {
 
@@ -80,7 +91,7 @@ TestBusService.prototype.setupRoutes = function(app) {
 
   var self = this;
 
-  app.get('/mempool', function(req, res) {
+  app.get('/p2p/mempool', function(req, res) {
     self.node.services.p2p.clearInventoryCache();
     var filter;
     if (req.query.filter) {
@@ -90,7 +101,7 @@ TestBusService.prototype.setupRoutes = function(app) {
     res.status(200).end();
   });
 
-  app.get('/blocks', function(req, res) {
+  app.get('/p2p/blocks', function(req, res) {
     self.node.services.p2p.clearInventoryCache();
     var filter;
     if (req.query.filter) {
@@ -100,7 +111,7 @@ TestBusService.prototype.setupRoutes = function(app) {
     res.status(200).end();
   });
 
-  app.get('/headers', function(req, res) {
+  app.get('/p2p/headers', function(req, res) {
     var filter;
     if (req.query.filter) {
       filter = JSON.parse(req.query.filter);
@@ -112,6 +123,37 @@ TestBusService.prototype.setupRoutes = function(app) {
   app.get('/info', function(req, res) {
     res.status(200).jsonp({ result: (self._ready && (self._bestHeight >= 0))});
   });
+
+  app.get('/block/hash/:height', function(req, res) {
+    self.node.services.block.getBlockHash(req.params.height, function(err, hash) {
+      res.status(200).jsonp({ hash: hash, height: parseInt(req.params.height) });
+    });
+  });
+
+  app.get('/block/height/:hash', function(req, res) {
+    self.node.services.block.getBlockHeight(req.params.hash, function(err, height) {
+      res.status(200).jsonp({ hash: req.params.hash, height: height });
+    });
+  });
+
+  app.get('/timestamp/time/:hash', function(req, res) {
+    self.node.services.timestamp.getTimestamp(req.params.hash, function(err, timestamp) {
+      res.status(200).jsonp({ hash: req.params.hash, timestamp: timestamp });
+    });
+  });
+
+  app.get('/timestamp/hash/:time', function(req, res) {
+    self.node.services.timestamp.getHash(req.params.time, function(err, hash) {
+      res.status(200).jsonp({ hash: hash, timestamp: parseInt(req.params.time) });
+    });
+  });
+
+  app.get('/utxo/:address', function(req, res) {
+    self.node.services.utxo.getUtxosForAddress(req.params.address, function(err, utxos) {
+      res.status(200).jsonp({ utxos: utxos });
+    });
+  });
+
 };
 
 TestBusService.prototype.getRoutePrefix = function() {

regtest/test_web.js

@@ -55,7 +55,6 @@ TestWebService.prototype.setupRoutes = function(app) {
 
   app.get('/info', function(req, res) {
     var tip = self.node.services.block.tip;
-console.log(tip);
     if (tip) {
       return res.status(200).jsonp({ tip: JSON.stringify(tip) });
     }