Merge pull request #254 from kleetus/bug/tx_events_mempool

Tx event firing

.travis.yml

@@ -10,6 +10,7 @@ before_install:
 script:
   - _mocha -R spec integration/regtest.js
   - _mocha -R spec integration/regtest-node.js
+  - _mocha -R spec integration/p2p.js
   - _mocha -R spec --recursive
 cache:
   directories:

etc/bitcoin.patch

@@ -381,6 +381,18 @@ index c65e842..0e44bb5 100644
      CLevelDBWrapper(const boost::filesystem::path& path, size_t nCacheSize, bool fMemory = false, bool fWipe = false);
      ~CLevelDBWrapper();
 
+diff --git a/src/main.cpp b/src/main.cpp
+index 8f82abf..42bea1c 100644
+--- a/src/main.cpp
++++ b/src/main.cpp
+@@ -1058,6 +1058,7 @@ bool AcceptToMemoryPool(CTxMemPool& pool, CValidationState &state, const CTransa
+
+         // Store transaction in memory
+         pool.addUnchecked(hash, entry, !IsInitialBlockDownload());
++        GetNodeSignals().TxToMemPool(tx);
+     }
+
+     SyncWithWallets(tx, NULL);
 diff --git a/src/net.cpp b/src/net.cpp
 index 3908be6..cf3ffd4 100644
 --- a/src/net.cpp
@@ -395,3 +407,17 @@ index 3908be6..cf3ffd4 100644
 -                nLocalHostNonce, FormatSubVersion(CLIENT_NAME, CLIENT_VERSION, std::vector<string>()), nBestHeight, true);
 +                nLocalHostNonce, FormatSubVersion(CLIENT_NAME, CLIENT_VERSION, bitcore_node), nBestHeight, true);
  }
