Browse Source

[ZMQ] append a message sequence number to every ZMQ notification

0.13
Jonas Schnelli 9 years ago
parent
commit
0b25a9fb42
No known key found for this signature in database
GPG Key ID: 29D4BCB6416F53EC
  1. 14
      contrib/zmq/zmq_sub.py
  2. 9
      doc/release-notes.md
  3. 5
      doc/zmq.md
  4. 13
      qa/rpc-tests/zmq_test.py
  5. 29
      src/zmq/zmqpublishnotifier.cpp
  6. 12
      src/zmq/zmqpublishnotifier.h

14
contrib/zmq/zmq_sub.py

@ -3,6 +3,7 @@ @@ -3,6 +3,7 @@
import array
import binascii
import zmq
import struct
port = 28332
@ -19,18 +20,21 @@ try: @@ -19,18 +20,21 @@ try:
msg = zmqSubSocket.recv_multipart()
topic = str(msg[0])
body = msg[1]
sequence = "Unknown";
if len(msg[-1]) == 4:
msgSequence = struct.unpack('<I', msg[-1])[-1]
sequence = str(msgSequence)
if topic == "hashblock":
print "- HASH BLOCK -"
print '- HASH BLOCK ('+sequence+') -'
print binascii.hexlify(body)
elif topic == "hashtx":
print '- HASH TX -'
print '- HASH TX ('+sequence+') -'
print binascii.hexlify(body)
elif topic == "rawblock":
print "- RAW BLOCK HEADER -"
print '- RAW BLOCK HEADER ('+sequence+') -'
print binascii.hexlify(body[:80])
elif topic == "rawtx":
print '- RAW TX -'
print '- RAW TX ('+sequence+') -'
print binascii.hexlify(body)
except KeyboardInterrupt:

9
doc/release-notes.md

