aboutsummaryrefslogtreecommitdiff
path: root/src/zmq
diff options
context:
space:
mode:
Diffstat (limited to 'src/zmq')
-rw-r--r--src/zmq/zmqpublishnotifier.cpp29
-rw-r--r--src/zmq/zmqpublishnotifier.h12
2 files changed, 33 insertions, 8 deletions
diff --git a/src/zmq/zmqpublishnotifier.cpp b/src/zmq/zmqpublishnotifier.cpp
index a9ce9c48c4..b6c907980f 100644
--- a/src/zmq/zmqpublishnotifier.cpp
+++ b/src/zmq/zmqpublishnotifier.cpp
@@ -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)
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 44d5cbea67..22f02a3d0d 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();
};