text: // Copyright (c) 2015-2018 The Worldcoin Core developers
  // Distributed under the MIT software license, see the accompanying
  // file COPYING or http://www.opensource.org/licenses/mit-license.php.
  
  #include <zmq/zmqnotificationinterface.h>
  #include <zmq/zmqpublishnotifier.h>
  
  #include <version.h>
  #include <validation.h>
  #include <streams.h>
  #include <util.h>
  
  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<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i)
      {
          delete *i;
      }
  }
  
  std::list<const CZMQAbstractNotifier*> CZMQNotificationInterface::GetActiveNotifiers() const
  {
      std::list<const CZMQAbstractNotifier*> result;
      for (const auto* n : notifiers) {
          result.push_back(n);
      }
      return result;
  }
  
  CZMQNotificationInterface* CZMQNotificationInterface::Create()
  {
      CZMQNotificationInterface* notificationInterface = nullptr;
      std::map<std::string, CZMQNotifierFactory> factories;
      std::list<CZMQAbstractNotifier*> notifiers;
  
      factories["pubhashblock"] = CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>;
      factories["pubhashtx"] = CZMQAbstractNotifier::Create<CZMQPublishHashTransactionNotifier>;
      factories["pubrawblock"] = CZMQAbstractNotifier::Create<CZMQPublishRawBlockNotifier>;
      factories["pubrawtx"] = CZMQAbstractNotifier::Create<CZMQPublishRawTransactionNotifier>;
  
      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<CZMQAbstractNotifier*>::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<CZMQAbstractNotifier*>::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)
  {
      if (fInitialDownload || pindexNew == pindexFork) // In IBD or blocks were disconnected without any new ones
          return;
  
      for (std::list<CZMQAbstractNotifier*>::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<CZMQAbstractNotifier*>::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<const CBlock>& pblock, const CBlockIndex* pindexConnected, const std::vector<CTransactionRef>& 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<const CBlock>& pblock)
  {
      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;
  