src/zmq/zmqnotificationinterface.cpp

   1 // Copyright (c) 2015 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 "main.h"
  10 #include "streams.h"
  11 #include "util.h"
  12
  13 void zmqError(const char *str)
  14 {
  15     LogPrint("zmq", "Error: %s, errno=%s\n", str, zmq_strerror(errno));
  16 }
  17
  18 CZMQNotificationInterface::CZMQNotificationInterface() : pcontext(NULL)
  19 {
  20 }
  21
  22 CZMQNotificationInterface::~CZMQNotificationInterface()
  23 {
  24     // ensure Shutdown if Initialize is called
  25     assert(!pcontext);
  26
  27     for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i)
  28     {
  29         delete *i;
  30     }
  31 }
  32
  33 CZMQNotificationInterface* CZMQNotificationInterface::CreateWithArguments(const std::map<std::string, std::string> &args)
  34 {
  35     CZMQNotificationInterface* notificationInterface = NULL;
  36     std::map<std::string, CZMQNotifierFactory> factories;
  37     std::list<CZMQAbstractNotifier*> notifiers;
  38
  39     factories["pubhashblock"] = CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>;
  40     factories["pubhashtx"] = CZMQAbstractNotifier::Create<CZMQPublishHashTransactionNotifier>;
  41     factories["pubrawblock"] = CZMQAbstractNotifier::Create<CZMQPublishRawBlockNotifier>;
  42     factories["pubrawtx"] = CZMQAbstractNotifier::Create<CZMQPublishRawTransactionNotifier>;
  43
  44     for (std::map<std::string, CZMQNotifierFactory>::const_iterator i=factories.begin(); i!=factories.end(); ++i)
  45     {
  46         std::map<std::string, std::string>::const_iterator j = args.find("-zmq" + i->first);
  47         if (j!=args.end())
  48         {
  49             CZMQNotifierFactory factory = i->second;
  50             std::string address = j->second;
  51             CZMQAbstractNotifier *notifier = factory();
  52             notifier->SetType(i->first);
  53             notifier->SetAddress(address);
  54             notifiers.push_back(notifier);
  55         }
  56     }
  57
  58     if (!notifiers.empty())
  59     {
  60         notificationInterface = new CZMQNotificationInterface();
  61         notificationInterface->notifiers = notifiers;
  62     }
  63
  64     return notificationInterface;
  65 }
  66
  67 // Called at startup to conditionally set up ZMQ socket(s)
  68 bool CZMQNotificationInterface::Initialize()
  69 {
  70     LogPrint("zmq", "Initialize notification interface\n");
  71     assert(!pcontext);
  72
  73     pcontext = zmq_init(1);
  74
  75     if (!pcontext)
  76     {
  77         zmqError("Unable to initialize context");
  78         return false;
  79     }
  80
  81     std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin();
  82     for (; i!=notifiers.end(); ++i)
  83     {
  84         CZMQAbstractNotifier *notifier = *i;
  85         if (notifier->Initialize(pcontext))
  86         {
  87             LogPrint("zmq", "  Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
  88         }
  89         else
  90         {
  91             LogPrint("zmq", "  Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
  92             break;
  93         }
  94     }
  95
  96     if (i!=notifiers.end())
  97     {
  98         Shutdown();
  99         return false;
 100     }
 101
 102     return false;
 103 }
 104
 105 // Called during shutdown sequence
 106 void CZMQNotificationInterface::Shutdown()
 107 {
 108     LogPrint("zmq", "Shutdown notification interface\n");
 109     if (pcontext)
 110     {
 111         for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i)
 112         {
 113             CZMQAbstractNotifier *notifier = *i;
 114             LogPrint("zmq", "   Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress());
 115             notifier->Shutdown();
 116         }
 117         zmq_ctx_destroy(pcontext);
 118
 119         pcontext = 0;
 120     }
 121 }
 122
 123 void CZMQNotificationInterface::UpdatedBlockTip(const uint256 &hash)
 124 {
 125     for (std::list<CZMQAbstractNotifier*>::iterator i = notifiers.begin(); i!=notifiers.end(); )
 126     {
 127         CZMQAbstractNotifier *notifier = *i;
 128         if (notifier->NotifyBlock(hash))
 129         {
 130             i++;
 131         }
 132         else
 133         {
 134             notifier->Shutdown();
 135             i = notifiers.erase(i);
 136         }
 137     }
 138 }
 139
 140 void CZMQNotificationInterface::SyncTransaction(const CTransaction &tx, const CBlock *pblock)
 141 {
 142     for (std::list<CZMQAbstractNotifier*>::iterator i = notifiers.begin(); i!=notifiers.end(); )
 143     {
 144         CZMQAbstractNotifier *notifier = *i;
 145         if (notifier->NotifyTransaction(tx))
 146         {
 147             i++;
 148         }
 149         else
 150         {
 151             notifier->Shutdown();
 152             i = notifiers.erase(i);
 153         }
 154     }
 155 }