InfoSub.h

#pragma once
 
#include <xrpl/basics/CountedObject.h>
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/json/json_value.h>
#include <xrpl/protocol/AccountID.h>
#include <xrpl/protocol/Book.h>
#include <xrpl/protocol/ErrorCodes.h>
#include <xrpl/protocol/UintTypes.h>
#include <xrpl/resource/Consumer.h>
#include <xrpl/server/Manifest.h>
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
 
namespace xrpl {
 
// Operations that clients may wish to perform against the network
// Master operational handler, server sequencer, network tracker

constexpr std::size_t kMaxSubscriptionsPerConnection = 100'000;

[[nodiscard]] constexpr bool
exceedsSubscriptionCap(
    std::size_t current,
    std::size_t additional,
    std::size_t cap = kMaxSubscriptionsPerConnection)
{
    return additional > cap || current > cap - additional;
}
 
class InfoSubRequest : public CountedObject<InfoSubRequest>
{
public:
    using pointer = std::shared_ptr<InfoSubRequest>;
 
    virtual ~InfoSubRequest() = default;
 
    virtual json::Value
    doClose() = 0;
    virtual json::Value
    doStatus(json::Value const&) = 0;
};

class InfoSub : public CountedObject<InfoSub>
{
public:
    using pointer = std::shared_ptr<InfoSub>;
 
    // VFALCO TODO Standardize on the names of weak / strong pointer type
    // aliases.
    using Wptr = std::weak_ptr<InfoSub>;
 
    using Ref = std::shared_ptr<InfoSub> const&;
 
    using Consumer = resource::Consumer;
 
public:
    class Source
    {
    public:
        virtual ~Source() = default;
 
        // For some reason, these were originally called "rt"
        // for "real time". They actually refer to whether
        // you get transactions as they occur or once their
        // results are confirmed
        virtual void
        subAccount(Ref ispListener, HashSet<AccountID> const& vnaAccountIDs, bool realTime) = 0;
 
        // for normal use, removes from InfoSub and server
        virtual void
        unsubAccount(Ref isplistener, HashSet<AccountID> const& vnaAccountIDs, bool realTime) = 0;
 
        // for use during InfoSub destruction
        // Removes only from the server
        virtual void
        unsubAccountInternal(
            std::uint64_t uListener,
            HashSet<AccountID> const& vnaAccountIDs,
            bool realTime) = 0;

        virtual ErrorCodeI
        subAccountHistory(Ref ispListener, AccountID const& account) = 0;

        virtual void
        unsubAccountHistory(Ref ispListener, AccountID const& account, bool historyOnly) = 0;
 
        virtual void
        unsubAccountHistoryInternal(
            std::uint64_t uListener,
            AccountID const& account,
            bool historyOnly) = 0;

        virtual void
        unsubMPTInternal(std::uint64_t uListener, MPTID const& mptID) = 0;

        virtual void
        scheduleAccountCleanup(
            std::uint64_t seq,
            HashSet<AccountID> rtAccounts,
            HashSet<AccountID> normalAccounts,
            HashSet<AccountID> historyAccounts) = 0;
 
        // VFALCO TODO Document the bool return value
        virtual bool
        subLedger(Ref ispListener, json::Value& jvResult) = 0;
        virtual bool
        unsubLedger(std::uint64_t uListener) = 0;
 
        virtual bool
        subBookChanges(Ref ispListener) = 0;
        virtual bool
        unsubBookChanges(std::uint64_t uListener) = 0;
 
        virtual bool
        subManifests(Ref ispListener) = 0;
        virtual bool
        unsubManifests(std::uint64_t uListener) = 0;
        virtual void
        pubManifest(Manifest const&) = 0;
 
        virtual bool
        subServer(Ref ispListener, json::Value& jvResult, bool admin) = 0;
        virtual bool
        unsubServer(std::uint64_t uListener) = 0;
 
        virtual bool
        subBook(Ref ispListener, Book const&) = 0;

        virtual bool
        unsubBook(Ref ispListener, Book const&) = 0;

        virtual bool
        unsubBookInternal(std::uint64_t uListener, Book const&) = 0;
 
        virtual bool
        subTransactions(Ref ispListener) = 0;
        virtual bool
        unsubTransactions(std::uint64_t uListener) = 0;
 
