TransactionFeed.hpp

#pragma once
 
#include "data/AmendmentCenterInterface.hpp"
#include "data/BackendInterface.hpp"
#include "data/Types.hpp"
#include "feed/Types.hpp"
#include "feed/impl/TrackableSignal.hpp"
#include "feed/impl/TrackableSignalMap.hpp"
#include "feed/impl/Util.hpp"
#include "util/async/AnyExecutionContext.hpp"
#include "util/async/AnyStrand.hpp"
#include "util/log/Logger.hpp"
#include "util/prometheus/Gauge.hpp"
 
#include <boost/asio/io_context.hpp>
#include <boost/asio/strand.hpp>
#include <fmt/format.h>
#include <xrpl/protocol/AccountID.h>
#include <xrpl/protocol/Book.h>
#include <xrpl/protocol/LedgerHeader.h>
 
#include <cstdint>
#include <functional>
#include <memory>
#include <string>
#include <unordered_set>
 
namespace feed::impl {
 
class TransactionFeed {
    // Hold two versions of transaction messages
    struct AllVersionsMsgsType {
        std::string v1;
        std::string v2;
    };
 
    struct TransactionSlot {
        std::reference_wrapper<TransactionFeed> feed;
        std::weak_ptr<Subscriber> subscriptionContextWeakPtr;
 
        TransactionSlot(TransactionFeed& feed, SubscriberSharedPtr const& connection)
            : feed(feed), subscriptionContextWeakPtr(connection)
        {
        }
 
        void
        operator()(std::shared_ptr<AllVersionsMsgsType> const& allVersionMsgs) const;
    };
 
    util::Logger logger_{"Subscriptions"};
 
    util::async::AnyStrand strand_;
    std::reference_wrapper<util::prometheus::GaugeInt> subAllCount_;
    std::reference_wrapper<util::prometheus::GaugeInt> subAccountCount_;
    std::reference_wrapper<util::prometheus::GaugeInt> subBookCount_;
 
    TrackableSignalMap<ripple::AccountID, Subscriber, std::shared_ptr<AllVersionsMsgsType> const&>
        accountSignal_;
    TrackableSignalMap<ripple::Book, Subscriber, std::shared_ptr<AllVersionsMsgsType> const&>
        bookSignal_;
    TrackableSignal<Subscriber, std::shared_ptr<AllVersionsMsgsType> const&> signal_;
 
    // Signals for proposed tx subscribers
    TrackableSignalMap<ripple::AccountID, Subscriber, std::shared_ptr<AllVersionsMsgsType> const&>
        accountProposedSignal_;
    TrackableSignal<Subscriber, std::shared_ptr<AllVersionsMsgsType> const&> txProposedSignal_;
 
    std::unordered_set<SubscriberPtr> notified_;  // Used by slots to prevent double notifications
                                                  // if tx contains multiple subscribed accounts
 
public:
    TransactionFeed(util::async::AnyExecutionContext& executionCtx)
        : strand_(executionCtx.makeStrand())
        , subAllCount_(getSubscriptionsGaugeInt("tx"))
        , subAccountCount_(getSubscriptionsGaugeInt("account"))
        , subBookCount_(getSubscriptionsGaugeInt("book"))
    {
    }

    TransactionFeed(TransactionFeed&&) = delete;

    void
    sub(SubscriberSharedPtr const& subscriber);

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

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

    void
    subProposed(SubscriberSharedPtr const& subscriber);

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

    void
    unsub(SubscriberSharedPtr const& subscriber);

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

    void
    unsubProposed(SubscriberSharedPtr const& subscriber);

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

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

    void
    pub(data::TransactionAndMetadata const& txMeta,
        ripple::LedgerHeader const& lgrInfo,
        std::shared_ptr<data::BackendInterface const> const& backend,
        std::shared_ptr<data::AmendmentCenterInterface const> const& amendmentCenter,
        uint32_t networkID);

    std::uint64_t
    transactionSubCount() const;

    std::uint64_t
    accountSubCount() const;

    std::uint64_t
    bookSubCount() const;
 
private:
    void
    unsubInternal(SubscriberPtr subscriber);
 
    void
    unsubInternal(ripple::AccountID const& account, SubscriberPtr subscriber);
 
    void
    unsubProposedInternal(SubscriberPtr subscriber);
 
    void
    unsubProposedInternal(ripple::AccountID const& account, SubscriberPtr subscriber);
 
    void
    unsubInternal(ripple::Book const& book, SubscriberPtr subscriber);
};
}  // namespace feed::impl