SubscriptionManager.hpp

#pragma once
 
#include "data/AmendmentCenterInterface.hpp"
#include "data/BackendInterface.hpp"
#include "data/Types.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "feed/Types.hpp"
#include "feed/impl/BookChangesFeed.hpp"
#include "feed/impl/ForwardFeed.hpp"
#include "feed/impl/LedgerFeed.hpp"
#include "feed/impl/ProposedTransactionFeed.hpp"
#include "feed/impl/TransactionFeed.hpp"
#include "util/async/AnyExecutionContext.hpp"
#include "util/async/context/BasicExecutionContext.hpp"
#include "util/config/ConfigDefinition.hpp"
#include "util/log/Logger.hpp"
 
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <xrpl/protocol/AccountID.h>
#include <xrpl/protocol/Book.h>
#include <xrpl/protocol/Fees.h>
#include <xrpl/protocol/LedgerHeader.h>
 
#include <cstdint>
#include <memory>
#include <string>
#include <utility>
#include <vector>

namespace feed {

class SubscriptionManager : public SubscriptionManagerInterface {
    std::shared_ptr<data::BackendInterface const> backend_;
    std::shared_ptr<data::AmendmentCenterInterface const> amendmentCenter_;
    util::async::AnyExecutionContext ctx_;
    impl::ForwardFeed manifestFeed_;
    impl::ForwardFeed validationsFeed_;
    impl::LedgerFeed ledgerFeed_;
    impl::BookChangesFeed bookChangesFeed_;
    impl::TransactionFeed transactionFeed_;
    impl::ProposedTransactionFeed proposedTransactionFeed_;
    uint32_t networkID_{0};
 
public:
    static std::shared_ptr<SubscriptionManager>
    makeSubscriptionManager(
        util::config::ClioConfigDefinition const& config,
        std::shared_ptr<data::BackendInterface const> const& backend,
        std::shared_ptr<data::AmendmentCenterInterface const> const& amendmentCenter
    )
    {
        auto const workersNum = config.get<uint64_t>("subscription_workers");
 
        util::Logger const logger{"Subscriptions"};
        LOG(logger.info()) << "Starting subscription manager with " << workersNum << " workers";
 
        return std::make_shared<feed::SubscriptionManager>(
            util::async::PoolExecutionContext(workersNum), backend, amendmentCenter
        );
    }

    SubscriptionManager(
        util::async::AnyExecutionContext&& executor,
        std::shared_ptr<data::BackendInterface const> const& backend,
        std::shared_ptr<data::AmendmentCenterInterface const> const& amendmentCenter
    )
        : backend_(backend)
        , amendmentCenter_(amendmentCenter)
        , ctx_(std::move(executor))
        , manifestFeed_(ctx_, "manifest")
        , validationsFeed_(ctx_, "validations")
        , ledgerFeed_(ctx_)
        , bookChangesFeed_(ctx_)
        , transactionFeed_(ctx_)
        , proposedTransactionFeed_(ctx_)
    {
    }

    ~SubscriptionManager() override
    {
        stop();
    }

    void
    stop() override
    {
        ctx_.stop();
        ctx_.join();
    }

    void
    subBookChanges(SubscriberSharedPtr const& subscriber) final;

    void
    unsubBookChanges(SubscriberSharedPtr const& subscriber) final;

    void
    pubBookChanges(
        ripple::LedgerHeader const& lgrInfo,
        std::vector<data::TransactionAndMetadata> const& transactions
    ) final;

    void
    subProposedTransactions(SubscriberSharedPtr const& subscriber) final;

    void
    unsubProposedTransactions(SubscriberSharedPtr const& subscriber) final;

    void
    subProposedAccount(
        ripple::AccountID const& account,
        SubscriberSharedPtr const& subscriber
    ) final;

    void
    unsubProposedAccount(
        ripple::AccountID const& account,
        SubscriberSharedPtr const& subscriber
    ) final;

    void
    forwardProposedTransaction(boost::json::object const& receivedTxJson) final;

    boost::json::object
    subLedger(boost::asio::yield_context yield, SubscriberSharedPtr const& subscriber) final;

    void
    unsubLedger(SubscriberSharedPtr const& subscriber) final;

    void
    pubLedger(
        ripple::LedgerHeader const& lgrInfo,
        ripple::Fees const& fees,
        std::string const& ledgerRange,
        std::uint32_t txnCount
    ) final;

    void
    subManifest(SubscriberSharedPtr const& subscriber) final;

    void
    unsubManifest(SubscriberSharedPtr const& subscriber) final;

    void
    forwardManifest(boost::json::object const& manifestJson) final;

    void
    subValidation(SubscriberSharedPtr const& subscriber) final;

    void
    unsubValidation(SubscriberSharedPtr const& subscriber) final;

    void
    forwardValidation(boost::json::object const& validationJson) final;

    void
    subTransactions(SubscriberSharedPtr const& subscriber) final;

    void
    unsubTransactions(SubscriberSharedPtr const& subscriber) final;

    void
    subAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) final;

    void
    unsubAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) final;

    void
    subBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber) final;

    void
    unsubBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber) final;

    void
    pubTransaction(
        data::TransactionAndMetadata const& txMeta,
        ripple::LedgerHeader const& lgrInfo
    ) final;

    boost::json::object
    report() const final;

    void
    setNetworkID(uint32_t networkID) final;

    uint32_t
    getNetworkID() const final;
};
 
}  // namespace feed