Start of mempool address index.

docs/services/bitcoind.md

@@ -13,7 +13,7 @@ The bitcoin service adds a native interface to Bitcoin Core for querying informa
 - `bitcoind.sendTransaction(transaction, allowAbsurdFees)` - Will attempt to add a transaction to the mempool and broadcast to peers.
 - `bitcoind.getTransaction(txid, queryMempool, callback)` - Get any tx asynchronously by reading it from disk, with an argument to optionally not include the mempool.
 - `bitcoind.getTransactionWithBlockInfo(txid, queryMempool, callback)` - Similar to getTransaction but will also include the block timestamp and height.
-- `bitcoind.getMempoolOutputs(address)` - Will return an array of outputs that match an address from the mempool.
+- `bitcoind.getMempoolTransactions()` - Will return an array of transaction buffers.
 - `bitcoind.getInfo()` - Basic information about the chain including total number of blocks.
 - `bitcoind.isSynced()` - Returns a boolean if the daemon is fully synced (not the initial block download)
 - `bitcoind.syncPercentage()` - Returns the current estimate of blockchain download as a percentage.

integration/regtest-node.js

@@ -661,5 +661,39 @@ describe('Node Functionality', function() {
 
       });
     });
+
+    describe('Mempool Index', function() {
+      var unspentOutput;
+      before(function(done) {
+        node.services.address.getUnspentOutputs(address, false, function(err, results) {
+          if (err) {
+            throw err;
+          }
+          results.length.should.equal(1);
+          unspentOutput = results[0];
+          done();
+        });
+      });
+
+      it('will update the mempool index after new tx', function(done) {
+
+        var tx = new Transaction();
+        tx.from(unspentOutput);
+        tx.to(address, unspentOutput.satoshis - 1000);
+        tx.fee(1000);
+        tx.sign(testKey);
+
+        node.services.bitcoind.sendTransaction(tx.serialize());
+
+        setImmediate(function() {
+          var length = node.services.address.mempoolIndex[address].length;
+          length.should.equal(1);
+          should.exist(node.services.address.mempoolIndex[address]);
+          done();
+        });
+
+      });
+
+    });
   });
 });

integration/regtest.js

