streaming_synchronizer.hpp

#pragma once
 
#include "../../data_interfaces/source/ifdv2_synchronizer.hpp"
 
#include <launchdarkly/async/cancellation.hpp>
#include <launchdarkly/async/promise.hpp>
#include <launchdarkly/fdv2_protocol_handler.hpp>
#include <launchdarkly/logging/logger.hpp>
#include <launchdarkly/server_side/config/built/all_built.hpp>
#include <launchdarkly/sse/client.hpp>
 
#include <boost/asio/any_io_executor.hpp>
#include <boost/url/url.hpp>
 
#include <chrono>
#include <deque>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
 
namespace launchdarkly::server_side::data_systems {
 
class FDv2StreamingSynchronizerTestPeer;
 
class FDv2StreamingSynchronizer final
    : public data_interfaces::IFDv2Synchronizer {
    friend class FDv2StreamingSynchronizerTestPeer;
 
   public:
    FDv2StreamingSynchronizer(
        boost::asio::any_io_executor const& executor,
        Logger const& logger,
        std::string streaming_base_url,
        config::built::HttpProperties const& http_properties,
        std::optional<std::string> filter_key,
        std::chrono::milliseconds initial_reconnect_delay);
 
    ~FDv2StreamingSynchronizer() 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 SSE callbacks needs to be inside
    // this class, managed by a shared_ptr. All mutable members are guarded by
    // the mutex.
    class State {
        friend class FDv2StreamingSynchronizerTestPeer;
 
       public:
        State(Logger logger,
              boost::asio::any_io_executor const& executor,
              std::string streaming_base_url,
              config::built::HttpProperties const& http_properties,
              std::optional<std::string> filter_key,
              std::chrono::milliseconds initial_reconnect_delay);
 
        async::Future<data_interfaces::FDv2SourceResult> Next(
            data_model::Selector const& selector,
            std::shared_ptr<State> self);
 
        void ClearPendingPromise();
 
        void Shutdown();
 
       private:
        using HttpRequest =
            boost::beast::http::request<boost::beast::http::string_body>;
        using HttpResponseHeader = boost::beast::http::response_header<>;
 
        void EnsureStarted(data_model::Selector const& selector,
                           std::shared_ptr<State> self);
 
        void Notify(data_interfaces::FDv2SourceResult result);
 
        // SSE client callbacks.
        void OnConnect(HttpRequest* req);
        void OnResponse(HttpResponseHeader const& headers);
        void OnEvent(sse::Event const& event);
        void OnError(sse::Error const& error);
 
        // Logger is itself thread-safe.
        Logger logger_;
 
        // Immutable state
        std::string const streaming_base_url_;
        config::built::HttpProperties const http_properties_;
        std::optional<std::string> const filter_key_;
        std::chrono::milliseconds const initial_reconnect_delay_;
        boost::asio::any_io_executor const executor_;
 
        // Touched only from SSE callbacks, which all run on the same strand.
        // No lock required.
        FDv2ProtocolHandler protocol_handler_;
 
        // Mutable state, all guarded by mutex_.
        std::mutex mutex_;
        bool started_ = false;
        bool closed_ = false;
        // FDv1 fallback directive from the most recent SSE response.
        std::optional<data_interfaces::FDv1FallbackDirective>
            latest_fdv1_fallback_;
        data_model::Selector latest_selector_;
        std::optional<boost::urls::url> base_url_;
        std::shared_ptr<sse::Client> sse_client_;
        std::optional<async::Promise<data_interfaces::FDv2SourceResult>>
            pending_promise_;
        std::deque<data_interfaces::FDv2SourceResult> result_queue_;
    };
 
    // Resolved by Close() or on destruction, cancelling any outstanding Next().
    async::Promise<std::monostate> close_promise_;
 
    // Shared with async SSE callbacks.
    std::shared_ptr<State> state_;
};
 
}  // namespace launchdarkly::server_side::data_systems