aboutsummaryrefslogtreecommitdiff
path: root/src/zmq
diff options
context:
space:
mode:
Diffstat (limited to 'src/zmq')
-rw-r--r--src/zmq/zmqnotificationinterface.cpp34
-rw-r--r--src/zmq/zmqnotificationinterface.h7
-rw-r--r--src/zmq/zmqpublishnotifier.cpp16
3 files changed, 42 insertions, 15 deletions
diff --git a/src/zmq/zmqnotificationinterface.cpp b/src/zmq/zmqnotificationinterface.cpp
index 431d8c9ac9..c063898056 100644
--- a/src/zmq/zmqnotificationinterface.cpp
+++ b/src/zmq/zmqnotificationinterface.cpp
@@ -12,7 +12,7 @@
void zmqError(const char *str)
{
- LogPrint("zmq", "zmq: Error: %s, errno=%s\n", str, zmq_strerror(errno));
+ LogPrint(BCLog::ZMQ, "zmq: Error: %s, errno=%s\n", str, zmq_strerror(errno));
}
CZMQNotificationInterface::CZMQNotificationInterface() : pcontext(NULL)
@@ -72,7 +72,7 @@ CZMQNotificationInterface* CZMQNotificationInterface::Create()
// Called at startup to conditionally set up ZMQ socket(s)
bool CZMQNotificationInterface::Initialize()
{
- LogPrint("zmq", "zmq: Initialize notification interface\n");
+ LogPrint(BCLog::ZMQ, "zmq: Initialize notification interface\n");
assert(!pcontext);
pcontext = zmq_init(1);
@@ -89,11 +89,11 @@ bool CZMQNotificationInterface::Initialize()
CZMQAbstractNotifier *notifier = *i;
if (notifier->Initialize(pcontext))
{
- LogPrint("zmq", " Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
+ LogPrint(BCLog::ZMQ, " Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
}
else
{
- LogPrint("zmq", " Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
+ LogPrint(BCLog::ZMQ, " Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
break;
}
}
@@ -109,13 +109,13 @@ bool CZMQNotificationInterface::Initialize()
// Called during shutdown sequence
void CZMQNotificationInterface::Shutdown()
{
- LogPrint("zmq", "zmq: Shutdown notification interface\n");
+ LogPrint(BCLog::ZMQ, "zmq: Shutdown notification interface\n");
if (pcontext)
{
for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i)
{
CZMQAbstractNotifier *notifier = *i;
- LogPrint("zmq", " Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress());
+ LogPrint(BCLog::ZMQ, " Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress());
notifier->Shutdown();
}
zmq_ctx_destroy(pcontext);
@@ -144,8 +144,12 @@ void CZMQNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, co
}
}
-void CZMQNotificationInterface::SyncTransaction(const CTransaction& tx, const CBlockIndex* pindex, int posInBlock)
+void CZMQNotificationInterface::TransactionAddedToMempool(const CTransactionRef& ptx)
{
+ // Used by BlockConnected and BlockDisconnected as well, because they're
+ // all the same external callback.
+ const CTransaction& tx = *ptx;
+
for (std::list<CZMQAbstractNotifier*>::iterator i = notifiers.begin(); i!=notifiers.end(); )
{
CZMQAbstractNotifier *notifier = *i;
@@ -160,3 +164,19 @@ void CZMQNotificationInterface::SyncTransaction(const CTransaction& tx, const CB
}
}
}
+
+void CZMQNotificationInterface::BlockConnected(const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindexConnected, const std::vector<CTransactionRef>& vtxConflicted)
+{
+ for (const CTransactionRef& ptx : pblock->vtx) {
+ // Do a normal notify for each transaction added in the block
+ TransactionAddedToMempool(ptx);
+ }
+}
+
+void CZMQNotificationInterface::BlockDisconnected(const std::shared_ptr<const CBlock>& pblock)
+{
+ for (const CTransactionRef& ptx : pblock->vtx) {
+ // Do a normal notify for each transaction removed in block disconnection
+ TransactionAddedToMempool(ptx);
+ }
+}
diff --git a/src/zmq/zmqnotificationinterface.h b/src/zmq/zmqnotificationinterface.h
index beabb78da6..eec6f7bc64 100644
--- a/src/zmq/zmqnotificationinterface.h
+++ b/src/zmq/zmqnotificationinterface.h
@@ -8,6 +8,7 @@
#include "validationinterface.h"
#include <string>
#include <map>
+#include <list>
class CBlockIndex;
class CZMQAbstractNotifier;
@@ -24,8 +25,10 @@ protected:
void Shutdown();
// CValidationInterface
- void SyncTransaction(const CTransaction& tx, const CBlockIndex *pindex, int posInBlock);
- void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload);
+ void TransactionAddedToMempool(const CTransactionRef& tx) override;
+ void BlockConnected(const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindexConnected, const std::vector<CTransactionRef>& vtxConflicted) override;
+ void BlockDisconnected(const std::shared_ptr<const CBlock>& pblock) override;
+ void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) override;
private:
CZMQNotificationInterface();
diff --git a/src/zmq/zmqpublishnotifier.cpp b/src/zmq/zmqpublishnotifier.cpp
index caca1248a1..700c39f66e 100644
--- a/src/zmq/zmqpublishnotifier.cpp
+++ b/src/zmq/zmqpublishnotifier.cpp
@@ -2,6 +2,7 @@
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+#include "chain.h"
#include "chainparams.h"
#include "streams.h"
#include "zmqpublishnotifier.h"
@@ -30,6 +31,7 @@ static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
if (rc != 0)
{
zmqError("Unable to initialize ZMQ msg");
+ va_end(args);
return -1;
}
@@ -43,6 +45,7 @@ static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
{
zmqError("Unable to send ZMQ msg");
zmq_msg_close(&msg);
+ va_end(args);
return -1;
}
@@ -53,6 +56,7 @@ static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
size = va_arg(args, size_t);
}
+ va_end(args);
return 0;
}
@@ -86,7 +90,7 @@ bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
}
else
{
- LogPrint("zmq", "zmq: Reusing socket for address %s\n", address);
+ LogPrint(BCLog::ZMQ, "zmq: Reusing socket for address %s\n", address);
psocket = i->second->psocket;
mapPublishNotifiers.insert(std::make_pair(address, this));
@@ -116,7 +120,7 @@ void CZMQAbstractPublishNotifier::Shutdown()
if (count == 1)
{
- LogPrint("zmq", "Close socket at address %s\n", address);
+ LogPrint(BCLog::ZMQ, "Close socket at address %s\n", address);
int linger = 0;
zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger));
zmq_close(psocket);
@@ -145,7 +149,7 @@ bool CZMQAbstractPublishNotifier::SendMessage(const char *command, const void* d
bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
{
uint256 hash = pindex->GetBlockHash();
- LogPrint("zmq", "zmq: Publish hashblock %s\n", hash.GetHex());
+ LogPrint(BCLog::ZMQ, "zmq: Publish hashblock %s\n", hash.GetHex());
char data[32];
for (unsigned int i = 0; i < 32; i++)
data[31 - i] = hash.begin()[i];
@@ -155,7 +159,7 @@ bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
{
uint256 hash = transaction.GetHash();
- LogPrint("zmq", "zmq: Publish hashtx %s\n", hash.GetHex());
+ LogPrint(BCLog::ZMQ, "zmq: Publish hashtx %s\n", hash.GetHex());
char data[32];
for (unsigned int i = 0; i < 32; i++)
data[31 - i] = hash.begin()[i];
@@ -164,7 +168,7 @@ bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &t
bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
{
- LogPrint("zmq", "zmq: Publish rawblock %s\n", pindex->GetBlockHash().GetHex());
+ LogPrint(BCLog::ZMQ, "zmq: Publish rawblock %s\n", pindex->GetBlockHash().GetHex());
const Consensus::Params& consensusParams = Params().GetConsensus();
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags());
@@ -186,7 +190,7 @@ bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
bool CZMQPublishRawTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
{
uint256 hash = transaction.GetHash();
- LogPrint("zmq", "zmq: Publish rawtx %s\n", hash.GetHex());
+ LogPrint(BCLog::ZMQ, "zmq: Publish rawtx %s\n", hash.GetHex());
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION | RPCSerializationFlags());
ss << transaction;
return SendMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
generated by cgit v1.2.3 (git 2.26.2) at 2024-11-02 17:08:03 +0000