polling_synchronizer.hpp

#pragma once
 
#include "../../data_interfaces/source/ifdv2_synchronizer.hpp"
 
#include <launchdarkly/async/cancellation.hpp>
#include <launchdarkly/logging/logger.hpp>
#include <launchdarkly/network/requester.hpp>
#include <launchdarkly/server_side/config/built/all_built.hpp>
 
#include <boost/asio/any_io_executor.hpp>
 
#include <chrono>
#include <mutex>
#include <optional>
#include <string>
 
namespace launchdarkly::server_side::data_systems {
 
class FDv2PollingSynchronizer final
    : public data_interfaces::IFDv2Synchronizer {
   public:
    FDv2PollingSynchronizer(
        boost::asio::any_io_executor const& executor,
        Logger const& logger,
        std::string polling_base_url,
        config::built::HttpProperties const& http_properties,
        std::optional<std::string> filter_key,
        std::chrono::seconds poll_interval);
 
    ~FDv2PollingSynchronizer() override;
 
    async::Future<data_interfaces::FDv2SourceResult> Next(
        data_model::Selector selector) override;
 
    void Close() override;
 
    [[nodiscard]] std::string const& Identity() const override;
 
   private:
    // Any state that may be accessed by async callbacks needs to be inside this
    // class, managed by a shared_ptr. All mutable members are guarded by the
    // mutex.
    class State {
       public:
        State(Logger logger,
              boost::asio::any_io_executor const& executor,
              std::chrono::seconds poll_interval,
              std::string polling_base_url,
              config::built::HttpProperties const& http_properties,
              std::optional<std::string> filter_key);
 
        async::Future<network::HttpResult> Request(
            data_model::Selector const& selector) const;
 
        data_interfaces::FDv2SourceResult HandlePollResult(
            network::HttpResult const& res);
 
        async::Future<bool> CreatePollDelayFuture(
            async::CancellationToken token = {});
 
        void RecordPollStarted();
 
        async::Future<bool> Delay(std::chrono::nanoseconds duration,
                                  async::CancellationToken token = {});
 
       private:
        // Logger is itself thread-safe.
        Logger logger_;
 
        // Immutable state
        std::chrono::seconds const poll_interval_;
        std::string const polling_base_url_;
        config::built::HttpProperties const http_properties_;
        std::optional<std::string> const filter_key_;
        network::Requester const requester_;
        boost::asio::any_io_executor const executor_;
 
        // Mutable state, guarded by mutex_.
        std::mutex mutex_;
        std::optional<std::chrono::time_point<std::chrono::steady_clock>>
            last_poll_start_;
    };
 
    static async::Future<data_interfaces::FDv2SourceResult> DoNext(
        std::shared_ptr<State> state,
        async::Future<std::monostate> closed,
        data_model::Selector selector);
 
    static async::Future<data_interfaces::FDv2SourceResult> DoPoll(
        std::shared_ptr<State> state,
        async::Future<std::monostate> closed,
        data_model::Selector const& selector);
 
    // Resolved by Close() or on destruction, cancelling any outstanding Next()
    // calls.
    async::Promise<std::monostate> close_promise_;
 
    // Shared with async callbacks.
    std::shared_ptr<State> state_;
};
 
}  // namespace launchdarkly::server_side::data_systems