SubscriptionSource.hpp

#pragma once
 
#include "etl/NetworkValidatedLedgersInterface.hpp"
#include "etl/Source.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Mutex.hpp"
#include "util/Retry.hpp"
#include "util/StopHelper.hpp"
#include "util/log/Logger.hpp"
#include "util/prometheus/Gauge.hpp"
#include "util/requests/Types.hpp"
#include "util/requests/WsConnection.hpp"
 
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/http/field.hpp>
#include <fmt/format.h>
 
#include <atomic>
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>
 
namespace etl::impl {

class SubscriptionSource {
public:
    using OnConnectHook = SourceBase::OnConnectHook;
    using OnDisconnectHook = SourceBase::OnDisconnectHook;
    using OnLedgerClosedHook = SourceBase::OnLedgerClosedHook;
 
private:
    util::Logger log_;
    util::requests::WsConnectionBuilder wsConnectionBuilder_;
    util::requests::WsConnectionPtr wsConnection_;
 
    struct ValidatedLedgersData {
        std::vector<std::pair<uint32_t, uint32_t>> validatedLedgers;
        std::string validatedLedgersRaw{"N/A"};
    };
    util::Mutex<ValidatedLedgersData> validatedLedgersData_;
 
    std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers_;
    std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
 
    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
 
    std::chrono::steady_clock::duration wsTimeout_;
 
    util::Retry retry_;
 
    OnConnectHook onConnect_;
    OnDisconnectHook onDisconnect_;
    OnLedgerClosedHook onLedgerClosed_;
 
    std::atomic_bool isConnected_{false};
    std::atomic_bool stop_{false};
    std::atomic_bool isForwarding_{false};
 
    util::Mutex<std::chrono::steady_clock::time_point> lastMessageTime_;
 
    std::reference_wrapper<util::prometheus::GaugeInt> lastMessageTimeSecondsSinceEpoch_;
 
    util::StopHelper stopHelper_;
 
    static constexpr std::chrono::seconds kWS_TIMEOUT{30};
    static constexpr std::chrono::seconds kRETRY_MAX_DELAY{30};
    static constexpr std::chrono::seconds kRETRY_DELAY{1};
 
public:
    SubscriptionSource(
        boost::asio::io_context& ioContext,
        std::string const& ip,
        std::string const& wsPort,
        std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
        OnConnectHook onConnect,
        OnDisconnectHook onDisconnect,
        OnLedgerClosedHook onLedgerClosed,
        std::chrono::steady_clock::duration const wsTimeout = SubscriptionSource::kWS_TIMEOUT,
        std::chrono::steady_clock::duration const retryDelay = SubscriptionSource::kRETRY_DELAY
    );

    void
    run();

    bool
    hasLedger(uint32_t sequence) const;

    bool
    isConnected() const;

    bool
    isForwarding() const;

    void
    setForwarding(bool isForwarding);

    std::chrono::steady_clock::time_point
    lastMessageTime() const;

    std::string const&
    validatedRange() const;

    void
    stop(boost::asio::yield_context yield);
 
private:
    void
    subscribe();
 
    std::optional<util::requests::RequestError>
    handleMessage(std::string const& message);
 
    void
    handleError(util::requests::RequestError const& error, boost::asio::yield_context yield);
 
    void
    logError(util::requests::RequestError const& error) const;
 
    void
    setLastMessageTime();
 
    void
    setValidatedRange(std::string range);
 
    static std::string const&
    getSubscribeCommandJson();
};
 
}  // namespace etl::impl