refactor: implement fdv2 polling initializer / synchronizer

libs/internal/include/launchdarkly/async/promise.hpp

@@ -437,6 +437,10 @@ class Future {
     std::shared_ptr<PromiseInternal<T>> internal_;
 };
 
+// An executor that runs work immediately on the calling thread. Pass this
+// to Then() when no specific thread is required for the continuation.
+inline auto const kInlineExecutor = [](Continuation<void()> f) { f(); };
+
 // WhenAll takes a variadic list of Futures (each with potentially different
 // value types) and returns a Future<std::monostate> that resolves once all
 // of the input futures have resolved. The result carries no value; callers
@@ -528,4 +532,16 @@ Future<std::size_t> WhenAny(Future<Ts>... futures) {
     return result;
 }
 
+// MakeFuture returns an already-resolved Future<T>. Useful in flattening Then
+// continuations where some branches produce a result immediately and others
+// return a Future, requiring a uniform Future<T> return type across all
+// branches.
+template <typename T>
+Future<T> MakeFuture(T value) {
+    Promise<T> p;
+    auto f = p.GetFuture();
+    p.Resolve(std::move(value));
+    return f;
+}
+
 }  // namespace launchdarkly::async

libs/internal/include/launchdarkly/async/timer.hpp

@@ -0,0 +1,32 @@
+#pragma once
+
+#include <launchdarkly/async/promise.hpp>
+
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/error.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/system/error_code.hpp>
+
+#include <chrono>
+#include <memory>
+
+namespace launchdarkly::async {
+
+// Returns a Future<bool> that resolves once the given duration elapses.
+// The future resolves with true if the timer fired normally, or false if
+// the timer was cancelled before it expired.
+template <typename Rep, typename Period>
+Future<bool> Delay(boost::asio::any_io_executor executor,
+                   std::chrono::duration<Rep, Period> duration) {
+    auto timer = std::make_shared<boost::asio::steady_timer>(executor);
+    timer->expires_after(duration);
+    Promise<bool> promise;
+    auto future = promise.GetFuture();
+    timer->async_wait([p = std::move(promise),
+                       timer](boost::system::error_code code) mutable {
+        p.Resolve(code != boost::asio::error::operation_aborted);
+    });
+    return future;
+}
+
+}  // namespace launchdarkly::async

libs/internal/include/launchdarkly/fdv2_protocol_handler.hpp

@@ -0,0 +1,114 @@
+#pragma once
+
+#include <launchdarkly/data_model/fdv2_change.hpp>
+#include <launchdarkly/serialization/json_fdv2_events.hpp>
+
+#include <boost/json/value.hpp>
+
+#include <string_view>
+#include <variant>
+#include <vector>
+
+namespace launchdarkly {
+
+/**
+ * Protocol state machine for the FDv2 wire format.
+ *
+ * Accumulates put-object and delete-object events between a server-intent
+ * and payload-transferred event, then emits a complete FDv2ChangeSet.
+ *
+ * Shared between the polling and streaming synchronizers.
+ */
+class FDv2ProtocolHandler {
+   public:
+    /**
+     * Typed error returned by HandleEvent. Carries the original underlying
+     * error context rather than converting to a plain string.
+     */
+    struct Error {
+        enum class Kind {
+            kJsonError,      // Failed to deserialise an event's data field.
+            kProtocolError,  // Out-of-order or unexpected event.
+            kServerError,    // Server sent a valid 'error' event.
+        };
+
+        Kind kind;
+        std::string message;
+
+        /**
+         * Set for kJsonError when the tl::expected parse returned an error.
+         * Nullopt when parse succeeded but the data value was null.
+         */
+        std::optional<JsonError> json_error;
+
+        /**
+         * Set for kServerError: the full wire error including id and reason.
+         */
+        std::optional<FDv2Error> server_error;
+
+        /** JSON deserialisation failed — carries the original JsonError. */
+        static Error JsonParseError(JsonError err, std::string msg) {
+            return {Kind::kJsonError, std::move(msg), err, std::nullopt};
+        }
+        /** Parse succeeded but data was null — no underlying JsonError. */
+        static Error JsonParseError(std::string msg) {
+            return {Kind::kJsonError, std::move(msg), std::nullopt,
+                    std::nullopt};
+        }
+        /** Out-of-order or unexpected protocol event. */
+        static Error ProtocolError(std::string msg) {
+            return {Kind::kProtocolError, std::move(msg), std::nullopt,
+                    std::nullopt};
+        }
+        /** Server sent a well-formed 'error' event. */
+        static Error ServerError(FDv2Error err) {
+            return {Kind::kServerError, err.reason, std::nullopt,
+                    std::move(err)};
+        }
+    };
+
+    /**
+     * Result of handling a single FDv2 event:
+     * - monostate: no output yet (accumulating, heartbeat, or unknown event)
+     * - FDv2ChangeSet: complete changeset ready to apply
+     * - Error: protocol error (JSON parse failure, protocol violation, or
+     *          server-sent error event)
+     * - Goodbye: server is closing; caller should rotate sources
+     */
+    using Result =
+        std::variant<std::monostate, data_model::FDv2ChangeSet, Error, Goodbye>;
+
+    /**
+     * Process one FDv2 event.
+     *
+     * @param event_type The event type string (e.g. "server-intent",
+     *                   "put-object", "payload-transferred").
+     * @param data       The parsed JSON value for the event's data field.
+     * @return           A Result indicating what (if anything) the caller
+     *                   should act on.
+     */
+    Result HandleEvent(std::string_view event_type,
+                       boost::json::value const& data);
+
+    /**
+     * Reset accumulated state. Call on reconnect before processing new events.
+     */
+    void Reset();
+
+    FDv2ProtocolHandler() = default;
+
+   private:
+    enum class State { kInactive, kFull, kPartial };
+
+    Result HandleServerIntent(boost::json::value const& data);
+    Result HandlePutObject(boost::json::value const& data);
+    Result HandleDeleteObject(boost::json::value const& data);
+    Result HandlePayloadTransferred(boost::json::value const& data);
+    Result HandleError(boost::json::value const& data);
+    Result HandleGoodbye(boost::json::value const& data);
+
+    State state_ = State::kInactive;
+    std::vector<data_model::FDv2Change> changes_;
+};
+
+}  // namespace launchdarkly

