Differential D10286 Diff 30387 src/zmq/zmqpublishnotifier.cpp

Changeset View

Standalone View

src/zmq/zmqpublishnotifier.cpp

Show First 20 Lines • Show All 147 Lines • ▼ Show 20 Lines	if (count == 1) {
int linger = 0;		int linger = 0;
zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger));		zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger));
zmq_close(psocket);		zmq_close(psocket);
}		}

psocket = nullptr;		psocket = nullptr;
}		}

bool CZMQAbstractPublishNotifier::SendMessage(const char *command,		bool CZMQAbstractPublishNotifier::SendZmqMessage(const char *command,
const void *data, size_t size) {		const void *data,
		size_t size) {
assert(psocket);		assert(psocket);

/* send three parts, command & data & a LE 4byte sequence number */		/* send three parts, command & data & a LE 4byte sequence number */
uint8_t msgseq[sizeof(uint32_t)];		uint8_t msgseq[sizeof(uint32_t)];
WriteLE32(&msgseq[0], nSequence);		WriteLE32(&msgseq[0], nSequence);
int rc = zmq_send_multipart(psocket, command, strlen(command), data, size,		int rc = zmq_send_multipart(psocket, command, strlen(command), data, size,
msgseq, (size_t)sizeof(uint32_t), nullptr);		msgseq, (size_t)sizeof(uint32_t), nullptr);
if (rc == -1) {		if (rc == -1) {
return false;		return false;
}		}

/* 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\n", hash.GetHex());
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 SendMessage(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\n", txid.GetHex());
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 SendMessage(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\n",
pindex->GetBlockHash().GetHex());		pindex->GetBlockHash().GetHex());

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 SendMessage(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\n", txid.GetHex());
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION \| RPCSerializationFlags());		CDataStream ss(SER_NETWORK, PROTOCOL_VERSION \| RPCSerializationFlags());
ss << transaction;		ss << transaction;
return SendMessage(MSG_RAWTX, &(*ss.begin()), ss.size());		return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
}		}