SourceImpl.hpp

#pragma once
 
#include "etl/InitialLoadObserverInterface.hpp"
#include "etl/LoadBalancerInterface.hpp"
#include "etl/Source.hpp"
#include "etl/impl/ForwardingSource.hpp"
#include "etl/impl/GrpcSource.hpp"
#include "etl/impl/SubscriptionSource.hpp"
#include "rpc/Errors.hpp"
 
#include <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <grpcpp/support/status.h>
#include <org/xrpl/rpc/v1/get_ledger.pb.h>
 
#include <chrono>
#include <cstdint>
#include <expected>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
 
namespace etl::impl {

template <
    typename GrpcSourceType = GrpcSource,
    typename SubscriptionSourceTypePtr = std::unique_ptr<impl::SubscriptionSource>,
    typename ForwardingSourceType = impl::ForwardingSource>
class SourceImpl : public SourceBase {
    std::string ip_;
    std::string wsPort_;
    std::string grpcPort_;
 
    GrpcSourceType grpcSource_;
    SubscriptionSourceTypePtr subscriptionSource_;
    ForwardingSourceType forwardingSource_;
 
public:
    template <typename SomeGrpcSourceType, typename SomeForwardingSourceType>
        requires std::is_same_v<GrpcSourceType, SomeGrpcSourceType> and
                     std::is_same_v<ForwardingSourceType, SomeForwardingSourceType>
    SourceImpl(
        std::string ip,
        std::string wsPort,
        std::string grpcPort,
        SomeGrpcSourceType&& grpcSource,
        SubscriptionSourceTypePtr subscriptionSource,
        SomeForwardingSourceType&& forwardingSource
    )
        : ip_(std::move(ip))
        , wsPort_(std::move(wsPort))
        , grpcPort_(std::move(grpcPort))
        , grpcSource_(std::forward<SomeGrpcSourceType>(grpcSource))
        , subscriptionSource_(std::move(subscriptionSource))
        , forwardingSource_(std::forward<SomeForwardingSourceType>(forwardingSource))
    {
    }

    void
    run() final
    {
        subscriptionSource_->run();
    }
 
    void
    stop(boost::asio::yield_context yield) final
    {
        subscriptionSource_->stop(yield);
        grpcSource_.stop(yield);
    }

    [[nodiscard]] bool
    isConnected() const final
    {
        return subscriptionSource_->isConnected();
    }

    void
    setForwarding(bool isForwarding) final
    {
        subscriptionSource_->setForwarding(isForwarding);
    }

    [[nodiscard]] boost::json::object
    toJson() const final
    {
        boost::json::object res;
 
        res["validated_range"] = subscriptionSource_->validatedRange();
        res["is_connected"] = std::to_string(static_cast<int>(subscriptionSource_->isConnected()));
        res["ip"] = ip_;
        res["ws_port"] = wsPort_;
        res["grpc_port"] = grpcPort_;
 
        auto last = subscriptionSource_->lastMessageTime();
        if (last.time_since_epoch().count() != 0) {
            res["last_msg_age_seconds"] = std::to_string(
                std::chrono::duration_cast<std::chrono::seconds>(
                    std::chrono::steady_clock::now() - last
                )
                    .count()
            );
        }
 
        return res;
    }

    [[nodiscard]] std::string
    toString() const final
    {
        return "{validated range: " + subscriptionSource_->validatedRange() + ", ip: " + ip_ +
            ", web socket port: " + wsPort_ + ", grpc port: " + grpcPort_ + "}";
    }

    [[nodiscard]] bool
    hasLedger(uint32_t sequence) const final
    {
        return subscriptionSource_->hasLedger(sequence);
    }

    std::pair<grpc::Status, org::xrpl::rpc::v1::GetLedgerResponse>
    fetchLedger(uint32_t sequence, bool getObjects = true, bool getObjectNeighbors = false) final
    {
        return grpcSource_.fetchLedger(sequence, getObjects, getObjectNeighbors);
    }

    InitialLedgerLoadResult
    loadInitialLedger(
        uint32_t sequence,
        std::uint32_t numMarkers,
        InitialLoadObserverInterface& loader
    ) final
    {
        return grpcSource_.loadInitialLedger(sequence, numMarkers, loader);
    }

    [[nodiscard]] std::expected<boost::json::object, rpc::ClioError>
    forwardToRippled(
        boost::json::object const& request,
        std::optional<std::string> const& forwardToRippledClientIp,
        std::string_view xUserValue,
        boost::asio::yield_context yield
    ) const final
    {
        return forwardingSource_.forwardToRippled(
            request, forwardToRippledClientIp, xUserValue, yield
        );
    }
};
 
}  // namespace etl::impl