Changeset View
Changeset View
Standalone View
Standalone View
src/zmq/zmqpublishnotifier.cpp
Show First 20 Lines • Show All 179 Lines • ▼ Show 20 Lines | bool 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)); | ||||
} | } |