From de821d56e1f458fbe580520c77ac066107f4d77c Mon Sep 17 00:00:00 2001 From: Jonas Schnelli Date: Tue, 29 Mar 2016 11:34:25 +0200 Subject: [PATCH 1/2] [ZMQ] refactor message string --- src/zmq/zmqpublishnotifier.cpp | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/zmq/zmqpublishnotifier.cpp b/src/zmq/zmqpublishnotifier.cpp index f5839620f..a9ce9c48c 100644 --- a/src/zmq/zmqpublishnotifier.cpp +++ b/src/zmq/zmqpublishnotifier.cpp @@ -9,6 +9,11 @@ static std::multimap mapPublishNotifiers; +static const char *MSG_HASHBLOCK = "hashblock"; +static const char *MSG_HASHTX = "hashtx"; +static const char *MSG_RAWBLOCK = "rawblock"; +static const char *MSG_RAWTX = "rawtx"; + // Internal function to send multipart message static int zmq_send_multipart(void *sock, const void* data, size_t size, ...) { @@ -125,7 +130,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, "hashblock", 9, data, 32, 0); + int rc = zmq_send_multipart(psocket, MSG_HASHBLOCK, 9, data, 32, 0); return rc == 0; } @@ -136,7 +141,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, "hashtx", 6, data, 32, 0); + int rc = zmq_send_multipart(psocket, MSG_HASHTX, 6, data, 32, 0); return rc == 0; } @@ -158,7 +163,7 @@ bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex) ss << block; } - int rc = zmq_send_multipart(psocket, "rawblock", 8, &(*ss.begin()), ss.size(), 0); + int rc = zmq_send_multipart(psocket, MSG_RAWBLOCK, 8, &(*ss.begin()), ss.size(), 0); return rc == 0; } @@ -168,6 +173,6 @@ 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, "rawtx", 5, &(*ss.begin()), ss.size(), 0); + int rc = zmq_send_multipart(psocket, MSG_RAWTX, 5, &(*ss.begin()), ss.size(), 0); return rc == 0; } From 0b25a9fb42d5df54ea35ddb2bb4837e1e29355fd Mon Sep 17 00:00:00 2001 From: Jonas Schnelli Date: Tue, 29 Mar 2016 14:30:02 +0200 Subject: [PATCH 2/2] [ZMQ] append a message sequence number to every ZMQ notification --- contrib/zmq/zmq_sub.py | 14 +++++++++----- doc/release-notes.md | 9 +++++++++ doc/zmq.md | 5 +++++ qa/rpc-tests/zmq_test.py | 13 +++++++++++++ src/zmq/zmqpublishnotifier.cpp | 29 +++++++++++++++++++++-------- src/zmq/zmqpublishnotifier.h | 12 ++++++++++++ 6 files changed, 69 insertions(+), 13 deletions(-) diff --git a/contrib/zmq/zmq_sub.py b/contrib/zmq/zmq_sub.py index decf29d42..6268123dd 100755 --- a/contrib/zmq/zmq_sub.py +++ b/contrib/zmq/zmq_sub.py @@ -3,6 +3,7 @@ import array import binascii import zmq +import struct port = 28332 @@ -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('GetBlockHash(); @@ -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 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) 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 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()); } diff --git a/src/zmq/zmqpublishnotifier.h b/src/zmq/zmqpublishnotifier.h index 44d5cbea6..22f02a3d0 100644 --- a/src/zmq/zmqpublishnotifier.h +++ b/src/zmq/zmqpublishnotifier.h @@ -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(); };