src/zmq/zmqnotificationinterface.cpp

   1 // Copyright (c) 2015-2016 The Bitcoin Core developers
   2 // Distributed under the MIT software license, see the accompanying
   3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
   4
   5 #include "zmqnotificationinterface.h"
   6 #include "zmqpublishnotifier.h"
   7
   8 #include "version.h"
   9 #include "validation.h"
  10 #include "streams.h"
  11 #include "util.h"
  12
  13 void zmqError(const char *str)
  14 {
  15     LogPrint(BCLog::ZMQ, "zmq: Error: %s, errno=%s\n", str, zmq_strerror(errno));
  16 }
  17
  18 CZMQNotificationInterface::CZMQNotificationInterface() : pcontext(nullptr)
  19 {
  20 }
  21
  22 CZMQNotificationInterface::~CZMQNotificationInterface()
  23 {
  24     Shutdown();
  25
  26     for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i)
  27     {
  28         delete *i;
  29     }
  30 }
  31
  32 CZMQNotificationInterface* CZMQNotificationInterface::Create()
  33 {
  34     CZMQNotificationInterface* notificationInterface = nullptr;
  35     std::map<std::string, CZMQNotifierFactory> factories;
  36     std::list<CZMQAbstractNotifier*> notifiers;
  37
  38     factories["pubhashblock"] = CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>;
  39     factories["pubhashtx"] = CZMQAbstractNotifier::Create<CZMQPublishHashTransactionNotifier>;
  40     factories["pubrawblock"] = CZMQAbstractNotifier::Create<CZMQPublishRawBlockNotifier>;
  41     factories["pubrawtx"] = CZMQAbstractNotifier::Create<CZMQPublishRawTransactionNotifier>;
  42
  43     for (std::map<std::string, CZMQNotifierFactory>::const_iterator i=factories.begin(); i!=factories.end(); ++i)
  44     {
  45         std::string arg("-zmq" + i->first);
  46         if (gArgs.IsArgSet(arg))
  47         {
  48             CZMQNotifierFactory factory = i->second;
  49             std::string address = gArgs.GetArg(arg, "");
  50             CZMQAbstractNotifier *notifier = factory();
  51             notifier->SetType(i->first);
  52             notifier->SetAddress(address);
  53             notifiers.push_back(notifier);
  54         }
  55     }
  56
  57     if (!notifiers.empty())
  58     {
  59         notificationInterface = new CZMQNotificationInterface();
  60         notificationInterface->notifiers = notifiers;
  61
  62         if (!notificationInterface->Initialize())
  63         {
  64             delete notificationInterface;
  65             notificationInterface = nullptr;
  66         }
  67     }
  68
  69     return notificationInterface;
  70 }
  71
  72 // Called at startup to conditionally set up ZMQ socket(s)
  73 bool CZMQNotificationInterface::Initialize()
  74 {
  75     LogPrint(BCLog::ZMQ, "zmq: Initialize notification interface\n");
  76     assert(!pcontext);
  77
  78     pcontext = zmq_init(1);
  79
  80     if (!pcontext)
  81     {
  82         zmqError("Unable to initialize context");
  83         return false;
  84     }
  85
  86     std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin();
  87     for (; i!=notifiers.end(); ++i)
  88     {
  89         CZMQAbstractNotifier *notifier = *i;
  90         if (notifier->Initialize(pcontext))
  91         {
  92             LogPrint(BCLog::ZMQ, "  Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
  93         }
  94         else
  95         {
  96             LogPrint(BCLog::ZMQ, "  Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
  97             break;
  98         }
  99     }
 100
 101     if (i!=notifiers.end())
 102     {
 103         return false;
 104     }
 105
 106     return true;
 107 }
 108
 109 // Called during shutdown sequence
 110 void CZMQNotificationInterface::Shutdown()
 111 {
 112     LogPrint(BCLog::ZMQ, "zmq: Shutdown notification interface\n");
 113     if (pcontext)
 114     {
 115         for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i)
 116         {
 117             CZMQAbstractNotifier *notifier = *i;
 118             LogPrint(BCLog::ZMQ, "   Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress());
 119             notifier->Shutdown();
 120         }
 121         zmq_ctx_destroy(pcontext);
 122
 123         pcontext = nullptr;
 124     }
 125 }
 126
 127 void CZMQNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload)
 128 {
 129     if (fInitialDownload || pindexNew == pindexFork) // In IBD or blocks were disconnected without any new ones
 130         return;
 131
 132     for (std::list<CZMQAbstractNotifier*>::iterator i = notifiers.begin(); i!=notifiers.end(); )
 133     {
 134         CZMQAbstractNotifier *notifier = *i;
 135         if (notifier->NotifyBlock(pindexNew))
 136         {
 137             i++;
 138         }
 139         else
 140         {
 141             notifier->Shutdown();
 142             i = notifiers.erase(i);
 143         }
 144     }
 145 }
 146
 147 void CZMQNotificationInterface::TransactionAddedToMempool(const CTransactionRef& ptx)
 148 {
 149     // Used by BlockConnected and BlockDisconnected as well, because they're
 150     // all the same external callback.
 151     const CTransaction& tx = *ptx;
 152
 153     for (std::list<CZMQAbstractNotifier*>::iterator i = notifiers.begin(); i!=notifiers.end(); )
 154     {
 155         CZMQAbstractNotifier *notifier = *i;
 156         if (notifier->NotifyTransaction(tx))
 157         {
 158             i++;
 159         }
 160         else
 161         {
 162             notifier->Shutdown();
 163             i = notifiers.erase(i);
 164         }
 165     }
 166 }
 167
 168 void CZMQNotificationInterface::BlockConnected(const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindexConnected, const std::vector<CTransactionRef>& vtxConflicted)
 169 {
 170     for (const CTransactionRef& ptx : pblock->vtx) {
 171         // Do a normal notify for each transaction added in the block
 172         TransactionAddedToMempool(ptx);
 173     }
 174 }
 175
 176 void CZMQNotificationInterface::BlockDisconnected(const std::shared_ptr<const CBlock>& pblock)
 177 {
 178     for (const CTransactionRef& ptx : pblock->vtx) {
 179         // Do a normal notify for each transaction removed in block disconnection
 180         TransactionAddedToMempool(ptx);
 181     }
 182 }