Bitcoin ABC  0.24.7
P2P Digital Currency
zmqpublishnotifier.cpp
Go to the documentation of this file.
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 
6 
7 #include <blockdb.h>
8 #include <chain.h>
9 #include <chainparams.h>
10 #include <config.h>
11 #include <primitives/blockhash.h>
12 #include <primitives/txid.h>
13 #include <rpc/server.h>
14 #include <streams.h>
15 #include <util/system.h>
16 #include <zmq/zmqutil.h>
17 
18 #include <zmq.h>
19 
20 #include <cstdarg>
21 #include <cstddef>
22 #include <map>
23 #include <string>
24 #include <utility>
25 
26 static std::multimap<std::string, CZMQAbstractPublishNotifier *>
28 
29 static const char *MSG_HASHBLOCK = "hashblock";
30 static const char *MSG_HASHTX = "hashtx";
31 static const char *MSG_RAWBLOCK = "rawblock";
32 static const char *MSG_RAWTX = "rawtx";
33 static const char *MSG_SEQUENCE = "sequence";
34 
35 // Internal function to send multipart message
36 static int zmq_send_multipart(void *sock, const void *data, size_t size, ...) {
37  va_list args;
38  va_start(args, size);
39 
40  while (1) {
41  zmq_msg_t msg;
42 
43  int rc = zmq_msg_init_size(&msg, size);
44  if (rc != 0) {
45  zmqError("Unable to initialize ZMQ msg");
46  va_end(args);
47  return -1;
48  }
49 
50  void *buf = zmq_msg_data(&msg);
51  memcpy(buf, data, size);
52 
53  data = va_arg(args, const void *);
54 
55  rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0);
56  if (rc == -1) {
57  zmqError("Unable to send ZMQ msg");
58  zmq_msg_close(&msg);
59  va_end(args);
60  return -1;
61  }
62 
63  zmq_msg_close(&msg);
64 
65  if (!data) {
66  break;
67  }
68 
69  size = va_arg(args, size_t);
70  }
71  va_end(args);
72  return 0;
73 }
74 
76  assert(!psocket);
77 
78  // check if address is being used by other publish notifier
79  std::multimap<std::string, CZMQAbstractPublishNotifier *>::iterator i =
81 
82  if (i == mapPublishNotifiers.end()) {
83  psocket = zmq_socket(pcontext, ZMQ_PUB);
84  if (!psocket) {
85  zmqError("Failed to create socket");
86  return false;
87  }
88 
90  "zmq: Outbound message high water mark for %s at %s is %d\n",
92 
93  int rc = zmq_setsockopt(psocket, ZMQ_SNDHWM,
96  if (rc != 0) {
97  zmqError("Failed to set outbound message high water mark");
98  zmq_close(psocket);
99  return false;
100  }
101 
102  const int so_keepalive_option{1};
103  rc = zmq_setsockopt(psocket, ZMQ_TCP_KEEPALIVE, &so_keepalive_option,
104  sizeof(so_keepalive_option));
105  if (rc != 0) {
106  zmqError("Failed to set SO_KEEPALIVE");
107  zmq_close(psocket);
108  return false;
109  }
110 
111  rc = zmq_bind(psocket, address.c_str());
112  if (rc != 0) {
113  zmqError("Failed to bind address");
114  zmq_close(psocket);
115  return false;
116  }
117 
118  // register this notifier for the address, so it can be reused for other
119  // publish notifier
120  mapPublishNotifiers.insert(std::make_pair(address, this));
121  return true;
122  } else {
123  LogPrint(BCLog::ZMQ, "zmq: Reusing socket for address %s\n", address);
125  "zmq: Outbound message high water mark for %s at %s is %d\n",
127 
128  psocket = i->second->psocket;
129  mapPublishNotifiers.insert(std::make_pair(address, this));
130 
131  return true;
132  }
133 }
134 
136  // Early return if Initialize was not called
137  if (!psocket) {
138  return;
139  }
140 
141  int count = mapPublishNotifiers.count(address);
142 
143  // remove this notifier from the list of publishers using this address
144  typedef std::multimap<std::string, CZMQAbstractPublishNotifier *>::iterator
145  iterator;
146  std::pair<iterator, iterator> iterpair =
147  mapPublishNotifiers.equal_range(address);
148 
149  for (iterator it = iterpair.first; it != iterpair.second; ++it) {
150  if (it->second == this) {
151  mapPublishNotifiers.erase(it);
152  break;
153  }
154  }
155 
156  if (count == 1) {
157  LogPrint(BCLog::ZMQ, "zmq: Close socket at address %s\n", address);
158  int linger = 0;
159  zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger));
160  zmq_close(psocket);
161  }
162 
163  psocket = nullptr;
164 }
165 
167  const void *data,
168  size_t size) {
169  assert(psocket);
170 
171  /* send three parts, command & data & a LE 4byte sequence number */
172  uint8_t msgseq[sizeof(uint32_t)];
173  WriteLE32(&msgseq[0], nSequence);
174  int rc = zmq_send_multipart(psocket, command, strlen(command), data, size,
175  msgseq, (size_t)sizeof(uint32_t), nullptr);
176  if (rc == -1) {
177  return false;
178  }
179 
180  /* increment memory only sequence number after sending */
181  nSequence++;
182 
183  return true;
184 }
185 
187  BlockHash hash = pindex->GetBlockHash();
188  LogPrint(BCLog::ZMQ, "zmq: Publish hashblock %s to %s\n", hash.GetHex(),
189  this->address);
190  char data[32];
191  for (unsigned int i = 0; i < 32; i++) {
192  data[31 - i] = hash.begin()[i];
193  }
194  return SendZmqMessage(MSG_HASHBLOCK, data, 32);
195 }
196 
198  const CTransaction &transaction) {
199  TxId txid = transaction.GetId();
200  LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s to %s\n", txid.GetHex(),
201  this->address);
202  char data[32];
203  for (unsigned int i = 0; i < 32; i++) {
204  data[31 - i] = txid.begin()[i];
205  }
206  return SendZmqMessage(MSG_HASHTX, data, 32);
207 }
208 
210  LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s to %s\n",
211  pindex->GetBlockHash().GetHex(), this->address);
212 
213  const Config &config = GetConfig();
215  {
216  LOCK(cs_main);
217  CBlock block;
218  if (!ReadBlockFromDisk(block, pindex,
219  config.GetChainParams().GetConsensus())) {
220  zmqError("Can't read block from disk");
221  return false;
222  }
223 
224  ss << block;
225  }
226 
227  return SendZmqMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size());
228 }
229 
231  const CTransaction &transaction) {
232  TxId txid = transaction.GetId();
233  LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s to %s\n", txid.GetHex(),
234  this->address);
236  ss << transaction;
237  return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
238 }
239 
240 // TODO: Dedup this code to take label char, log string
242  const CBlockIndex *pindex) {
243  BlockHash hash = pindex->GetBlockHash();
244  LogPrint(BCLog::ZMQ, "zmq: Publish sequence block connect %s to %s\n",
245  hash.GetHex(), this->address);
246  char data[sizeof(BlockHash) + 1];
247  for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
248  data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
249  }
250  // Block (C)onnect
251  data[sizeof(data) - 1] = 'C';
252  return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
253 }
254 
256  const CBlockIndex *pindex) {
257  BlockHash hash = pindex->GetBlockHash();
258  LogPrint(BCLog::ZMQ, "zmq: Publish sequence block disconnect %s to %s\n",
259  hash.GetHex(), this->address);
260  char data[sizeof(BlockHash) + 1];
261  for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
262  data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
263  }
264  // Block (D)isconnect
265  data[sizeof(data) - 1] = 'D';
266  return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
267 }
268 
270  const CTransaction &transaction, uint64_t mempool_sequence) {
271  TxId txid = transaction.GetId();
272  LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool acceptance %s to %s\n",
273  txid.GetHex(), this->address);
274  uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
275  for (unsigned int i = 0; i < sizeof(TxId); i++) {
276  data[sizeof(TxId) - 1 - i] = txid.begin()[i];
277  }
278  // Mempool (A)cceptance
279  data[sizeof(TxId)] = 'A';
280  WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
281  return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
282 }
283 
285  const CTransaction &transaction, uint64_t mempool_sequence) {
286  TxId txid = transaction.GetId();
287  LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool removal %s to %s\n",
288  txid.GetHex(), this->address);
289  uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
290  for (unsigned int i = 0; i < sizeof(TxId); i++) {
291  data[sizeof(TxId) - 1 - i] = txid.begin()[i];
292  }
293  // Mempool (R)emoval
294  data[sizeof(TxId)] = 'R';
295  WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
296  return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
297 }
CZMQPublishSequenceNotifier::NotifyBlockDisconnect
bool NotifyBlockDisconnect(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:255
MSG_RAWBLOCK
static const char * MSG_RAWBLOCK
Definition: zmqpublishnotifier.cpp:31
count
static int count
Definition: tests.c:41
BCLog::ZMQ
@ ZMQ
Definition: logging.h:43
CZMQPublishSequenceNotifier::NotifyBlockConnect
bool NotifyBlockConnect(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:241
WriteLE32
static void WriteLE32(uint8_t *ptr, uint32_t x)
Definition: common.h:40
streams.h
CDataStream::begin
const_iterator begin() const
Definition: streams.h:276
blockhash.h
CChainParams::GetConsensus
const Consensus::Params & GetConsensus() const
Definition: chainparams.h:59
CZMQAbstractNotifier::type
std::string type
Definition: zmqabstractnotifier.h:64
zmqError
void zmqError(const char *str)
Definition: zmqutil.cpp:11
chainparams.h
mapPublishNotifiers
static std::multimap< std::string, CZMQAbstractPublishNotifier * > mapPublishNotifiers
Definition: zmqpublishnotifier.cpp:27
CZMQAbstractPublishNotifier::Shutdown
void Shutdown() override
Definition: zmqpublishnotifier.cpp:135
CBlockIndex::GetBlockHash
BlockHash GetBlockHash() const
Definition: blockindex.h:133
CTransaction
The basic transaction that is broadcasted on the network and contained in blocks.
Definition: transaction.h:194
CZMQAbstractPublishNotifier::nSequence
uint32_t nSequence
upcounting per message sequence number
Definition: zmqpublishnotifier.h:15
SER_NETWORK
@ SER_NETWORK
Definition: serialize.h:165
CZMQPublishRawBlockNotifier::NotifyBlock
bool NotifyBlock(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:209
Config
Definition: config.h:17
blockdb.h
zmqutil.h
CZMQPublishSequenceNotifier::NotifyTransactionAcceptance
bool NotifyTransactionAcceptance(const CTransaction &transaction, uint64_t mempool_sequence) override
Definition: zmqpublishnotifier.cpp:269
CZMQAbstractNotifier::psocket
void * psocket
Definition: zmqabstractnotifier.h:63
CTransaction::GetId
const TxId GetId() const
Definition: transaction.h:244
cs_main
RecursiveMutex cs_main
Global state.
Definition: validation.cpp:103
WriteLE64
static void WriteLE64(uint8_t *ptr, uint64_t x)
Definition: common.h:45
CZMQAbstractNotifier::outbound_message_high_water_mark
int outbound_message_high_water_mark
Definition: zmqabstractnotifier.h:66
base_blob::GetHex
std::string GetHex() const
Definition: uint256.cpp:16
CZMQPublishHashTransactionNotifier::NotifyTransaction
bool NotifyTransaction(const CTransaction &transaction) override
Definition: zmqpublishnotifier.cpp:197
ReadBlockFromDisk
bool ReadBlockFromDisk(CBlock &block, const FlatFilePos &pos, const Consensus::Params &params)
Functions for disk access for blocks.
Definition: blockdb.cpp:33
TxId
A TxId is the identifier of a transaction.
Definition: txid.h:14
LogPrint
#define LogPrint(category,...)
Definition: logging.h:193
base_blob::begin
uint8_t * begin()
Definition: uint256.h:83
CZMQAbstractNotifier::address
std::string address
Definition: zmqabstractnotifier.h:65
zmqpublishnotifier.h
CDataStream::size
size_type size() const
Definition: streams.h:280
BlockHash
A BlockHash is a unqiue identifier for a block.
Definition: blockhash.h:13
system.h
CBlock
Definition: block.h:55
MSG_HASHTX
static const char * MSG_HASHTX
Definition: zmqpublishnotifier.cpp:30
Config::GetChainParams
virtual const CChainParams & GetChainParams() const =0
LOCK
#define LOCK(cs)
Definition: sync.h:241
CZMQAbstractPublishNotifier::Initialize
bool Initialize(void *pcontext) override
Definition: zmqpublishnotifier.cpp:75
CZMQPublishRawTransactionNotifier::NotifyTransaction
bool NotifyTransaction(const CTransaction &transaction) override
Definition: zmqpublishnotifier.cpp:230
RPCSerializationFlags
int RPCSerializationFlags()
Retrieves any serialization flags requested in command line argument.
Definition: server.cpp:570
MSG_RAWTX
static const char * MSG_RAWTX
Definition: zmqpublishnotifier.cpp:32
CDataStream
Double ended buffer combining vector and stream-like interfaces.
Definition: streams.h:197
config.h
txid.h
CZMQAbstractPublishNotifier::SendZmqMessage
bool SendZmqMessage(const char *command, const void *data, size_t size)
Definition: zmqpublishnotifier.cpp:166
CZMQPublishSequenceNotifier::NotifyTransactionRemoval
bool NotifyTransactionRemoval(const CTransaction &transaction, uint64_t mempool_sequence) override
Definition: zmqpublishnotifier.cpp:284
server.h
CBlockIndex
The block chain is a tree shaped structure starting with the genesis block at the root,...
Definition: blockindex.h:23
GetConfig
const Config & GetConfig()
Definition: config.cpp:34
CZMQPublishHashBlockNotifier::NotifyBlock
bool NotifyBlock(const CBlockIndex *pindex) override
Definition: zmqpublishnotifier.cpp:186
MSG_HASHBLOCK
static const char * MSG_HASHBLOCK
Definition: zmqpublishnotifier.cpp:29
zmq_send_multipart
static int zmq_send_multipart(void *sock, const void *data, size_t size,...)
Definition: zmqpublishnotifier.cpp:36
PROTOCOL_VERSION
static const int PROTOCOL_VERSION
network protocol versioning
Definition: version.h:11
MSG_SEQUENCE
static const char * MSG_SEQUENCE
Definition: zmqpublishnotifier.cpp:33