diff --git a/src/zmq/zmqpublishnotifier.cpp b/src/zmq/zmqpublishnotifier.cpp index d90ecc6c0..c2fe8172b 100644 --- a/src/zmq/zmqpublishnotifier.cpp +++ b/src/zmq/zmqpublishnotifier.cpp @@ -1,224 +1,225 @@ // Copyright (c) 2015-2016 The Bitcoin Core developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include 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, ...) { va_list args; va_start(args, size); while (1) { zmq_msg_t msg; int rc = zmq_msg_init_size(&msg, size); if (rc != 0) { zmqError("Unable to initialize ZMQ msg"); va_end(args); return -1; } void *buf = zmq_msg_data(&msg); memcpy(buf, data, size); data = va_arg(args, const void *); rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0); if (rc == -1) { zmqError("Unable to send ZMQ msg"); zmq_msg_close(&msg); va_end(args); return -1; } zmq_msg_close(&msg); if (!data) { break; } size = va_arg(args, size_t); } va_end(args); return 0; } bool CZMQAbstractPublishNotifier::Initialize(void *pcontext) { assert(!psocket); // check if address is being used by other publish notifier std::multimap::iterator i = mapPublishNotifiers.find(address); if (i == mapPublishNotifiers.end()) { psocket = zmq_socket(pcontext, ZMQ_PUB); if (!psocket) { zmqError("Failed to create socket"); return false; } LogPrint(BCLog::ZMQ, "zmq: Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark); int rc = zmq_setsockopt(psocket, ZMQ_SNDHWM, &outbound_message_high_water_mark, sizeof(outbound_message_high_water_mark)); if (rc != 0) { zmqError("Failed to set outbound message high water mark"); zmq_close(psocket); return false; } rc = zmq_bind(psocket, address.c_str()); if (rc != 0) { zmqError("Failed to bind address"); zmq_close(psocket); return false; } // register this notifier for the address, so it can be reused for other // publish notifier mapPublishNotifiers.insert(std::make_pair(address, this)); return true; } else { LogPrint(BCLog::ZMQ, "zmq: Reusing socket for address %s\n", address); LogPrint(BCLog::ZMQ, "zmq: Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark); psocket = i->second->psocket; mapPublishNotifiers.insert(std::make_pair(address, this)); return true; } } void CZMQAbstractPublishNotifier::Shutdown() { // Early return if Initialize was not called if (!psocket) { return; } int count = mapPublishNotifiers.count(address); // remove this notifier from the list of publishers using this address typedef std::multimap::iterator iterator; std::pair iterpair = mapPublishNotifiers.equal_range(address); for (iterator it = iterpair.first; it != iterpair.second; ++it) { if (it->second == this) { mapPublishNotifiers.erase(it); break; } } if (count == 1) { LogPrint(BCLog::ZMQ, "zmq: Close socket at address %s\n", address); int linger = 0; zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger)); zmq_close(psocket); } psocket = nullptr; } -bool CZMQAbstractPublishNotifier::SendMessage(const char *command, - const void *data, size_t size) { +bool CZMQAbstractPublishNotifier::SendZmqMessage(const char *command, + const void *data, + size_t size) { assert(psocket); /* send three parts, command & data & a LE 4byte sequence number */ uint8_t 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), nullptr); if (rc == -1) { return false; } /* increment memory only sequence number after sending */ nSequence++; return true; } bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex) { BlockHash hash = pindex->GetBlockHash(); LogPrint(BCLog::ZMQ, "zmq: Publish hashblock %s\n", hash.GetHex()); char data[32]; for (unsigned int i = 0; i < 32; i++) { data[31 - i] = hash.begin()[i]; } - return SendMessage(MSG_HASHBLOCK, data, 32); + return SendZmqMessage(MSG_HASHBLOCK, data, 32); } bool CZMQPublishHashTransactionNotifier::NotifyTransaction( const CTransaction &transaction) { TxId txid = transaction.GetId(); LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s\n", txid.GetHex()); char data[32]; for (unsigned int i = 0; i < 32; i++) { data[31 - i] = txid.begin()[i]; } - return SendMessage(MSG_HASHTX, data, 32); + return SendZmqMessage(MSG_HASHTX, data, 32); } bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex) { LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s\n", pindex->GetBlockHash().GetHex()); const Config &config = GetConfig(); CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags()); { LOCK(cs_main); CBlock block; if (!ReadBlockFromDisk(block, pindex, config.GetChainParams().GetConsensus())) { zmqError("Can't read block from disk"); return false; } ss << block; } - return SendMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size()); + return SendZmqMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size()); } bool CZMQPublishRawTransactionNotifier::NotifyTransaction( const CTransaction &transaction) { TxId txid = transaction.GetId(); LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s\n", txid.GetHex()); CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags()); ss << transaction; - return SendMessage(MSG_RAWTX, &(*ss.begin()), ss.size()); + return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size()); } diff --git a/src/zmq/zmqpublishnotifier.h b/src/zmq/zmqpublishnotifier.h index 4b6c98389..e9091ad4b 100644 --- a/src/zmq/zmqpublishnotifier.h +++ b/src/zmq/zmqpublishnotifier.h @@ -1,50 +1,50 @@ // Copyright (c) 2015-2016 The Bitcoin Core developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. #ifndef BITCOIN_ZMQ_ZMQPUBLISHNOTIFIER_H #define BITCOIN_ZMQ_ZMQPUBLISHNOTIFIER_H #include class CBlockIndex; class CZMQAbstractPublishNotifier : public CZMQAbstractNotifier { private: //! upcounting per message sequence number uint32_t nSequence{0U}; public: /* send zmq multipart message parts: * command * data * message sequence number */ - bool SendMessage(const char *command, const void *data, size_t size); + bool SendZmqMessage(const char *command, const void *data, size_t size); bool Initialize(void *pcontext) override; void Shutdown() override; }; class CZMQPublishHashBlockNotifier : public CZMQAbstractPublishNotifier { public: bool NotifyBlock(const CBlockIndex *pindex) override; }; class CZMQPublishHashTransactionNotifier : public CZMQAbstractPublishNotifier { public: bool NotifyTransaction(const CTransaction &transaction) override; }; class CZMQPublishRawBlockNotifier : public CZMQAbstractPublishNotifier { public: bool NotifyBlock(const CBlockIndex *pindex) override; }; class CZMQPublishRawTransactionNotifier : public CZMQAbstractPublishNotifier { public: bool NotifyTransaction(const CTransaction &transaction) override; }; #endif // BITCOIN_ZMQ_ZMQPUBLISHNOTIFIER_H