Page MenuHomePhabricator
Differential D10304 Diff 30459 src/zmq/zmqpublishnotifier.cpp

Changeset View

Standalone View

View Options

src/zmq/zmqpublishnotifier.cpp

Show All 24 Lines
static std::multimap<std::string, CZMQAbstractPublishNotifier *> static std::multimap<std::string, CZMQAbstractPublishNotifier *>
mapPublishNotifiers; mapPublishNotifiers;
static const char *MSG_HASHBLOCK = "hashblock"; static const char *MSG_HASHBLOCK = "hashblock";
static const char *MSG_HASHTX = "hashtx"; static const char *MSG_HASHTX = "hashtx";
static const char *MSG_RAWBLOCK = "rawblock"; static const char *MSG_RAWBLOCK = "rawblock";
static const char *MSG_RAWTX = "rawtx"; static const char *MSG_RAWTX = "rawtx";
static const char *MSG_SEQUENCE = "sequence";
// Internal function to send multipart message // Internal function to send multipart message
static int zmq_send_multipart(void *sock, const void *data, size_t size, ...) { static int zmq_send_multipart(void *sock, const void *data, size_t size, ...) {
va_list args; va_list args;
va_start(args, size); va_start(args, size);
while (1) { while (1) {
zmq_msg_t msg; zmq_msg_t msg;
▲ Show 20 Lines • Show All 177 Lines • ▼ Show 20 Lines
bool CZMQPublishRawTransactionNotifier::NotifyTransaction( bool CZMQPublishRawTransactionNotifier::NotifyTransaction(
const CTransaction &transaction) { const CTransaction &transaction) {
TxId txid = transaction.GetId(); TxId txid = transaction.GetId();
LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s\n", txid.GetHex()); LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s\n", txid.GetHex());
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags()); CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags());
ss << transaction; ss << transaction;
return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size()); return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
} }
// TODO: Dedup this code to take label char, log string
bool CZMQPublishSequenceNotifier::NotifyBlockConnect(
const CBlockIndex *pindex) {
BlockHash hash = pindex->GetBlockHash();
LogPrint(BCLog::ZMQ, "zmq: Publish sequence block connect %s\n",
hash.GetHex());
char data[sizeof(BlockHash) + 1];
for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
}
// Block (C)onnect
data[sizeof(data) - 1] = 'C';
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}
bool CZMQPublishSequenceNotifier::NotifyBlockDisconnect(
const CBlockIndex *pindex) {
BlockHash hash = pindex->GetBlockHash();
LogPrint(BCLog::ZMQ, "zmq: Publish sequence block disconnect %s\n",
hash.GetHex());
char data[sizeof(BlockHash) + 1];
for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
}
// Block (D)isconnect
data[sizeof(data) - 1] = 'D';
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}
bool CZMQPublishSequenceNotifier::NotifyTransactionAcceptance(
const CTransaction &transaction, uint64_t mempool_sequence) {
TxId txid = transaction.GetId();
LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool acceptance %s\n",
txid.GetHex());
uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
for (unsigned int i = 0; i < sizeof(TxId); i++) {
data[sizeof(TxId) - 1 - i] = txid.begin()[i];
}
// Mempool (A)cceptance
data[sizeof(TxId)] = 'A';
WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}
bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval(
const CTransaction &transaction, uint64_t mempool_sequence) {
TxId txid = transaction.GetId();
LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool removal %s\n",
txid.GetHex());
uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
for (unsigned int i = 0; i < sizeof(TxId); i++) {
data[sizeof(TxId) - 1 - i] = txid.begin()[i];
}
// Mempool (R)emoval
data[sizeof(TxId)] = 'R';
WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
}