feat: Add the FDv2 connection modes to the client

libs/client-sdk/src/CMakeLists.txt

@@ -23,6 +23,9 @@ target_sources(${LIBNAME} PRIVATE
         data_sources/fdv2/streaming_synchronizer.cpp
         data_sources/fdv2/fdv2_data_source.cpp
         data_sources/fdv2/cache_initializer.cpp
+        data_sources/fdv2/source_factories.cpp
+        data_sources/fdv2/mode_sources.cpp
+        data_sources/fdv2/fdv1_adapter_synchronizer.cpp
         data_sources/data_source_event_handler.cpp
         data_sources/polling_data_source.cpp
         flag_manager/flag_store.cpp
@@ -51,6 +54,9 @@ target_sources(${LIBNAME} PRIVATE
         data_sources/fdv2/streaming_synchronizer.hpp
         data_sources/fdv2/fdv2_data_source.hpp
         data_sources/fdv2/cache_initializer.hpp
+        data_sources/fdv2/source_factories.hpp
+        data_sources/fdv2/mode_sources.hpp
+        data_sources/fdv2/fdv1_adapter_synchronizer.hpp
         flag_manager/flag_store.hpp
         flag_manager/flag_updater.hpp
         bindings/c/sdk.cpp

libs/client-sdk/src/client_impl.cpp

@@ -1,3 +1,3 @@
 #include "client_impl.hpp"
 #include "data_sources/null_data_source.hpp"
 #include "data_sources/polling_data_source.hpp"
@@ -160,6 +160,10 @@
 
 std::future<bool> ClientImpl::IdentifyAsync(Context context) {
     UpdateContextSynchronized(context);
+    // A selector describes one context's data, so it is never carried over.
+    // Any flag data already loaded stays available for evaluation until a
+    // full data set arrives for the new context.
+    flag_manager_.ClearSelector();
     flag_manager_.LoadCache(context);
     event_processor_->SendAsync(events::IdentifyEventParams{
         std::chrono::system_clock::now(), std::move(context)});
@@ -211,7 +215,7 @@
     std::unordered_map<Client::FlagKey, Value> result;
     for (auto& [key, descriptor] : flag_manager_.Store().GetAll()) {
         if (descriptor->item) {
             result.try_emplace(key, descriptor->item->Detail().Value());
         }
     }
     return result;
@@ -244,7 +248,7 @@
 }
 
 template <typename T>
 EvaluationDetail<T> ClientImpl::VariationInternal(FlagKey const& key,
                                                   Value default_value,
                                                   bool check_type,
                                                   bool detailed,
@@ -301,7 +305,7 @@
 
     LD_ASSERT(desc->item);
 
     auto const& flag = *(desc->item);
     auto const& detail = flag.Detail();
 
     // The Prerequisites vector represents the evaluated prerequisites of

libs/client-sdk/src/data_sources/fdv2/fdv1_adapter_synchronizer.cpp

@@ -0,0 +1,206 @@
+#include "fdv1_adapter_synchronizer.hpp"
+
+#include <utility>
+
+namespace launchdarkly::client_side::data_sources {
+
+using DataSourceState = DataSourceStatus::DataSourceState;
+
+// ----- State -----
+
+FDv1AdapterSynchronizer::State::State(async::Future<std::monostate> closed)
+    : closed_(std::move(closed)) {}
+
+async::Future<FDv2SourceResult> FDv1AdapterSynchronizer::State::GetNext() {
+    std::lock_guard lock(mutex_);
+    if (!result_queue_.empty()) {
+        auto result = std::move(result_queue_.front());
+        result_queue_.pop_front();
+        return async::MakeFuture(std::move(result));
+    }
+    return pending_promise_.emplace().GetFuture();
+}
+
+void FDv1AdapterSynchronizer::State::ResolvePendingAsShutdown() {
+    std::optional<async::Promise<FDv2SourceResult>> promise;
+    {
+        std::lock_guard lock(mutex_);
+        if (pending_promise_) {
+            promise = std::move(pending_promise_);
+            pending_promise_.reset();
+        }
+    }
+    if (promise) {
+        promise->Resolve(FDv2SourceResult{FDv2SourceResult::Shutdown{}});
+    }
+}
+
+void FDv1AdapterSynchronizer::State::Notify(FDv2SourceResult result) {
+    std::optional<async::Promise<FDv2SourceResult>> promise;
+    {
+        std::lock_guard lock(mutex_);
+        if (closed_.IsFinished()) {
+            return;
+        }
+        if (pending_promise_) {
+            promise = std::move(pending_promise_);
+            pending_promise_.reset();
+        } else {
+            result_queue_.push_back(std::move(result));
+            return;
+        }
+    }
+    // Resolve outside the lock. Promise::Resolve may invoke inline
+    // continuations that could call back into Notify or GetNext.
+    promise->Resolve(std::move(result));
+}
+
+// ----- ConvertingSink -----
+
+FDv1AdapterSynchronizer::ConvertingSink::ConvertingSink(
+    std::weak_ptr<State> state)
+    : state_(std::move(state)) {}
+
+void FDv1AdapterSynchronizer::ConvertingSink::Init(
+    Context const& /* context */,
+    std::unordered_map<std::string, ItemDescriptor> data) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    FlagChangeSetData changes;
+    changes.reserve(data.size());
+    for (auto& [key, item] : data) {
+        changes.push_back(FlagChange{key, std::move(item)});
+    }
+    state->Notify(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        FlagChangeSet{data_model::ChangeSetType::kFull, std::move(changes),
+                      data_model::Selector{}}}});
+}
+
+void FDv1AdapterSynchronizer::ConvertingSink::Upsert(
+    Context const& /* context */,
+    std::string key,
+    ItemDescriptor item) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    state->Notify(FDv2SourceResult{FDv2SourceResult::ChangeSet{
+        FlagChangeSet{data_model::ChangeSetType::kPartial,
+                      {FlagChange{std::move(key), std::move(item)}},
+                      data_model::Selector{}}}});
+}
+
+void FDv1AdapterSynchronizer::ConvertingSink::Apply(
+    Context const& /* context */,
+    FlagChangeSet change_set,
+    bool /* from_cache */) {
+    auto state = state_.lock();
+    if (!state) {
+        return;
+    }
+    state->Notify(
+        FDv2SourceResult{FDv2SourceResult::ChangeSet{std::move(change_set)}});
+}
+
+// ----- FDv1AdapterSynchronizer -----
+
+namespace {
+
+// Turns the wrapped source's status into the result the orchestrator acts on.
+// A valid status carries no error and needs no result. The changeset that
+// accompanied it already reported the recovery.
+std::optional<FDv2SourceResult> ResultForStatus(
+    DataSourceStatus const& status) {
+    auto const error = status.LastError();
+    if (!error) {
+        return std::nullopt;
+    }
+    switch (status.State()) {
+        case DataSourceState::kInterrupted:
+        // An error encountered before the source ever became valid is
+        // reported as still initializing, but it is the same recoverable
+        // failure.
+        case DataSourceState::kInitializing:
+            return FDv2SourceResult{FDv2SourceResult::Interrupted{*error}};
+        case DataSourceState::kShutdown:
+            return FDv2SourceResult{FDv2SourceResult::TerminalError{*error}};
+        case DataSourceState::kValid:
+        case DataSourceState::kSetOffline:
+            return std::nullopt;
+    }
+    return std::nullopt;
+}
+
+}  // namespace
+
+FDv1AdapterSynchronizer::FDv1AdapterSynchronizer(SourceBuilder source_builder)
+    : state_(std::make_shared<State>(close_promise_.GetFuture())),
+      sink_(std::make_shared<ConvertingSink>(state_)),
+      status_manager_(std::make_shared<DataSourceStatusManager>()),
+      status_subscription_(status_manager_->OnDataSourceStatusChange(
+          [state = state_](DataSourceStatus status) {
+              if (auto result = ResultForStatus(status)) {
+                  state->Notify(std::move(*result));
+              }
+          })),
+      fdv1_source_(source_builder(sink_.get(), status_manager_.get())) {}
+
+FDv1AdapterSynchronizer::~FDv1AdapterSynchronizer() {
+    Close();
+}
+
+async::Future<FDv2SourceResult> FDv1AdapterSynchronizer::Next(
+    data_model::Selector /* selector */) {
+    auto closed = close_promise_.GetFuture();
+    if (closed.IsFinished()) {
+        return async::MakeFuture(
+            FDv2SourceResult{FDv2SourceResult::Shutdown{}});
+    }
+    {
+        std::lock_guard lock(lifecycle_mutex_);
+        if (!started_) {
+            started_ = true;
+            fdv1_source_->Start();
+        }
+    }
+    auto result_future = state_->GetNext();
+    if (result_future.IsFinished()) {
+        return result_future;
+    }
+    return async::WhenAny(closed, result_future)
+        .Then(
+            [state = state_, result_future](std::size_t const& idx) mutable
+            -> async::Future<FDv2SourceResult> {
+                if (idx == 0) {
+                    state->ResolvePendingAsShutdown();
+                    return async::MakeFuture(
+                        FDv2SourceResult{FDv2SourceResult::Shutdown{}});
+                }
+                return result_future;
+            },
+            async::kInlineExecutor);
+}
+
+void FDv1AdapterSynchronizer::Close() {
+    if (!close_promise_.Resolve(std::monostate{})) {
+        return;
+    }
+    std::lock_guard lock(lifecycle_mutex_);
+    bool const was_started = started_;
+    started_ = true;
+    if (was_started) {
+        // The sink and status manager are captured so that they outlive any
+        // callback the source has already queued.
+        fdv1_source_->ShutdownAsync(
+            [sink = sink_, status = status_manager_, source = fdv1_source_] {});
+    }
+}
+
+std::string const& FDv1AdapterSynchronizer::Identity() const {
+    static std::string const identity = "FDv1 fallback adapter";
+    return identity;
+}
+
+}  // namespace launchdarkly::client_side::data_sources

