zmqnotificationinterface.cpp raw
1 // Copyright (c) 2015-2022 The Limenka 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 <limenka-build-config.h> // IWYU pragma: keep
6
7 #include <zmq/zmqnotificationinterface.h>
8
9 #include <common/args.h>
10 #include <kernel/chain.h>
11 #include <kernel/mempool_entry.h>
12 #include <logging.h>
13 #include <netbase.h>
14 #include <primitives/block.h>
15 #include <primitives/transaction.h>
16 #include <validationinterface.h>
17 #include <zmq/zmqabstractnotifier.h>
18 #include <zmq/zmqpublishnotifier.h>
19 #include <zmq/zmqutil.h>
20
21 #include <zmq.h>
22
23 #include <cassert>
24 #include <map>
25 #include <string>
26 #include <utility>
27 #include <vector>
28
29 #ifdef ENABLE_WALLET
30 #include <wallet/wallet.h>
31 #endif
32
33 CZMQNotificationInterface::CZMQNotificationInterface() = default;
34
35 CZMQNotificationInterface::~CZMQNotificationInterface()
36 {
37 Shutdown();
38 }
39
40 std::list<const CZMQAbstractNotifier*> CZMQNotificationInterface::GetActiveNotifiers() const
41 {
42 std::list<const CZMQAbstractNotifier*> result;
43 for (const auto& n : notifiers) {
44 result.push_back(n.get());
45 }
46 return result;
47 }
48
49 std::unique_ptr<CZMQNotificationInterface> CZMQNotificationInterface::Create(std::function<bool(std::vector<uint8_t>&, const CBlockIndex&)> get_block_by_index)
50 {
51 std::map<std::string, CZMQNotifierFactory> factories;
52 factories["pubhashblock"] = CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>;
53 factories["pubhashtx"] = CZMQAbstractNotifier::Create<CZMQPublishHashTransactionNotifier>;
54 factories["pubhashwallettx"] = CZMQAbstractNotifier::Create<CZMQPublishHashWalletTransactionNotifier>;
55 factories["pubrawblock"] = [&get_block_by_index]() -> std::unique_ptr<CZMQAbstractNotifier> {
56 return std::make_unique<CZMQPublishRawBlockNotifier>(get_block_by_index);
57 };
58 factories["pubrawtx"] = CZMQAbstractNotifier::Create<CZMQPublishRawTransactionNotifier>;
59 factories["pubrawwallettx"] = CZMQAbstractNotifier::Create<CZMQPublishRawWalletTransactionNotifier>;
60 factories["pubsequence"] = CZMQAbstractNotifier::Create<CZMQPublishSequenceNotifier>;
61
62 std::list<std::unique_ptr<CZMQAbstractNotifier>> notifiers;
63 for (const auto& entry : factories)
64 {
65 std::string arg("-zmq" + entry.first);
66 const auto& factory = entry.second;
67 for (std::string& address : gArgs.GetArgs(arg)) {
68 // libzmq uses prefix "ipc://" for UNIX domain sockets
69 if (address.substr(0, ADDR_PREFIX_UNIX.length()) == ADDR_PREFIX_UNIX) {
70 address.replace(0, ADDR_PREFIX_UNIX.length(), ADDR_PREFIX_IPC);
71 }
72
73 std::unique_ptr<CZMQAbstractNotifier> notifier = factory();
74 notifier->SetType(entry.first);
75 notifier->SetAddress(address);
76 notifier->SetOutboundMessageHighWaterMark(static_cast<int>(gArgs.GetIntArg(arg + "hwm", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM)));
77 notifiers.push_back(std::move(notifier));
78 }
79 }
80
81 if (!notifiers.empty())
82 {
83 std::unique_ptr<CZMQNotificationInterface> notificationInterface(new CZMQNotificationInterface());
84 notificationInterface->notifiers = std::move(notifiers);
85
86 if (notificationInterface->Initialize()) {
87 return notificationInterface;
88 }
89 }
90
91 return nullptr;
92 }
93
94 // Called at startup to conditionally set up ZMQ socket(s)
95 bool CZMQNotificationInterface::Initialize()
96 {
97 int major = 0, minor = 0, patch = 0;
98 zmq_version(&major, &minor, &patch);
99 LogDebug(BCLog::ZMQ, "version %d.%d.%d\n", major, minor, patch);
100
101 LogDebug(BCLog::ZMQ, "Initialize notification interface\n");
102 assert(!pcontext);
103
104 #ifdef ENABLE_WALLET
105 m_wtx_added_connection = wallet::CWallet::TransactionAddedToWallet.connect(std::bind(&CZMQNotificationInterface::TransactionAddedToWallet, this, std::placeholders::_1, std::placeholders::_2));
106 #endif
107
108 pcontext = zmq_ctx_new();
109
110 if (!pcontext)
111 {
112 zmqError("Unable to initialize context");
113 return false;
114 }
115
116 for (auto& notifier : notifiers) {
117 if (notifier->Initialize(pcontext)) {
118 LogDebug(BCLog::ZMQ, "Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
119 } else {
120 LogDebug(BCLog::ZMQ, "Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
121 return false;
122 }
123 }
124
125 return true;
126 }
127
128 // Called during shutdown sequence
129 void CZMQNotificationInterface::Shutdown()
130 {
131 LogDebug(BCLog::ZMQ, "Shutdown notification interface\n");
132
133 #ifdef ENABLE_WALLET
134 m_wtx_added_connection.disconnect();
135 #endif
136
137 if (pcontext)
138 {
139 for (auto& notifier : notifiers) {
140 LogDebug(BCLog::ZMQ, "Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress());
141 notifier->Shutdown();
142 }
143 zmq_ctx_term(pcontext);
144
145 pcontext = nullptr;
146 }
147 }
148
149 namespace {
150
151 template <typename Function>
152 void TryForEachAndRemoveFailed(std::list<std::unique_ptr<CZMQAbstractNotifier>>& notifiers, const Function& func)
153 {
154 for (auto i = notifiers.begin(); i != notifiers.end(); ++i) {
155 CZMQAbstractNotifier* notifier = i->get();
156 func(notifier);
157 }
158 }
159
160 } // anonymous namespace
161
162 void CZMQNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload)
163 {
164 if (fInitialDownload || pindexNew == pindexFork) // In IBD or blocks were disconnected without any new ones
165 return;
166
167 TryForEachAndRemoveFailed(notifiers, [pindexNew](CZMQAbstractNotifier* notifier) {
168 return notifier->NotifyBlock(pindexNew);
169 });
170 }
171
172 void CZMQNotificationInterface::TransactionAddedToMempool(const NewMempoolTransactionInfo& ptx, uint64_t mempool_sequence)
173 {
174 const CTransaction& tx = *(ptx.info.m_tx);
175
176 TryForEachAndRemoveFailed(notifiers, [&tx, mempool_sequence](CZMQAbstractNotifier* notifier) {
177 return notifier->NotifyTransaction(tx) && notifier->NotifyTransactionAcceptance(tx, mempool_sequence);
178 });
179 }
180
181 void CZMQNotificationInterface::TransactionRemovedFromMempool(const CTransactionRef& ptx, MemPoolRemovalReason reason, uint64_t mempool_sequence)
182 {
183 // Called for all non-block inclusion reasons
184 const CTransaction& tx = *ptx;
185
186 TryForEachAndRemoveFailed(notifiers, [&tx, mempool_sequence](CZMQAbstractNotifier* notifier) {
187 return notifier->NotifyTransactionRemoval(tx, mempool_sequence);
188 });
189 }
190
191 void CZMQNotificationInterface::BlockConnected(ChainstateRole role, const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindexConnected)
192 {
193 if (role == ChainstateRole::BACKGROUND) {
194 return;
195 }
196 for (const CTransactionRef& ptx : pblock->vtx) {
197 const CTransaction& tx = *ptx;
198 TryForEachAndRemoveFailed(notifiers, [&tx](CZMQAbstractNotifier* notifier) {
199 return notifier->NotifyTransaction(tx);
200 });
201 }
202
203 // Next we notify BlockConnect listeners for *all* blocks
204 TryForEachAndRemoveFailed(notifiers, [pindexConnected](CZMQAbstractNotifier* notifier) {
205 return notifier->NotifyBlockConnect(pindexConnected);
206 });
207 }
208
209 void CZMQNotificationInterface::BlockDisconnected(const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindexDisconnected)
210 {
211 for (const CTransactionRef& ptx : pblock->vtx) {
212 const CTransaction& tx = *ptx;
213 TryForEachAndRemoveFailed(notifiers, [&tx](CZMQAbstractNotifier* notifier) {
214 return notifier->NotifyTransaction(tx);
215 });
216 }
217
218 // Next we notify BlockDisconnect listeners for *all* blocks
219 TryForEachAndRemoveFailed(notifiers, [pindexDisconnected](CZMQAbstractNotifier* notifier) {
220 return notifier->NotifyBlockDisconnect(pindexDisconnected);
221 });
222 }
223
224 void CZMQNotificationInterface::TransactionAddedToWallet(const CTransactionRef& ptx, const uint256 &hashBlock) {
225 const CTransaction& tx = *ptx;
226
227 TryForEachAndRemoveFailed(notifiers, [&tx, &hashBlock](CZMQAbstractNotifier* notifier) {
228 return notifier->NotifyWalletTransaction(tx, hashBlock);
229 });
230 }
231
232 std::unique_ptr<CZMQNotificationInterface> g_zmq_notification_interface;
233