diff --git a/src/zmq/zmqabstractnotifier.cpp b/src/zmq/zmqabstractnotifier.cpp index 7183b5215a..1fc6798df3 100644 --- a/src/zmq/zmqabstractnotifier.cpp +++ b/src/zmq/zmqabstractnotifier.cpp @@ -1,19 +1,20 @@ // Copyright (c) 2015 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 "zmqabstractnotifier.h" -#include "util.h" +#include + +#include CZMQAbstractNotifier::~CZMQAbstractNotifier() { assert(!psocket); } bool CZMQAbstractNotifier::NotifyBlock(const CBlockIndex * /*CBlockIndex*/) { return true; } bool CZMQAbstractNotifier::NotifyTransaction( const CTransaction & /*transaction*/) { return true; } diff --git a/src/zmq/zmqabstractnotifier.h b/src/zmq/zmqabstractnotifier.h index 425d4a198b..0b658bc040 100644 --- a/src/zmq/zmqabstractnotifier.h +++ b/src/zmq/zmqabstractnotifier.h @@ -1,41 +1,41 @@ // Copyright (c) 2015 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_ZMQABSTRACTNOTIFIER_H #define BITCOIN_ZMQ_ZMQABSTRACTNOTIFIER_H -#include "zmqconfig.h" +#include class CBlockIndex; class CZMQAbstractNotifier; typedef CZMQAbstractNotifier *(*CZMQNotifierFactory)(); class CZMQAbstractNotifier { public: CZMQAbstractNotifier() : psocket(nullptr) {} virtual ~CZMQAbstractNotifier(); template static CZMQAbstractNotifier *Create() { return new T(); } std::string GetType() const { return type; } void SetType(const std::string &t) { type = t; } std::string GetAddress() const { return address; } void SetAddress(const std::string &a) { address = a; } virtual bool Initialize(void *pcontext) = 0; virtual void Shutdown() = 0; virtual bool NotifyBlock(const CBlockIndex *pindex); virtual bool NotifyTransaction(const CTransaction &transaction); protected: void *psocket; std::string type; std::string address; }; #endif // BITCOIN_ZMQ_ZMQABSTRACTNOTIFIER_H diff --git a/src/zmq/zmqconfig.h b/src/zmq/zmqconfig.h index bc6a97708a..dc3f654dc5 100644 --- a/src/zmq/zmqconfig.h +++ b/src/zmq/zmqconfig.h @@ -1,23 +1,23 @@ // Copyright (c) 2014 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_ZMQCONFIG_H #define BITCOIN_ZMQ_ZMQCONFIG_H #if defined(HAVE_CONFIG_H) -#include "config/bitcoin-config.h" +#include #endif #include #if ENABLE_ZMQ #include #endif -#include "primitives/block.h" -#include "primitives/transaction.h" +#include +#include void zmqError(const char *str); #endif // BITCOIN_ZMQ_ZMQCONFIG_H diff --git a/src/zmq/zmqnotificationinterface.cpp b/src/zmq/zmqnotificationinterface.cpp index 73b1d92a7e..3ec52677c0 100644 --- a/src/zmq/zmqnotificationinterface.cpp +++ b/src/zmq/zmqnotificationinterface.cpp @@ -1,170 +1,170 @@ // 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 "zmqnotificationinterface.h" -#include "zmqpublishnotifier.h" +#include +#include -#include "streams.h" -#include "util.h" -#include "validation.h" -#include "version.h" +#include +#include +#include +#include void zmqError(const char *str) { LogPrint(BCLog::ZMQ, "zmq: Error: %s, errno=%s\n", str, zmq_strerror(errno)); } CZMQNotificationInterface::CZMQNotificationInterface() : pcontext(nullptr) {} CZMQNotificationInterface::~CZMQNotificationInterface() { Shutdown(); for (std::list::iterator i = notifiers.begin(); i != notifiers.end(); ++i) { delete *i; } } CZMQNotificationInterface *CZMQNotificationInterface::Create() { CZMQNotificationInterface *notificationInterface = nullptr; std::map factories; std::list notifiers; factories["pubhashblock"] = CZMQAbstractNotifier::Create; factories["pubhashtx"] = CZMQAbstractNotifier::Create; factories["pubrawblock"] = CZMQAbstractNotifier::Create; factories["pubrawtx"] = CZMQAbstractNotifier::Create; for (const auto &entry : factories) { std::string arg("-zmq" + entry.first); if (gArgs.IsArgSet(arg)) { CZMQNotifierFactory factory = entry.second; std::string address = gArgs.GetArg(arg, ""); CZMQAbstractNotifier *notifier = factory(); notifier->SetType(entry.first); notifier->SetAddress(address); notifiers.push_back(notifier); } } if (!notifiers.empty()) { notificationInterface = new CZMQNotificationInterface(); notificationInterface->notifiers = notifiers; if (!notificationInterface->Initialize()) { delete notificationInterface; notificationInterface = nullptr; } } return notificationInterface; } // Called at startup to conditionally set up ZMQ socket(s) bool CZMQNotificationInterface::Initialize() { LogPrint(BCLog::ZMQ, "zmq: Initialize notification interface\n"); assert(!pcontext); pcontext = zmq_init(1); if (!pcontext) { zmqError("Unable to initialize context"); return false; } std::list::iterator i = notifiers.begin(); for (; i != notifiers.end(); ++i) { CZMQAbstractNotifier *notifier = *i; if (notifier->Initialize(pcontext)) { LogPrint(BCLog::ZMQ, " Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress()); } else { LogPrint(BCLog::ZMQ, " Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress()); break; } } if (i != notifiers.end()) { return false; } return true; } // Called during shutdown sequence void CZMQNotificationInterface::Shutdown() { LogPrint(BCLog::ZMQ, "zmq: Shutdown notification interface\n"); if (pcontext) { for (std::list::iterator i = notifiers.begin(); i != notifiers.end(); ++i) { CZMQAbstractNotifier *notifier = *i; LogPrint(BCLog::ZMQ, " Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress()); notifier->Shutdown(); } zmq_ctx_destroy(pcontext); pcontext = nullptr; } } void CZMQNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) { // In IBD or blocks were disconnected without any new ones if (fInitialDownload || pindexNew == pindexFork) return; for (std::list::iterator i = notifiers.begin(); i != notifiers.end();) { CZMQAbstractNotifier *notifier = *i; if (notifier->NotifyBlock(pindexNew)) { i++; } else { notifier->Shutdown(); i = notifiers.erase(i); } } } void CZMQNotificationInterface::TransactionAddedToMempool( const CTransactionRef &ptx) { // Used by BlockConnected and BlockDisconnected as well, because they're all // the same external callback. const CTransaction &tx = *ptx; for (std::list::iterator i = notifiers.begin(); i != notifiers.end();) { CZMQAbstractNotifier *notifier = *i; if (notifier->NotifyTransaction(tx)) { i++; } else { notifier->Shutdown(); i = notifiers.erase(i); } } } void CZMQNotificationInterface::BlockConnected( const std::shared_ptr &pblock, const CBlockIndex *pindexConnected, const std::vector &vtxConflicted) { for (const CTransactionRef &ptx : pblock->vtx) { // Do a normal notify for each transaction added in the block TransactionAddedToMempool(ptx); } } void CZMQNotificationInterface::BlockDisconnected( const std::shared_ptr &pblock) { for (const CTransactionRef &ptx : pblock->vtx) { // Do a normal notify for each transaction removed in block // disconnection TransactionAddedToMempool(ptx); } } diff --git a/src/zmq/zmqnotificationinterface.h b/src/zmq/zmqnotificationinterface.h index b06df9bfea..ffd536292e 100644 --- a/src/zmq/zmqnotificationinterface.h +++ b/src/zmq/zmqnotificationinterface.h @@ -1,45 +1,45 @@ // 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_ZMQNOTIFICATIONINTERFACE_H #define BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H -#include "validationinterface.h" +#include #include #include class CBlockIndex; class CZMQAbstractNotifier; class CZMQNotificationInterface final : public CValidationInterface { public: virtual ~CZMQNotificationInterface(); static CZMQNotificationInterface *Create(); protected: bool Initialize(); void Shutdown(); // CValidationInterface void TransactionAddedToMempool(const CTransactionRef &tx) override; void BlockConnected(const std::shared_ptr &pblock, const CBlockIndex *pindexConnected, const std::vector &vtxConflicted) override; void BlockDisconnected(const std::shared_ptr &pblock) override; void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) override; private: CZMQNotificationInterface(); void *pcontext; std::list notifiers; }; #endif // BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H diff --git a/src/zmq/zmqpublishnotifier.cpp b/src/zmq/zmqpublishnotifier.cpp index 9847a867c5..cb8febf73e 100644 --- a/src/zmq/zmqpublishnotifier.cpp +++ b/src/zmq/zmqpublishnotifier.cpp @@ -1,190 +1,191 @@ // 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 "zmqpublishnotifier.h" -#include "chain.h" -#include "config.h" -#include "rpc/server.h" -#include "streams.h" -#include "util.h" -#include "validation.h" +#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"); 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); return -1; } zmq_msg_close(&msg); if (!data) { break; } size = va_arg(args, size_t); } 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; } int 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); psocket = i->second->psocket; mapPublishNotifiers.insert(std::make_pair(address, this)); return true; } } void CZMQAbstractPublishNotifier::Shutdown() { assert(psocket); 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, "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) { 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) { uint256 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); } bool CZMQPublishHashTransactionNotifier::NotifyTransaction( const CTransaction &transaction) { uint256 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); } 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)) { zmqError("Can't read block from disk"); return false; } ss << block; } return SendMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size()); } bool CZMQPublishRawTransactionNotifier::NotifyTransaction( const CTransaction &transaction) { uint256 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()); } diff --git a/src/zmq/zmqpublishnotifier.h b/src/zmq/zmqpublishnotifier.h index 2deed99f35..7a52d27cdc 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 "zmqabstractnotifier.h" +#include class CBlockIndex; class CZMQAbstractPublishNotifier : public CZMQAbstractNotifier { private: //!< upcounting per message sequence number uint32_t nSequence; 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) 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