Differential D10304 Diff 30446 src/zmq/zmqpublishnotifier.cpp

Changeset View

Standalone View

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) {
				uint256 hash = pindex->GetBlockHash();
				FabienUnsubmitted Not Done Inline Actions That's a BlockHash Fabien: That's a BlockHash
				LogPrint(BCLog::ZMQ, "zmq: Publish sequence block connect %s\n",
				hash.GetHex());
				char data[sizeof(uint256) + 1];
				for (unsigned int i = 0; i < sizeof(uint256); i++) {
				data[sizeof(uint256) - 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) {
				uint256 hash = pindex->GetBlockHash();
				FabienUnsubmitted Not Done Inline Actions dito Fabien: dito
				LogPrint(BCLog::ZMQ, "zmq: Publish sequence block disconnect %s\n",
				hash.GetHex());
				char data[sizeof(uint256) + 1];
				for (unsigned int i = 0; i < sizeof(uint256); i++) {
				data[sizeof(uint256) - 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) {
				uint256 hash = transaction.GetHash();
				FabienUnsubmitted Not Done Inline Actions And that would better be a TxId Fabien: And that would better be a TxId
				LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool acceptance %s\n",
				hash.GetHex());
				uint8_t data[sizeof(uint256) + sizeof(mempool_sequence) + 1];
				for (unsigned int i = 0; i < sizeof(uint256); i++) {
				data[sizeof(uint256) - 1 - i] = hash.begin()[i];
				}
				// Mempool (A)cceptance
				data[sizeof(uint256)] = 'A';
				WriteLE64(data + sizeof(uint256) + 1, mempool_sequence);
				return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
				}

				bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval(
				const CTransaction &transaction, uint64_t mempool_sequence) {
				uint256 hash = transaction.GetHash();
				LogPrint(BCLog::ZMQ, "zmq: Publish hashtx mempool removal %s\n",
				hash.GetHex());
				uint8_t data[sizeof(uint256) + sizeof(mempool_sequence) + 1];
				for (unsigned int i = 0; i < sizeof(uint256); i++) {
				data[sizeof(uint256) - 1 - i] = hash.begin()[i];
				}
				data[sizeof(uint256)] = 'R'; // Mempool (R)emoval
				WriteLE64(data + sizeof(uint256) + 1, mempool_sequence);
				return SendZmqMessage(MSG_SEQUENCE, data, sizeof(data));
				}