diff --git a/src/zmq/zmqnotificationinterface.cpp b/src/zmq/zmqnotificationinterface.cpp index 152bf6f9a..bfe41bbaf 100644 --- a/src/zmq/zmqnotificationinterface.cpp +++ b/src/zmq/zmqnotificationinterface.cpp @@ -1,189 +1,189 @@ // Copyright (c) 2015-2018 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 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; } } std::list CZMQNotificationInterface::GetActiveNotifiers() const { std::list result; for (const auto *n : notifiers) { result.push_back(n); } return result; } 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); notifier->SetOutboundMessageHighWaterMark( static_cast(gArgs.GetArg( arg + "hwm", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM))); 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() { int major = 0, minor = 0, patch = 0; zmq_version(&major, &minor, &patch); LogPrint(BCLog::ZMQ, "zmq: version %d.%d.%d\n", major, minor, patch); LogPrint(BCLog::ZMQ, "zmq: Initialize notification interface\n"); assert(!pcontext); pcontext = zmq_ctx_new(); 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, "zmq: Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress()); } else { LogPrint(BCLog::ZMQ, "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, "zmq: Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress()); notifier->Shutdown(); } zmq_ctx_term(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, const CBlockIndex *pindexDisconnected) { for (const CTransactionRef &ptx : pblock->vtx) { // Do a normal notify for each transaction removed in block // disconnection TransactionAddedToMempool(ptx); } } CZMQNotificationInterface *g_zmq_notification_interface = nullptr;