+
+
+diff --git a/src/net.h b/src/net.h
+index 17502b9..e181d68 100644
+--- a/src/net.h
++++ b/src/net.h
+@@ -99,6 +99,7 @@ struct CNodeSignals
+ {
+     boost::signals2::signal<int ()> GetHeight;
+     boost::signals2::signal<bool (CNode*), CombinerAll> ProcessMessages;
++    boost::signals2::signal<bool (const CTransaction&)> TxToMemPool;
+     boost::signals2::signal<bool (CNode*, bool), CombinerAll> SendMessages;
+     boost::signals2::signal<void (NodeId, const CNode*)> InitializeNode;
+     boost::signals2::signal<void (NodeId)> FinalizeNode;

src/libbitcoind.cc

@@ -81,10 +81,7 @@ static void
 async_get_tx_and_info_after(uv_work_t *req);
 
 static bool
-scan_messages(CNode* pfrom);
+queueTx(const CTransaction&);
-
-static bool
-scan_messages_after(CNode* pfrom);
 
 extern "C" void
 init(Handle<Object>);
@@ -93,7 +90,7 @@ init(Handle<Object>);
  * Private Global Variables
  * Used only by bitcoind functions.
  */
-static std::vector<CDataStream> txmon_messages;
+static std::vector<CTransaction> txQueue;
 static uv_async_t txmon_async;
 static Eternal<Function> txmon_callback;
 static bool txmon_callback_available;
@@ -278,8 +275,7 @@ NAN_METHOD(StartTxMon) {
   txmon_callback_available = true;
 
   CNodeSignals& nodeSignals = GetNodeSignals();
-  nodeSignals.ProcessMessages.connect(&scan_messages, boost::signals2::at_front);
+  nodeSignals.TxToMemPool.connect(&queueTx);
-  nodeSignals.ProcessMessages.connect(&scan_messages_after, boost::signals2::at_back);
 
   uv_async_init(uv_default_loop(), &txmon_async, tx_notifier);
 
@@ -291,133 +287,49 @@ tx_notifier(uv_async_t *handle) {
   Isolate* isolate = Isolate::GetCurrent();
   HandleScope scope(isolate);
 
-  {
+  Local<Array> results = Array::New(isolate);
+  int arrayIndex = 0;
 
-    LOCK(cs_main);
+  LOCK(cs_main);
+  BOOST_FOREACH(const CTransaction& tx, txQueue) {
 
-    Local<Array> results = Array::New(isolate);
+    CDataStream ssTx(SER_NETWORK, PROTOCOL_VERSION);
-    int arrayIndex = 0;
+    ssTx << tx;
+    std::string stx = ssTx.str();
+    Local<Value> txBuffer = node::Buffer::New(isolate, stx.c_str(), stx.size());
 
-    BOOST_FOREACH(CDataStream& vRecvCopy, txmon_messages) {
+    uint256 hash = tx.GetHash();
 
-      std::string vRecvStr = vRecvCopy.str();
+    Local<Object> obj = NanNew<Object>();
 
-      Local<Value> txBuffer = node::Buffer::New(isolate, vRecvStr.c_str(), vRecvStr.size());
+    obj->Set(NanNew<String>("buffer"), txBuffer);
-
+    obj->Set(NanNew<String>("hash"), NanNew<String>(hash.GetHex()));
-      CTransaction tx;
+    obj->Set(NanNew<String>("mempool"), NanNew<Boolean>(true));
-      vRecvCopy >> tx;
-      uint256 hash = tx.GetHash();
-
-      Local<Object> obj = NanNew<Object>();
-
-      bool existsInMempool = false;
-
-      CTransaction mtx;
-
-      if (mempool.lookup(hash, mtx))
-      {
-        existsInMempool = true;
-      }
-
-      obj->Set(NanNew<String>("buffer"), txBuffer);
-      obj->Set(NanNew<String>("hash"), NanNew<String>(hash.GetHex()));
-      obj->Set(NanNew<String>("mempool"), NanNew<Boolean>(existsInMempool));
-
-      results->Set(arrayIndex, obj);
-      arrayIndex++;
-    }
-
-    const unsigned argc = 1;
-    Local<Value> argv[argc] = {
-      Local<Value>::New(isolate, results)
-    };
-
-    Local<Function> cb = txmon_callback.Get(isolate);
-
-    cb->Call(isolate->GetCurrentContext()->Global(), argc, argv);
-
-    txmon_messages.clear();
 
+    results->Set(arrayIndex, obj);
+    arrayIndex++;
   }
 
+  const unsigned argc = 1;
+  Local<Value> argv[argc] = {
+    Local<Value>::New(isolate, results)
+  };
+
+  Local<Function> cb = txmon_callback.Get(isolate);
+
+  cb->Call(isolate->GetCurrentContext()->Global(), argc, argv);
+
+  txQueue.clear();
+
 }
-
 static bool
-scan_messages_after(CNode* pfrom) {
+queueTx(const CTransaction& tx) {
-  if(txmon_messages.size() > 0) {
+  LOCK(cs_main);
-    uv_async_send(&txmon_async);
+  txQueue.push_back(tx);
-  }
+  uv_async_send(&txmon_async);
   return true;
 }
 
-static bool
-scan_messages(CNode* pfrom) {
-
-  bool fOk = true;
-
-  std::deque<CNetMessage>::iterator it = pfrom->vRecvMsg.begin();
-  while (!pfrom->fDisconnect && it != pfrom->vRecvMsg.end()) {
-    // Don't bother if send buffer is too full to respond anyway
-    if (pfrom->nSendSize >= SendBufferSize()) {
-      break;
-    }
-
-    // get next message
-    CNetMessage& msg = *it;
-
-    // end, if an incomplete message is found
-    if (!msg.complete()) {
-      break;
-    }
-
-    // at this point, any failure means we can delete the current message
-    it++;
-
-    // Scan for message start
-    if (memcmp(msg.hdr.pchMessageStart, Params().MessageStart(), MESSAGE_START_SIZE) != 0) {
-      fOk = false;
-      break;
-    }
-
-    // Read header
-    CMessageHeader& hdr = msg.hdr;
-    if (!hdr.IsValid(Params().MessageStart())) {
-      continue;
-    }
-
-    std::string strCommand = hdr.GetCommand();
-
-    if (strCommand == (std::string)"tx") {
-
-      // Message size
-      unsigned int nMessageSize = hdr.nMessageSize;
-
-      // Checksum
-      CDataStream& vRecv = msg.vRecv;
-      uint256 hash = Hash(vRecv.begin(), vRecv.begin() + nMessageSize);
-      unsigned int nChecksum = 0;
-      memcpy(&nChecksum, &hash, sizeof(nChecksum));
-      if (nChecksum != hdr.nChecksum) {
-        continue;
-      }
-
-      // Copy the stream so that it can later be processed into the mempool
-      CDataStream vRecvCopy(vRecv.begin(), vRecv.end(), vRecv.GetType(), vRecv.GetVersion());
-
-      {
-        LOCK(cs_main);
-        txmon_messages.push_back(vRecvCopy);
-      }
-
-    }
-
-    boost::this_thread::interruption_point();
-    break;
-  }
-
-  return fOk;
-}
-
 /**
  * Functions
  */
@@ -1556,32 +1468,6 @@ NAN_METHOD(SendTransaction) {
   // Relay the transaction connect peers
   RelayTransaction(tx);
 
-  // Notify any listeners about the transaction
-  if(txmon_callback_available) {
-
-    Local<Array> results = Array::New(isolate);
-    Local<Object> obj = NanNew<Object>();
-
-    CDataStream ssTx(SER_NETWORK, PROTOCOL_VERSION);
-    ssTx << tx;
-    std::string stx = ssTx.str();
-    Local<Value> txBuffer = node::Buffer::New(isolate, stx.c_str(), stx.size());
-
-    obj->Set(NanNew<String>("buffer"), txBuffer);
-    obj->Set(NanNew<String>("hash"), NanNew<String>(hashTx.GetHex()));
-    obj->Set(NanNew<String>("mempool"), NanNew<Boolean>(true));
-
-    results->Set(0, obj);
-
-    const unsigned argc = 1;
-    Local<Value> argv[argc] = {
-      Local<Value>::New(isolate, results)
-    };
-    Local<Function> cb = txmon_callback.Get(isolate);
-
-    cb->Call(isolate->GetCurrentContext()->Global(), argc, argv);
-  }
-
   NanReturnValue(Local<Value>::New(isolate, NanNew<String>(hashTx.GetHex())));
 }