@@ -300,7 +300,7 @@ describe('Daemon Binding Functionality', function() {
 
       var serialized = tx.serialize();
 
-      bitcoind.on('tx', function(result) {
+      bitcoind.once('tx', function(result) {
         result.buffer.toString('hex').should.equal(serialized);
         result.hash.should.equal(tx.hash);
         result.mempool.should.equal(true);
@@ -357,6 +357,18 @@ describe('Daemon Binding Functionality', function() {
     tx.change(changeAddress);
     tx.sign(privateKey1);
 
+    var tx2;
+    var tx2Key;
+
+    before(function() {
+      tx2 = bitcore.Transaction();
+      tx2.from(utxos[3]);
+      tx2.change(privateKey.toAddress());
+      tx2.to(destKey.toAddress(), utxos[3].amount * 1e8 - 1000);
+      tx2Key = bitcore.PrivateKey.fromWIF(utxos[3].privateKeyWIF);
+      tx2.sign(tx2Key);
+    });
+
     it('will add an unchecked transaction', function() {
       var added = bitcoind.addMempoolUncheckedTransaction(tx.serialize());
       added.should.equal(true);
@@ -370,18 +382,17 @@ describe('Daemon Binding Functionality', function() {
 
     });
 
-    it('get outputs by address', function() {
+    it('get one transaction', function() {
-      var outputs = bitcoind.getMempoolOutputs(changeAddress);
+      var transactions = bitcoind.getMempoolTransactions();
-      var expected = [
+      transactions[0].toString('hex').should.equal(tx.serialize());
-        {
+    });
-          address: 'mgBCJAsvzgT2qNNeXsoECg2uPKrUsZ76up',
+
-          script: '76a914073b7eae2823efa349e3b9155b8a735526463a0f88ac',
+    it('get multiple transactions', function() {
-          satoshis: 40000,
+      bitcoind.sendTransaction(tx2.serialize());
-          txid: tx.hash,
+      var transactions = bitcoind.getMempoolTransactions();
-          outputIndex: 1
+      var expected = [tx.serialize(), tx2.serialize()];
-        }
+      expected.should.contain(transactions[0].toString('hex'));
-      ];
+      expected.should.contain(transactions[1].toString('hex'));
-      outputs.should.deep.equal(expected);
     });
 
   });

lib/services/address/index.js

@@ -6,6 +6,7 @@ var async = require('async');
 var index = require('../../');
 var log = index.log;
 var errors = index.errors;
+var Transaction = require('../../transaction');
 var bitcore = require('bitcore');
 var $ = bitcore.util.preconditions;
 var _ = bitcore.deps._;
@@ -25,6 +26,8 @@ var AddressService = function(options) {
 
   this.node.services.bitcoind.on('tx', this.transactionHandler.bind(this));
 
+  this.mempoolIndex = {};
+
 };
 
 inherits(AddressService, BaseService);
@@ -128,11 +131,67 @@ AddressService.prototype.transactionHandler = function(txInfo) {
     this.transactionOutputHandler(messages, tx, i, !txInfo.mempool);
   }
 
+  // Update mempool index
+  if (txInfo.mempool) {
+    this.updateMempoolIndex(tx);
+  }
+
   for (var key in messages) {
     this.transactionEventHandler(messages[key]);
   }
 };
 
+AddressService.prototype.updateMempoolIndex = function(tx) {
+  var outputLength = tx.outputs.length;
+  for (var outputIndex = 0; outputIndex < outputLength; outputIndex++) {
+    var output = tx.outputs[outputIndex];
+    if (!output.script) {
+      continue;
+    }
+    var addressInfo = this._extractAddressInfoFromScript(output.script);
+    if (!addressInfo) {
+      continue;
+    }
+
+    var addressStr = bitcore.Address({
+      hashBuffer: addressInfo.hashBuffer,
+      type: addressInfo.addressType,
+      network: this.node.network
+    }).toString();
+
+    if (!this.mempoolIndex[addressStr]) {
+      this.mempoolIndex[addressStr] = [];
+    }
+
+    this.mempoolIndex[addressStr].push({
+      txid: tx.hash, // TODO use buffer
+      outputIndex: outputIndex,
+      satoshis: output.satoshis,
+      script: output._scriptBuffer.toString('hex') //TODO use a buffer
+    });
+
+  }
+
+};
+
+AddressService.prototype.resetMempoolIndex = function(callback) {
+  log.info('Starting reset of mempool index for address service.');
+  var self = this;
+  var transactionBuffers = self.node.services.bitcoind.getMempoolTransactions();
+  this.mempoolIndex = {};
+  async.each(transactionBuffers, function(txBuffer, next) {
+    var tx = Transaction.fromBuffer(txBuffer);
+    this.updateMempoolIndex(tx);
+    setImmediate(next);
+  }, function(err) {
+    if (err) {
+      return callback(err);
+    }
+    log.info('Mempool index update with address size:', Object.keys(self.mempoolIndex).length);
+    callback();
+  });
+};
+
 AddressService.prototype._extractAddressInfoFromScript = function(script) {
   var hashBuffer;
   var addressType;
@@ -635,11 +694,11 @@ AddressService.prototype.getOutputs = function(addressStr, options, callback) {
 
     var output = {
       address: addressStr,
-      txid: key.txid.toString('hex'),
+      txid: key.txid.toString('hex'), //TODO use a buffer
       outputIndex: key.outputIndex,
       height: key.height,
       satoshis: value.satoshis,
-      script: value.scriptBuffer.toString('hex'),
+      script: value.scriptBuffer.toString('hex'), //TODO use a buffer
       confirmations: self.node.services.db.tip.__height - key.height + 1
     };
 
@@ -661,7 +720,17 @@ AddressService.prototype.getOutputs = function(addressStr, options, callback) {
     }
 
     if(options.queryMempool) {
-      outputs = outputs.concat(self.node.services.bitcoind.getMempoolOutputs(addressStr));
+      var mempoolOutputs = self.mempoolIndex[addressStr];
+      if (mempoolOutputs) {
+        for(var i = 0; i < mempoolOutputs.length; i++) {
+          // TODO copy
+          var newOutput = mempoolOutputs[i];
+          newOutput.address = addressStr;
+          newOutput.height = -1;
+          newOutput.confirmations = 0;
+          outputs.push(newOutput);
+        }
+      }
     }
     callback(null, outputs);
   });

lib/services/bitcoind.js

@@ -1,12 +1,13 @@
 'use strict';
 
+var fs = require('fs');
 var util = require('util');
 var bindings = require('bindings')('bitcoind.node');
 var mkdirp = require('mkdirp');
-var fs = require('fs');
+var async = require('async');
 var bitcore = require('bitcore');
+var Transaction = require('../transaction');
 var $ = bitcore.util.preconditions;
-var _ = bitcore.deps._;
 var index = require('../');
 var log = index.log;
 var Service = require('../service');
@@ -215,8 +216,8 @@ Bitcoin.prototype.getTransactionWithBlockInfo = function(txid, queryMempool, cal
   return bindings.getTransactionWithBlockInfo(txid, queryMempool, callback);
 };
 
-Bitcoin.prototype.getMempoolOutputs = function(address) {
+Bitcoin.prototype.getMempoolTransactions = function() {
-  return bindings.getMempoolOutputs(address);
+  return bindings.getMempoolTransactions();
 };
 
 Bitcoin.prototype.addMempoolUncheckedTransaction = function(txBuffer) {

lib/services/db.js

@@ -408,6 +408,20 @@ DB.prototype.disconnectBlock = function(block, callback) {
   this.runAllBlockHandlers(block, false, callback);
 };
 
+DB.prototype.runAllMempoolIndexes = function(callback) {
+  async.eachSeries(
+    this.node.services,
+    function(service, next) {
+      if (service.resetMempoolIndex) {
+        service.resetMempoolIndex(next);
+      } else {
+        setImmediate(next);
+      }
+    },
+    callback
+  );
+};
+
 /**
  * Will collect all database operations for a block from other services
  * and save to the database.
@@ -670,15 +684,22 @@ DB.prototype.sync = function() {
       return self.node.emit('error', err);
     }
 
-    self.bitcoindSyncing = false;
-
     if(self.node.stopping) {
       return;
     }
 
-    // If bitcoind is completely synced
     if (self.node.services.bitcoind.isSynced()) {
-      self.node.emit('synced');
+      self.runAllMempoolIndexes(function(err) {
+        if (err) {
+          Error.captureStackTrace(err);
+          return self.node.emit('error', err);
+        }
+
+        self.bitcoindSyncing = false;
+        self.node.emit('synced');
+      });
+    } else {
+      self.bitcoindSyncing = false;
     }
 
   });

src/libbitcoind.cc

@@ -1531,84 +1531,38 @@ NAN_METHOD(SendTransaction) {
 }
 
 /**
- * GetMempoolOutputs
+ * GetMempoolTransactions
- * bitcoindjs.getMempoolOutputs()
+ * bitcoind.getMempoolTransactions()
- * Will return outputs by address from the mempool.
+ * Will return an array of transaction buffers.
  */
-NAN_METHOD(GetMempoolOutputs) {
+NAN_METHOD(GetMempoolTransactions) {
-  Isolate* isolate = Isolate::GetCurrent();
+  Isolate* isolate = args.GetIsolate();
   HandleScope scope(isolate);
 
-  // Instatiate an empty array that we will fill later
+  Local<Array> transactions = Array::New(isolate);
-  // with matching outputs.
-  Local<Array> outputs = Array::New(isolate);
   int arrayIndex = 0;
 
-  // Decode the input address into the hash bytes
+  {
-  // that we can then match to the scriptPubKeys data
+    LOCK(mempool.cs);
-  v8::String::Utf8Value param1(args[0]->ToString());
-  std::string *input = new std::string(*param1);
-  const char* psz = input->c_str();
-  std::vector<unsigned char> vAddress;
-  DecodeBase58(psz, vAddress);
-  vector<unsigned char> hashBytes(vAddress.begin()+1, vAddress.begin()+21);
 
-  // Iterate through the entire mempool
+    // Iterate through the entire mempool
-  std::map<uint256, CTxMemPoolEntry> mapTx = mempool.mapTx;
+    std::map<uint256, CTxMemPoolEntry> mapTx = mempool.mapTx;
 
-  for(std::map<uint256, CTxMemPoolEntry>::iterator it = mapTx.begin(); it != mapTx.end(); it++) {
+    for(std::map<uint256, CTxMemPoolEntry>::iterator it = mapTx.begin();
-
+        it != mapTx.end();
-    uint256 txid = it->first;
+        it++) {
-    CTxMemPoolEntry entry = it->second;
+      CTxMemPoolEntry entry = it->second;
-    const CTransaction tx = entry.GetTx();
+      const CTransaction tx = entry.GetTx();
-
+      CDataStream dataStreamTx(SER_NETWORK, PROTOCOL_VERSION);
-    int outputIndex = 0;
+      dataStreamTx << tx;
-
+      std::string txString = dataStreamTx.str();
-    // Iterate through each output
+      Local<Value> txBuffer = node::Buffer::New(isolate, txString.c_str(), txString.size());
-    BOOST_FOREACH(const CTxOut& txout, tx.vout) {
+      transactions->Set(arrayIndex, txBuffer);
-
+      arrayIndex++;
-      CScript script = txout.scriptPubKey;
-
-      txnouttype type;
-      std::vector<std::vector<unsigned char> > hashResults;
-
-      if (Solver(script, type, hashResults)) {
-
-        // See if the script is any of the standard address types
-        if (type == TX_PUBKEYHASH || type == TX_SCRIPTHASH) {
-
-          vector<unsigned char> scripthashBytes = hashResults.front();
-
-          // Compare the hash bytes with the input hash bytes
-          if(equal(hashBytes.begin(), hashBytes.end(), scripthashBytes.begin())) {
-
-            Local<Object> output = NanNew<Object>();
-
-            output->Set(NanNew<String>("address"), NanNew<String>(psz));
-
-            std::string scriptHex = HexStr(script.begin(), script.end());
-            output->Set(NanNew<String>("script"), NanNew<String>(scriptHex));
-
-            uint64_t satoshis = txout.nValue;
-            output->Set(NanNew<String>("satoshis"), NanNew<Number>(satoshis)); // can't go above 2 ^ 53 -1
-            output->Set(NanNew<String>("txid"), NanNew<String>(txid.GetHex()));
-
-            output->Set(NanNew<String>("outputIndex"), NanNew<Number>(outputIndex));
-
-            // We have a match and push the results to the array
-            // that is returned as the result
-            outputs->Set(arrayIndex, output);
-            arrayIndex++;
-
-          }
-        }
-      }
-
-      outputIndex++;
     }
   }
 
-  NanReturnValue(outputs);
+  NanReturnValue(transactions);
 
 }
 
@@ -1668,7 +1622,7 @@ init(Handle<Object> target) {
   NODE_SET_METHOD(target, "isSpent", IsSpent);
   NODE_SET_METHOD(target, "getBlockIndex", GetBlockIndex);
   NODE_SET_METHOD(target, "isMainChain", IsMainChain);
-  NODE_SET_METHOD(target, "getMempoolOutputs", GetMempoolOutputs);
+  NODE_SET_METHOD(target, "getMempoolTransactions", GetMempoolTransactions);
   NODE_SET_METHOD(target, "addMempoolUncheckedTransaction", AddMempoolUncheckedTransaction);
   NODE_SET_METHOD(target, "sendTransaction", SendTransaction);
   NODE_SET_METHOD(target, "estimateFee", EstimateFee);

src/libbitcoind.h

@@ -29,7 +29,7 @@ NAN_METHOD(GetTransaction);
 NAN_METHOD(GetInfo);
 NAN_METHOD(IsSpent);
 NAN_METHOD(GetBlockIndex);
-NAN_METHOD(GetMempoolOutputs);
+NAN_METHOD(GetMempoolTransactions);
 NAN_METHOD(AddMempoolUncheckedTransaction);
 NAN_METHOD(SendTransaction);
 NAN_METHOD(EstimateFee);