137 lines
4.7 KiB
C++
137 lines
4.7 KiB
C++
// Copyright (c) 2017 The Zcash developers
|
|
// Distributed under the MIT software license, see the accompanying
|
|
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
|
|
|
#include "amqpnotificationinterface.h"
|
|
#include "amqppublishnotifier.h"
|
|
|
|
#include "version.h"
|
|
#include "main.h"
|
|
#include "streams.h"
|
|
#include "util.h"
|
|
|
|
// AMQP 1.0 Support
|
|
//
|
|
// The boost::signals2 signals and slot system is thread safe, so CValidationInterface listeners
|
|
// can be invoked from any thread.
|
|
//
|
|
// Currently signals are fired from main.cpp so the callbacks should be invoked on the same thread.
|
|
// It should be safe to share objects responsible for sending, as they should not be run concurrently
|
|
// across different threads.
|
|
//
|
|
// Developers should be mindful of where notifications are fired to avoid potential race conditions.
|
|
// For example, different signals targeting the same address could be fired from different threads
|
|
// in different parts of the system around the same time.
|
|
//
|
|
// Like the ZMQ notification interface, if a notifier fails to send a message, the notifier is shut down.
|
|
//
|
|
|
|
AMQPNotificationInterface::AMQPNotificationInterface()
|
|
{
|
|
}
|
|
|
|
AMQPNotificationInterface::~AMQPNotificationInterface()
|
|
{
|
|
Shutdown();
|
|
|
|
for (std::list<AMQPAbstractNotifier*>::iterator i = notifiers.begin(); i != notifiers.end(); ++i) {
|
|
delete *i;
|
|
}
|
|
}
|
|
|
|
AMQPNotificationInterface* AMQPNotificationInterface::CreateWithArguments(const std::map<std::string, std::string> &args)
|
|
{
|
|
AMQPNotificationInterface* notificationInterface = nullptr;
|
|
std::map<std::string, AMQPNotifierFactory> factories;
|
|
std::list<AMQPAbstractNotifier*> notifiers;
|
|
|
|
factories["pubhashblock"] = AMQPAbstractNotifier::Create<AMQPPublishHashBlockNotifier>;
|
|
factories["pubhashtx"] = AMQPAbstractNotifier::Create<AMQPPublishHashTransactionNotifier>;
|
|
factories["pubrawblock"] = AMQPAbstractNotifier::Create<AMQPPublishRawBlockNotifier>;
|
|
factories["pubrawtx"] = AMQPAbstractNotifier::Create<AMQPPublishRawTransactionNotifier>;
|
|
|
|
for (std::map<std::string, AMQPNotifierFactory>::const_iterator i=factories.begin(); i!=factories.end(); ++i) {
|
|
std::map<std::string, std::string>::const_iterator j = args.find("-amqp" + i->first);
|
|
if (j!=args.end()) {
|
|
AMQPNotifierFactory factory = i->second;
|
|
std::string address = j->second;
|
|
AMQPAbstractNotifier *notifier = factory();
|
|
notifier->SetType(i->first);
|
|
notifier->SetAddress(address);
|
|
notifiers.push_back(notifier);
|
|
}
|
|
}
|
|
|
|
if (!notifiers.empty()) {
|
|
notificationInterface = new AMQPNotificationInterface();
|
|
notificationInterface->notifiers = notifiers;
|
|
|
|
if (!notificationInterface->Initialize()) {
|
|
delete notificationInterface;
|
|
notificationInterface = nullptr;
|
|
}
|
|
}
|
|
|
|
return notificationInterface;
|
|
}
|
|
|
|
// Called at startup to conditionally set up
|
|
bool AMQPNotificationInterface::Initialize()
|
|
{
|
|
LogPrint("amqp", "amqp: Initialize notification interface\n");
|
|
|
|
std::list<AMQPAbstractNotifier*>::iterator i = notifiers.begin();
|
|
for (; i != notifiers.end(); ++i) {
|
|
AMQPAbstractNotifier *notifier = *i;
|
|
if (notifier->Initialize()) {
|
|
LogPrint("amqp", "amqp: Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress());
|
|
} else {
|
|
LogPrint("amqp", "amqp: Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress());
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (i != notifiers.end()) {
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
// Called during shutdown sequence
|
|
void AMQPNotificationInterface::Shutdown()
|
|
{
|
|
LogPrint("amqp", "amqp: Shutdown notification interface\n");
|
|
|
|
for (std::list<AMQPAbstractNotifier*>::iterator i = notifiers.begin(); i != notifiers.end(); ++i) {
|
|
AMQPAbstractNotifier *notifier = *i;
|
|
notifier->Shutdown();
|
|
}
|
|
}
|
|
|
|
void AMQPNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindex)
|
|
{
|
|
for (std::list<AMQPAbstractNotifier*>::iterator i = notifiers.begin(); i != notifiers.end(); ) {
|
|
AMQPAbstractNotifier *notifier = *i;
|
|
if (notifier->NotifyBlock(pindex)) {
|
|
i++;
|
|
} else {
|
|
notifier->Shutdown();
|
|
i = notifiers.erase(i);
|
|
}
|
|
}
|
|
}
|
|
|
|
void AMQPNotificationInterface::SyncTransaction(const CTransaction &tx, const CBlock *pblock)
|
|
{
|
|
for (std::list<AMQPAbstractNotifier*>::iterator i = notifiers.begin(); i != notifiers.end(); ) {
|
|
AMQPAbstractNotifier *notifier = *i;
|
|
if (notifier->NotifyTransaction(tx)) {
|
|
i++;
|
|
} else {
|
|
notifier->Shutdown();
|
|
i = notifiers.erase(i);
|
|
}
|
|
}
|
|
}
|