Differential D10304 Diff 30509 src/zmq/zmqnotificationinterface.cpp

Changeset View

Standalone View

src/zmq/zmqnotificationinterface.cpp

Show All 30 Lines	CZMQNotificationInterface *CZMQNotificationInterface::Create() {
factories["pubhashblock"] =		factories["pubhashblock"] =
CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>;		CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>;
factories["pubhashtx"] =		factories["pubhashtx"] =
CZMQAbstractNotifier::Create<CZMQPublishHashTransactionNotifier>;		CZMQAbstractNotifier::Create<CZMQPublishHashTransactionNotifier>;
factories["pubrawblock"] =		factories["pubrawblock"] =
CZMQAbstractNotifier::Create<CZMQPublishRawBlockNotifier>;		CZMQAbstractNotifier::Create<CZMQPublishRawBlockNotifier>;
factories["pubrawtx"] =		factories["pubrawtx"] =
CZMQAbstractNotifier::Create<CZMQPublishRawTransactionNotifier>;		CZMQAbstractNotifier::Create<CZMQPublishRawTransactionNotifier>;
		factories["pubsequence"] =
		CZMQAbstractNotifier::Create<CZMQPublishSequenceNotifier>;

std::list<std::unique_ptr<CZMQAbstractNotifier>> notifiers;		std::list<std::unique_ptr<CZMQAbstractNotifier>> notifiers;
for (const auto &entry : factories) {		for (const auto &entry : factories) {
std::string arg("-zmq" + entry.first);		std::string arg("-zmq" + entry.first);
if (gArgs.IsArgSet(arg)) {		if (gArgs.IsArgSet(arg)) {
const auto &factory = entry.second;		const auto &factory = entry.second;
const std::string address = gArgs.GetArg(arg, "");		const std::string address = gArgs.GetArg(arg, "");
std::unique_ptr<CZMQAbstractNotifier> notifier = factory();		std::unique_ptr<CZMQAbstractNotifier> notifier = factory();
▲ Show 20 Lines • Show All 93 Lines • ▼ Show 20 Lines	void CZMQNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew,

TryForEachAndRemoveFailed(notifiers,		TryForEachAndRemoveFailed(notifiers,
[pindexNew](CZMQAbstractNotifier *notifier) {		[pindexNew](CZMQAbstractNotifier *notifier) {
return notifier->NotifyBlock(pindexNew);		return notifier->NotifyBlock(pindexNew);
});		});
}		}

void CZMQNotificationInterface::TransactionAddedToMempool(		void CZMQNotificationInterface::TransactionAddedToMempool(
const CTransactionRef &ptx) {		const CTransactionRef &ptx, uint64_t mempool_sequence) {
// Used by BlockConnected and BlockDisconnected as well, because they're all
// the same external callback.
const CTransaction &tx = *ptx;		const CTransaction &tx = *ptx;

TryForEachAndRemoveFailed(notifiers, [&tx](CZMQAbstractNotifier *notifier) {		TryForEachAndRemoveFailed(
return notifier->NotifyTransaction(tx);		notifiers, [&tx, mempool_sequence](CZMQAbstractNotifier *notifier) {
		return notifier->NotifyTransaction(tx) &&
		notifier->NotifyTransactionAcceptance(tx, mempool_sequence);
		});
		}

		void CZMQNotificationInterface::TransactionRemovedFromMempool(
		const CTransactionRef &ptx, MemPoolRemovalReason reason,
		uint64_t mempool_sequence) {
		// Called for all non-block inclusion reasons
		const CTransaction &tx = *ptx;

		TryForEachAndRemoveFailed(
		notifiers, [&tx, mempool_sequence](CZMQAbstractNotifier *notifier) {
		return notifier->NotifyTransactionRemoval(tx, mempool_sequence);
});		});
}		}

void CZMQNotificationInterface::BlockConnected(		void CZMQNotificationInterface::BlockConnected(
const std::shared_ptr<const CBlock> &pblock,		const std::shared_ptr<const CBlock> &pblock,
const CBlockIndex *pindexConnected) {		const CBlockIndex *pindexConnected) {
for (const CTransactionRef &ptx : pblock->vtx) {		for (const CTransactionRef &ptx : pblock->vtx) {
// Do a normal notify for each transaction added in the block		const CTransaction &tx = *ptx;
TransactionAddedToMempool(ptx);		TryForEachAndRemoveFailed(notifiers,
		[&tx](CZMQAbstractNotifier *notifier) {
		return notifier->NotifyTransaction(tx);
		});
}		}

		// Next we notify BlockConnect listeners for all blocks
		TryForEachAndRemoveFailed(
		notifiers, [pindexConnected](CZMQAbstractNotifier *notifier) {
		return notifier->NotifyBlockConnect(pindexConnected);
		});
}		}

void CZMQNotificationInterface::BlockDisconnected(		void CZMQNotificationInterface::BlockDisconnected(
const std::shared_ptr<const CBlock> &pblock,		const std::shared_ptr<const CBlock> &pblock,
const CBlockIndex *pindexDisconnected) {		const CBlockIndex *pindexDisconnected) {
for (const CTransactionRef &ptx : pblock->vtx) {		for (const CTransactionRef &ptx : pblock->vtx) {
// Do a normal notify for each transaction removed in block		const CTransaction &tx = *ptx;
// disconnection		TryForEachAndRemoveFailed(notifiers,
TransactionAddedToMempool(ptx);		[&tx](CZMQAbstractNotifier *notifier) {
		return notifier->NotifyTransaction(tx);
		});
}		}

		// Next we notify BlockDisconnect listeners for all blocks
		TryForEachAndRemoveFailed(
		notifiers, [pindexDisconnected](CZMQAbstractNotifier *notifier) {
		return notifier->NotifyBlockDisconnect(pindexDisconnected);
		});
}		}

CZMQNotificationInterface *g_zmq_notification_interface = nullptr;		CZMQNotificationInterface *g_zmq_notification_interface = nullptr;