libs/client-sdk/src/data_sources/fdv2/fdv1_adapter_synchronizer.hpp

@@ -0,0 +1,128 @@
+#pragma once
+
+#include "../data_source.hpp"
+#include "../data_source_status_manager.hpp"
+#include "../data_source_update_sink.hpp"
+#include "ifdv2_synchronizer.hpp"
+
+#include <launchdarkly/async/promise.hpp>
+#include <launchdarkly/connection.hpp>
+#include <launchdarkly/context.hpp>
+
+#include <deque>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+
+namespace launchdarkly::client_side::data_sources {
+
+/**
+ * Presents an FDv1 data source as an FDv2 synchronizer, so that the
+ * orchestrator can run it while the service has directed the SDK away from
+ * FDv2.
+ *
+ * FDv1 has no selectors, so the changesets this reports carry none. The
+ * orchestrator therefore never asks the service for a delta against data
+ * FDv1 supplied.
+ *
+ * Thread safety: Next() and Close() may be called from any thread. Only one
+ * Next() may be outstanding at a time.
+ */
+class FDv1AdapterSynchronizer final : public IFDv2Synchronizer {
+   public:
+    /**
+     * Builds the wrapped FDv1 source. Called once during construction with
+     * the sink and status manager the source must report through, both of
+     * which the adapter keeps alive for the source's lifetime.
+     */
+    using SourceBuilder =
+        std::function<std::shared_ptr<IDataSource>(IDataSourceUpdateSink*,
+                                                   DataSourceStatusManager*)>;
+
+    explicit FDv1AdapterSynchronizer(SourceBuilder source_builder);
+
+    ~FDv1AdapterSynchronizer() override;
+
+    async::Future<FDv2SourceResult> Next(
+        data_model::Selector selector) override;
+
+    void Close() override;
+
+    [[nodiscard]] std::string const& Identity() const override;
+
+   private:
+    /**
+     * Holds the result queue and the pending Next() promise. Shared with the
+     * wrapped source's sink and status subscription. Thread-safe.
+     */
+    class State {
+       public:
+        explicit State(async::Future<std::monostate> closed);
+
+        async::Future<FDv2SourceResult> GetNext();
+
+        /**
+         * Resolves any pending Next() with Shutdown and clears it, so that a
+         * caller abandoned by Close() is not left waiting.
+         */
+        void ResolvePendingAsShutdown();
+
+        void Notify(FDv2SourceResult result);
+
+       private:
+        // Finished once the owning adapter's Close() has run. Read in Notify
+        // to drop late results.
+        async::Future<std::monostate> const closed_;
+
+        mutable std::mutex mutex_;
+        // Both protected by mutex_.
+        std::optional<async::Promise<FDv2SourceResult>> pending_promise_;
+        std::deque<FDv2SourceResult> result_queue_;
+    };
+
+    /**
+     * Turns the FDv1 source's Init and Upsert calls into FDv2 changesets
+     * queued on State. Thread-safe (delegates to State).
+     */
+    class ConvertingSink final : public IDataSourceUpdateSink {
+       public:
+        explicit ConvertingSink(std::weak_ptr<State> state);
+
+        void Init(
+            Context const& context,
+            std::unordered_map<std::string, ItemDescriptor> data) override;
+        void Upsert(Context const& context,
+                    std::string key,
+                    ItemDescriptor item) override;
+        void Apply(Context const& context,
+                   FlagChangeSet change_set,
+                   bool from_cache) override;
+
+       private:
+        std::weak_ptr<State> state_;
+    };
+
+    // Thread-safe primitive. Declared before state_ so state_'s constructor
+    // can take a future from it.
+    async::Promise<std::monostate> close_promise_;
+
+    // shared_ptr so async callbacks that fire after this is destroyed can
+    // hold their own reference.
+    std::shared_ptr<State> const state_;
+    std::shared_ptr<ConvertingSink> const sink_;
+    std::shared_ptr<DataSourceStatusManager> const status_manager_;
+    std::unique_ptr<IConnection> const status_subscription_;
+
+    std::shared_ptr<IDataSource> const fdv1_source_;
+
+    // Serializes Start and ShutdownAsync on fdv1_source_ across concurrent
+    // Next() and Close() calls.
+    std::mutex lifecycle_mutex_;
+    // Protected by lifecycle_mutex_. Set when Next() starts the source, or
+    // when Close() runs first, so that a later Next() cannot start it.
+    bool started_ = false;
+};
+
+}  // namespace launchdarkly::client_side::data_sources