Page MenuHomePhabricator
Differential D10339 Diff 30523 src/zmq/zmqpublishnotifier.cpp

Changeset View

Standalone View

View Options

src/zmq/zmqpublishnotifier.cpp

Show First 20 Lines • Show All 179 Lines • ▼ Show 20 Linesbool CZMQAbstractPublishNotifier::SendZmqMessage(const char *command,
/* increment memory only sequence number after sending */ /* increment memory only sequence number after sending */
nSequence++; nSequence++;
return true; return true;
} }
bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex) { bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex) {
BlockHash hash = pindex->GetBlockHash(); BlockHash hash = pindex->GetBlockHash();
LogPrint(BCLog::ZMQ, "zmq: Publish hashblock %s\n", hash.GetHex()); LogPrint(BCLog::ZMQ, "zmq: Publish hashblock %s to %s\n", hash.GetHex(),
this->address);
char data[32]; char data[32];
for (unsigned int i = 0; i < 32; i++) { for (unsigned int i = 0; i < 32; i++) {
data[31 - i] = hash.begin()[i]; data[31 - i] = hash.begin()[i];
} }
return SendZmqMessage(MSG_HASHBLOCK, data, 32); return SendZmqMessage(MSG_HASHBLOCK, data, 32);
} }
bool CZMQPublishHashTransactionNotifier::NotifyTransaction( bool CZMQPublishHashTransactionNotifier::NotifyTransaction(
const CTransaction &transaction) { const CTransaction &transaction) {
TxId txid = transaction.GetId(); TxId txid = transaction.GetId();
LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s\n", txid.GetHex()); LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s to %s\n", txid.GetHex(),
this->address);
char data[32]; char data[32];
for (unsigned int i = 0; i < 32; i++) { for (unsigned int i = 0; i < 32; i++) {
data[31 - i] = txid.begin()[i]; data[31 - i] = txid.begin()[i];
} }
return SendZmqMessage(MSG_HASHTX, data, 32); return SendZmqMessage(MSG_HASHTX, data, 32);
} }
bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex) { bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex) {
LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s\n", LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s to %s\n",
pindex->GetBlockHash().GetHex()); pindex->GetBlockHash().GetHex(), this->address);
const Config &config = GetConfig(); const Config &config = GetConfig();
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags()); CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags());
{ {
LOCK(cs_main); LOCK(cs_main);
CBlock block; CBlock block;
if (!ReadBlockFromDisk(block, pindex, if (!ReadBlockFromDisk(block, pindex,
config.GetChainParams().GetConsensus())) { config.GetChainParams().GetConsensus())) {
zmqError("Can't read block from disk"); zmqError("Can't read block from disk");
return false; return false;
} }
ss << block; ss << block;
} }
return SendZmqMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size()); return SendZmqMessage(MSG_RAWBLOCK, &(*ss.begin()), ss.size());
} }
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 to %s\n", txid.GetHex(),
this->address);
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 // TODO: Dedup this code to take label char, log string
bool CZMQPublishSequenceNotifier::NotifyBlockConnect( bool CZMQPublishSequenceNotifier::NotifyBlockConnect(
const CBlockIndex *pindex) { const CBlockIndex *pindex) {
BlockHash hash = pindex->GetBlockHash(); BlockHash hash = pindex->GetBlockHash();
LogPrint(BCLog::ZMQ, "zmq: Publish sequence block connect %s\n", LogPrint(BCLog::ZMQ, "zmq: Publish sequence block connect %s to %s\n",
hash.GetHex()); hash.GetHex(), this->address);
char data[sizeof(BlockHash) + 1]; char data[sizeof(BlockHash) + 1];
for (unsigned int i = 0; i < sizeof(BlockHash); i++) { for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
data[sizeof(BlockHash) - 1 - i] = hash.begin()[i]; data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
} }
// Block (C)onnect // Block (C)onnect
data[sizeof(data) - 1] = 'C'; data[sizeof(data) - 1] = 'C';
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data)); return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
} }
bool CZMQPublishSequenceNotifier::NotifyBlockDisconnect( bool CZMQPublishSequenceNotifier::NotifyBlockDisconnect(
const CBlockIndex *pindex) { const CBlockIndex *pindex) {
BlockHash hash = pindex->GetBlockHash(); BlockHash hash = pindex->GetBlockHash();
LogPrint(BCLog::ZMQ, "zmq: Publish sequence block disconnect %s\n", LogPrint(BCLog::ZMQ, "zmq: Publish sequence block disconnect %s to %s\n",
hash.GetHex()); hash.GetHex(), this->address);
char data[sizeof(BlockHash) + 1]; char data[sizeof(BlockHash) + 1];
for (unsigned int i = 0; i < sizeof(BlockHash); i++) { for (unsigned int i = 0; i < sizeof(BlockHash); i++) {
data[sizeof(BlockHash) - 1 - i] = hash.begin()[i]; data[sizeof(BlockHash) - 1 - i] = hash.begin()[i];
} }
// Block (D)isconnect // Block (D)isconnect
data[sizeof(data) - 1] = 'D'; data[sizeof(data) - 1] = 'D';
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data)); return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
} }
bool CZMQPublishSequenceNotifier::NotifyTransactionAcceptance( bool CZMQPublishSequenceNotifier::NotifyTransactionAcceptance(
const CTransaction &transaction, uint64_t mempool_sequence) { const CTransaction &transaction, uint64_t mempool_sequence) {
TxId txid = transaction.GetId(); TxId txid = transaction.GetId();
LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool acceptance %s\n", LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool acceptance %s to %s\n",
txid.GetHex()); txid.GetHex(), this->address);
uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1]; uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
for (unsigned int i = 0; i < sizeof(TxId); i++) { for (unsigned int i = 0; i < sizeof(TxId); i++) {
data[sizeof(TxId) - 1 - i] = txid.begin()[i]; data[sizeof(TxId) - 1 - i] = txid.begin()[i];
} }
// Mempool (A)cceptance // Mempool (A)cceptance
data[sizeof(TxId)] = 'A'; data[sizeof(TxId)] = 'A';
WriteLE64(data + sizeof(TxId) + 1, mempool_sequence); WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data)); return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
} }
bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval( bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval(
const CTransaction &transaction, uint64_t mempool_sequence) { const CTransaction &transaction, uint64_t mempool_sequence) {
TxId txid = transaction.GetId(); TxId txid = transaction.GetId();
LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool removal %s\n", LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool removal %s to %s\n",
txid.GetHex()); txid.GetHex(), this->address);
uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1]; uint8_t data[sizeof(TxId) + sizeof(mempool_sequence) + 1];
for (unsigned int i = 0; i < sizeof(TxId); i++) { for (unsigned int i = 0; i < sizeof(TxId); i++) {
data[sizeof(TxId) - 1 - i] = txid.begin()[i]; data[sizeof(TxId) - 1 - i] = txid.begin()[i];
} }
// Mempool (R)emoval // Mempool (R)emoval
data[sizeof(TxId)] = 'R'; data[sizeof(TxId)] = 'R';
WriteLE64(data + sizeof(TxId) + 1, mempool_sequence); WriteLE64(data + sizeof(TxId) + 1, mempool_sequence);
return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data)); return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
} }