@ -45,6 +45,15 @@ The following outputs are affected by this change: @@ -45,6 +45,15 @@ The following outputs are affected by this change:
- REST `/rest/block/` (JSON format when including extended tx details)
- `bitcoin-tx -json`
### ZMQ
Each ZMQ notification now contains an up-counting sequence number that allows
listeners to detect lost notifications.
The sequence number is always the last element in a multi-part ZMQ notification and
therefore backward compatible.
Each message type has its own counter.
(https://github.com/bitcoin/bitcoin/pull/7762)
### Configuration and command-line options
### Block and transaction handling

5
doc/zmq.md

@ -99,3 +99,8 @@ using other means such as firewalling. @@ -99,3 +99,8 @@ using other means such as firewalling.
Note that when the block chain tip changes, a reorganisation may occur
and just the tip will be notified. It is up to the subscriber to
retrieve the chain from the last known block to the new tip.
There are several possibilities that ZMQ notification can get lost
during transmission depending on the communication type your are
using. Bitcoind appends an up-counting sequence number to each
notification which allows listeners to detect lost notifications.

13
qa/rpc-tests/zmq_test.py

@ -11,6 +11,7 @@ from test_framework.test_framework import BitcoinTestFramework @@ -11,6 +11,7 @@ from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import *
import zmq
import binascii
import struct
try:
import http.client as httplib
@ -47,11 +48,17 @@ class ZMQTest (BitcoinTestFramework): @@ -47,11 +48,17 @@ class ZMQTest (BitcoinTestFramework):
print "listen..."
msg = self.zmqSubSocket.recv_multipart()
topic = msg[0]
assert_equal(topic, b"hashtx")
body = msg[1]
nseq = msg[2]
msgSequence = struct.unpack('<I', msg[-1])[-1]
assert_equal(msgSequence, 0) #must be sequence 0 on hashtx
msg = self.zmqSubSocket.recv_multipart()
topic = msg[0]
body = msg[1]
msgSequence = struct.unpack('<I', msg[-1])[-1]
assert_equal(msgSequence, 0) #must be sequence 0 on hashblock
blkhash = bytes_to_hex_str(body)
assert_equal(genhashes[0], blkhash) #blockhash from generate must be equal to the hash received over zmq
@ -61,12 +68,16 @@ class ZMQTest (BitcoinTestFramework): @@ -61,12 +68,16 @@ class ZMQTest (BitcoinTestFramework):
self.sync_all()
zmqHashes = []
blockcount = 0
for x in range(0,n*2):
msg = self.zmqSubSocket.recv_multipart()
topic = msg[0]
body = msg[1]
if topic == b"hashblock":
zmqHashes.append(bytes_to_hex_str(body))
msgSequence = struct.unpack('<I', msg[-1])[-1]
assert_equal(msgSequence, blockcount+1)
blockcount += 1
for x in range(0,n):
assert_equal(genhashes[x], zmqHashes[x]) #blockhash from generate must be equal to the hash received over zmq
@ -82,6 +93,8 @@ class ZMQTest (BitcoinTestFramework): @@ -82,6 +93,8 @@ class ZMQTest (BitcoinTestFramework):
hashZMQ = ""
if topic == b"hashtx":
hashZMQ = bytes_to_hex_str(body)
msgSequence = struct.unpack('<I', msg[-1])[-1]
assert_equal(msgSequence, blockcount+1)
assert_equal(hashRPC, hashZMQ) #blockhash from generate must be equal to the hash received over zmq

29
src/zmq/zmqpublishnotifier.cpp

@ -123,6 +123,23 @@ void CZMQAbstractPublishNotifier::Shutdown() @@ -123,6 +123,23 @@ void CZMQAbstractPublishNotifier::Shutdown()
psocket = 0;
}
bool CZMQAbstractPublishNotifier::SendMessage(const char *command, const void* data, size_t size)
{
assert(psocket);
/* send three parts, command & data & a LE 4byte sequence number */
unsigned char msgseq[sizeof(uint32_t)];
WriteLE32(&msgseq[0], nSequence);
int rc = zmq_send_multipart(psocket, command, strlen(command), data, size, msgseq, (size_t)sizeof(uint32_t), (void*)0);
if (rc == -1)
return false;
/* increment memory only sequence number after sending */
nSequence++;
return true;
}
bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
{
uint256 hash = pindex->GetBlockHash();
@ -130,8 +147,7 @@ bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex) @@ -130,8 +147,7 @@ bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
char data[32];
for (unsigned int i = 0; i < 32; i++)
data[31 - i] = hash.begin()[i];
int rc = zmq_send_multipart(psocket, MSG_HASHBLOCK, 9, data, 32, 0);
return rc == 0;
return SendMessage(MSG_HASHBLOCK, data, 32);
}
bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
@ -141,8 +157,7 @@ bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &t @@ -141,8 +157,7 @@ bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &t
char data[32];
for (unsigned int i = 0; i < 32; i++)
data[31 - i] = hash.begin()[i];
int rc = zmq_send_multipart(psocket, MSG_HASHTX, 6, data, 32, 0);
return rc == 0;
return SendMessage(MSG_HASHTX, data, 32);
}
bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
@ -163,8 +178,7 @@ bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex) @@ -163,8 +178,7 @@ bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
ss << block;
}
int rc = zmq_send_multipart(psocket, MSG_RAWBLOCK, 8, &(*ss.begin()), ss.size(), 0);
return rc == 0;
return SendMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size());
}
bool CZMQPublishRawTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
@ -173,6 +187,5 @@ bool CZMQPublishRawTransactionNotifier::NotifyTransaction(const CTransaction &tr @@ -173,6 +187,5 @@ bool CZMQPublishRawTransactionNotifier::NotifyTransaction(const CTransaction &tr
LogPrint("zmq", "zmq: Publish rawtx %s\n", hash.GetHex());
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION);
ss << transaction;
int rc = zmq_send_multipart(psocket, MSG_RAWTX, 5, &(*ss.begin()), ss.size(), 0);
return rc == 0;
return SendMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
}

12
src/zmq/zmqpublishnotifier.h

@ -11,7 +11,19 @@ class CBlockIndex; @@ -11,7 +11,19 @@ class CBlockIndex;
class CZMQAbstractPublishNotifier : public CZMQAbstractNotifier
{
private:
uint32_t nSequence; //! upcounting per message sequence number
public:
/* send zmq multipart message
parts:
* command
* data
* message sequence number
*/
bool SendMessage(const char *command, const void* data, size_t size);
bool Initialize(void *pcontext);
void Shutdown();
};

Loading…
Cancel
Save