libs/internal/include/launchdarkly/network/asio_requester.hpp

@@ -282,23 +282,17 @@ class AsioRequester {
     }
 
     template <typename CompletionToken>
-    auto Request(HttpRequest request, CompletionToken&& token) {
-        // TODO: Clang-tidy wants to pass the request by reference, but I am not
-        // confident that lifetime would make sense.
-
-        namespace asio = boost::asio;
-        namespace system = boost::system;
-
-        using Sig = void(HttpResult result);
-        using Result = asio::async_result<std::decay_t<CompletionToken>, Sig>;
-        using Handler = typename Result::completion_handler_type;
-
-        Handler handler(std::forward<decltype(token)>(token));
-        Result result(handler);
-
-        InnerRequest(net::make_strand(ctx_), request, std::move(handler), 0);
-
-        return result.get();
+    auto Request(HttpRequest request, CompletionToken&& token) const {
+        return boost::asio::async_initiate<CompletionToken, void(HttpResult)>(
+            [this](auto handler, HttpRequest req) {
+                InnerRequest(
+                    net::make_strand(ctx_), std::move(req),
+                    [h = std::move(handler)](HttpResult result) mutable {
+                        std::move(h)(std::move(result));
+                    },
+                    0);
+            },
+            token, std::move(request));
     }
 
    private:
@@ -313,7 +307,7 @@ class AsioRequester {
     void InnerRequest(boost::asio::any_io_executor exec,
                       std::optional<HttpRequest> request,
                       std::function<void(HttpResult)> callback,
-                      unsigned char redirect_count) {
+                      unsigned char redirect_count) const {
         if (redirect_count > kRedirectLimit) {
             boost::asio::post(exec, [callback, request]() mutable {
                 callback(
@@ -336,7 +330,7 @@ class AsioRequester {
                                  redirect_count]() mutable {
             auto beast_request = MakeBeastRequest(*request);
 
-            const auto& properties = request->Properties();
+            auto const& properties = request->Properties();
 
             std::string service =
                 request->Port().value_or(request->Https() ? "https" : "http");

libs/internal/include/launchdarkly/network/requester.hpp

@@ -1,10 +1,10 @@
 #pragma once
 
-#include "http_requester.hpp"
-#include <launchdarkly/config/shared/built/http_properties.hpp>
+#include <boost/asio/any_io_executor.hpp>
 #include <functional>
+#include <launchdarkly/config/shared/built/http_properties.hpp>
 #include <memory>
-#include <boost/asio/any_io_executor.hpp>
+#include "http_requester.hpp"
 
 namespace launchdarkly::network {
 
@@ -15,30 +15,32 @@ using TlsOptions = config::shared::built::TlsOptions;
 class IRequesterImpl;
 
 /**
- * Requester provides HTTP request functionality using either CURL or Boost.Beast
- * depending on the LD_CURL_NETWORKING compile-time flag.
+ * Requester provides HTTP request functionality using either CURL or
+ * Boost.Beast depending on the LD_CURL_NETWORKING compile-time flag.
  *
  * When LD_CURL_NETWORKING is ON: Uses CurlRequester (CURL-based implementation)
- * When LD_CURL_NETWORKING is OFF: Uses AsioRequester (Boost.Beast-based implementation)
+ * When LD_CURL_NETWORKING is OFF: Uses AsioRequester (Boost.Beast-based
+ * implementation)
  *
- * The implementation choice is made at library compile-time and hidden from users
- * via the pimpl idiom to avoid ABI issues.
+ * The implementation choice is made at library compile-time and hidden from
+ * users via the pimpl idiom to avoid ABI issues.
  */
 class Requester {
-public:
+   public:
     Requester(net::any_io_executor ctx, TlsOptions const& tls_options);
     ~Requester();
 
     // Move-only type
     Requester(Requester&&) noexcept;
     Requester& operator=(Requester&&) noexcept;
-    Requester(const Requester&) = delete;
-    Requester& operator=(const Requester&) = delete;
+    Requester(Requester const&) = delete;
+    Requester& operator=(Requester const&) = delete;
 
-    void Request(HttpRequest request, std::function<void(const HttpResult&)> cb);
+    void Request(HttpRequest request,
+                 std::function<void(HttpResult const&)> cb) const;
 
-private:
+   private:
     std::unique_ptr<IRequesterImpl> impl_;
 };
 
-} // namespace launchdarkly::network
+}  // namespace launchdarkly::network

libs/internal/src/CMakeLists.txt

@@ -36,6 +36,7 @@ set(INTERNAL_SOURCES
         serialization/value_mapping.cpp
         serialization/json_evaluation_result.cpp
         serialization/json_fdv2_events.cpp
+        fdv2_protocol_handler.cpp
         serialization/json_sdk_data_set.cpp
         serialization/json_segment.cpp
         serialization/json_primitives.cpp