        virtual bool
        subRTTransactions(Ref ispListener) = 0;
        virtual bool
        unsubRTTransactions(std::uint64_t uListener) = 0;
 
        virtual bool
        subValidations(Ref ispListener) = 0;
        virtual bool
        unsubValidations(std::uint64_t uListener) = 0;
 
        virtual bool
        subPeerStatus(Ref ispListener) = 0;
 
        virtual bool
        unsubPeerStatus(std::uint64_t uListener) = 0;
        virtual void
        pubPeerStatus(std::function<json::Value(void)> const&) = 0;
 
        virtual bool
        subConsensus(Ref ispListener) = 0;
        virtual bool
        unsubConsensus(std::uint64_t uListener) = 0;
 
        virtual void
        subMPT(InfoSub::Ref ispListener, HashSet<MPTID> const& mptIDs) = 0;
        virtual void
        unsubMPT(InfoSub::Ref ispListener, HashSet<MPTID> const& mptIDs) = 0;
 
        // VFALCO TODO Remove
        //             This was added for one particular partner, it
        //             "pushes" subscription data to a particular URL.
        //
        virtual pointer
        findRpcSub(std::string const& strUrl) = 0;
        virtual pointer
        addRpcSub(std::string const& strUrl, Ref rspEntry) = 0;
        virtual bool
        tryRemoveRpcSub(std::string const& strUrl) = 0;

        [[nodiscard]] virtual beast::Journal const&
        journal() const = 0;
    };
 
public:
    InfoSub(Source& source);
    InfoSub(Source& source, Consumer consumer);
 
    virtual ~InfoSub();
 
    Consumer&
    getConsumer();
 
    virtual void
    send(json::Value const& jvObj, bool broadcast) = 0;
 
    [[nodiscard]] std::uint64_t
    getSeq() const;

    [[nodiscard]] std::size_t
    totalSubscriptionCount() const;

    [[nodiscard]] bool
    tryReserveAccountSubscriptions(
        HashSet<AccountID> const& proposedAccounts,
        HashSet<AccountID> const& normalAccounts,
        std::size_t cap);

    [[nodiscard]] bool
    tryReserveMPTSubscriptions(HashSet<MPTID> const& mptIDs, std::size_t cap);

    [[nodiscard]] bool
    hasAccountHistorySubscription(AccountID const& account) const;
 
    void
    onSendEmpty();
 
    void
    insertSubAccountInfo(AccountID const& account, bool rt);
 
    void
    deleteSubAccountInfo(AccountID const& account, bool rt);

    void
    insertBookSubscription(Book const& book);

    void
    deleteBookSubscription(Book const& book);
 
    // return false if already subscribed to this account
    bool
    insertSubAccountHistory(AccountID const& account);
 
    void
    deleteSubAccountHistory(AccountID const& account);
 
    void
    clearRequest();
 
    void
    setRequest(std::shared_ptr<InfoSubRequest> const& req);
 
    std::shared_ptr<InfoSubRequest> const&
    getRequest();
 
    void
    setApiVersion(unsigned int apiVersion);
 
    [[nodiscard]] unsigned int
    getApiVersion() const noexcept;
 
    void
    insertSubMPTInfo(MPTID const& mptID);
 
    void
    deleteSubMPTInfo(MPTID const& mptID);
 
protected:
    // Mutable so the read-only totalSubscriptionCount() accessor can lock it
    // from a const method; locking semantics are otherwise unchanged.
    mutable std::mutex lock_;
 
private:
    // The lock type guarding this instance's subscription sets.
    using ScopedLock = std::scoped_lock<decltype(lock_)>;

    [[nodiscard]] std::size_t
    subscriptionCount(ScopedLock const& lock) const;
 
    Consumer consumer_;
    Source& source_;
    HashSet<AccountID> realTimeSubscriptions_;
    HashSet<AccountID> normalSubscriptions_;
    std::shared_ptr<InfoSubRequest> request_;
    std::uint64_t seq_;
    HashSet<AccountID> accountHistorySubscriptions_;
    HashSet<Book> bookSubscriptions_;
    HashSet<MPTID> mptSubscriptions_;
    unsigned int apiVersion_ = 0;
 
    static int
    assignId()
    {
        static std::atomic<std::uint64_t> kID(0);
        return ++kID;
    }
};
 
}  // namespace xrpl