feat: Expose server-side flag change notifier on Client

contract-tests/data-model/include/data_model/data_model.hpp

@@ -3,7 +3,7 @@
 #include <optional>
 #include <string>
 #include <unordered_map>
 #include "nlohmann/json.hpp"
 
 namespace nlohmann {
 template <typename T>
@@ -65,7 +65,7 @@
                                                 pollIntervalMs,
                                                 filter);
 
 struct ConfigEventParams {
     std::optional<std::string> baseUri;
     std::optional<uint32_t> capacity;
     std::optional<bool> enableDiagnostics;
@@ -93,7 +93,7 @@
                                                 polling,
                                                 events);
 
 struct ConfigClientSideParams {
     nlohmann::json initialContext;
     std::optional<bool> evaluationReasons;
     std::optional<bool> useReport;
@@ -136,7 +136,7 @@
                                                 data,
                                                 errors);
 
 struct ConfigHooksParams {
     std::vector<ConfigHookInstance> hooks;
 };
 
@@ -268,7 +268,7 @@
                                                 persistentDataStore,
                                                 bigSegments);
 
 struct ContextSingleParams {
     std::optional<std::string> kind;
     std::string key;
     std::optional<std::string> name;
@@ -363,7 +363,7 @@
                                                 defaultValue,
                                                 detail);
 
 struct EvaluateFlagResponse {
     nlohmann::json value;
     std::optional<uint32_t> variationIndex;
     std::optional<nlohmann::json> reason;
@@ -374,7 +374,7 @@
                                                 variationIndex,
                                                 reason);
 
 struct EvaluateAllFlagParams {
     std::optional<nlohmann::json> context;
     std::optional<bool> withReasons;
     std::optional<bool> clientSideOnly;
@@ -387,7 +387,7 @@
                                                 clientSideOnly,
                                                 detailsOnlyForTrackedFlags);
 
 struct EvaluateAllFlagsResponse {
     nlohmann::json state;
 };
 
@@ -409,7 +409,7 @@
                                                 omitNullData,
                                                 metricValue);
 
 struct IdentifyEventParams {
     nlohmann::json context;
 };
 
@@ -424,6 +424,23 @@
                                                 available,
                                                 stale);
 
+struct RegisterFlagChangeListenerParams {
+    std::string listenerId;
+    std::string callbackUri;
+};
+
+NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE_WITH_DEFAULT(
+    RegisterFlagChangeListenerParams,
+    listenerId,
+    callbackUri);
+
+struct UnregisterListenerParams {
+    std::string listenerId;
+};
+
+NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE_WITH_DEFAULT(UnregisterListenerParams,
+                                                listenerId);
+
 enum class Command {
     Unknown = -1,
     EvaluateFlag,
@@ -433,20 +450,24 @@
     FlushEvents,
     ContextBuild,
     ContextConvert,
-    GetBigSegmentStoreStatus
-};
-
-NLOHMANN_JSON_SERIALIZE_ENUM(Command,
-                             {{Command::Unknown, nullptr},
-                              {Command::EvaluateFlag, "evaluate"},
-                              {Command::EvaluateAllFlags, "evaluateAll"},
-                              {Command::IdentifyEvent, "identifyEvent"},
-                              {Command::CustomEvent, "customEvent"},
-                              {Command::FlushEvents, "flushEvents"},
-                              {Command::ContextBuild, "contextBuild"},
-                              {Command::ContextConvert, "contextConvert"},
-                              {Command::GetBigSegmentStoreStatus,
-                               "getBigSegmentStoreStatus"}});
+    GetBigSegmentStoreStatus,
+    RegisterFlagChangeListener,
+    UnregisterListener
+};
+
+NLOHMANN_JSON_SERIALIZE_ENUM(
+    Command,
+    {{Command::Unknown, nullptr},
+     {Command::EvaluateFlag, "evaluate"},
+     {Command::EvaluateAllFlags, "evaluateAll"},
+     {Command::IdentifyEvent, "identifyEvent"},
+     {Command::CustomEvent, "customEvent"},
+     {Command::FlushEvents, "flushEvents"},
+     {Command::ContextBuild, "contextBuild"},
+     {Command::ContextConvert, "contextConvert"},
+     {Command::GetBigSegmentStoreStatus, "getBigSegmentStoreStatus"},
+     {Command::RegisterFlagChangeListener, "registerFlagChangeListener"},
+     {Command::UnregisterListener, "unregisterListener"}});
 
 struct CommandParams {
     Command command;
@@ -456,6 +477,8 @@
     std::optional<IdentifyEventParams> identifyEvent;
     std::optional<ContextBuildParams> contextBuild;
     std::optional<ContextConvertParams> contextConvert;
+    std::optional<RegisterFlagChangeListenerParams> registerFlagChangeListener;
+    std::optional<UnregisterListenerParams> unregisterListener;
     CommandParams();
 };
 
@@ -466,4 +489,6 @@
                                                 customEvent,
                                                 identifyEvent,
                                                 contextBuild,
-                                                contextConvert);
+                                                contextConvert,
+                                                registerFlagChangeListener,
+                                                unregisterListener);

contract-tests/server-contract-tests/CMakeLists.txt

@@ -18,6 +18,7 @@ add_executable(server-tests
         src/client_entity.cpp
         src/contract_test_hook.cpp
         src/contract_test_big_segment_store.cpp
+        src/http_callback.cpp
         )
 
 target_link_libraries(server-tests PRIVATE

contract-tests/server-contract-tests/include/client_entity.hpp

@@ -1,13 +1,19 @@
 #pragma once
 
 #include <data_model/data_model.hpp>
+#include <launchdarkly/connection.hpp>
 #include <launchdarkly/server_side/client.hpp>
+
+#include <boost/asio/any_io_executor.hpp>
+
 #include <memory>
+#include <string>
+#include <unordered_map>
 
 class ClientEntity {
    public:
-    explicit ClientEntity(
-        std::unique_ptr<launchdarkly::server_side::Client> client);
+    ClientEntity(std::unique_ptr<launchdarkly::server_side::Client> client,
+                 boost::asio::any_io_executor executor);
 
     tl::expected<nlohmann::json, std::string> Command(CommandParams params);
 
@@ -29,7 +35,18 @@
 
     tl::expected<nlohmann::json, std::string> GetBigSegmentStoreStatus();
 
+    tl::expected<nlohmann::json, std::string> RegisterFlagChangeListener(
+        RegisterFlagChangeListenerParams const&);
+
+    tl::expected<nlohmann::json, std::string> UnregisterListener(
+        UnregisterListenerParams const&);
+
+    boost::asio::any_io_executor executor_;
     std::unique_ptr<launchdarkly::server_side::Client> client_;
+    // Declared after client_ so listeners disconnect before the client is
+    // destroyed.
+    std::unordered_map<std::string, std::unique_ptr<launchdarkly::IConnection>>
+        listeners_;
 };
 
 static tl::expected<nlohmann::json, std::string> ContextConvert(

contract-tests/server-contract-tests/include/http_callback.hpp

@@ -0,0 +1,18 @@
+#pragma once
+
+#include <boost/asio/any_io_executor.hpp>
+#include <nlohmann/json.hpp>
+
+#include <string>
+
+/**
+ * Synchronously POSTs a JSON payload to a test harness callback URI. Errors
+ * are ignored. Blocks the calling thread until the response is read, so
+ * sequential calls arrive at the harness in order.
+ * @param executor Executor used to construct the socket.
+ * @param callback_uri The URI to post to.
+ * @param payload The JSON body.
+ */
+void PostJsonCallback(boost::asio::any_io_executor executor,
+                      std::string const& callback_uri,
+                      nlohmann::json const& payload);

contract-tests/server-contract-tests/src/client_entity.cpp

@@ -1,4 +1,5 @@
 #include "client_entity.hpp"
+#include "http_callback.hpp"
 
 #include <boost/core/ignore_unused.hpp>
 #include <boost/json.hpp>
@@ -16,7 +17,7 @@
 
 using namespace launchdarkly::server_side;
 
 tl::expected<launchdarkly::Context, std::string> ParseContext(
     nlohmann::json value) {
     boost::system::error_code ec;
     auto boost_json_val = boost::json::parse(value.dump(), ec);
@@ -40,8 +41,9 @@
 }
 
 ClientEntity::ClientEntity(
-    std::unique_ptr<launchdarkly::server_side::Client> client)
-    : client_(std::move(client)) {}
+    std::unique_ptr<launchdarkly::server_side::Client> client,
+    boost::asio::any_io_executor executor)
+    : executor_(std::move(executor)), client_(std::move(client)) {}
 
 tl::expected<nlohmann::json, std::string> ClientEntity::Identify(
     IdentifyEventParams const& params) {
@@ -359,6 +361,37 @@
 
     return result;
 }
+tl::expected<nlohmann::json, std::string>
+ClientEntity::RegisterFlagChangeListener(
+    RegisterFlagChangeListenerParams const& params) {
+    auto connection = client_->FlagNotifier().OnFlagChange(
+        [executor = executor_, listener_id = params.listenerId,
+         callback_uri = params.callbackUri](auto changes) {
+            for (auto const& flag_key : *changes) {
+                PostJsonCallback(
+                    executor, callback_uri,
+                    {{"listenerId", listener_id}, {"flagKey", flag_key}});
+            }
+        });
+    if (auto existing = listeners_.find(params.listenerId);
+        existing != listeners_.end()) {
+        existing->second->Disconnect();
+    }
+    listeners_[params.listenerId] = std::move(connection);
+    return nlohmann::json{};
+}
+
+tl::expected<nlohmann::json, std::string> ClientEntity::UnregisterListener(
+    UnregisterListenerParams const& params) {
+    auto it = listeners_.find(params.listenerId);
+    if (it == listeners_.end()) {
+        return tl::make_unexpected("unknown listener: " + params.listenerId);
+    }
+    it->second->Disconnect();
+    listeners_.erase(it);
+    return nlohmann::json{};
+}
+
 tl::expected<nlohmann::json, std::string> ClientEntity::Command(
     CommandParams params) {
     switch (params.command) {
@@ -399,6 +432,19 @@
             return ContextConvert(*params.contextConvert);
         case Command::GetBigSegmentStoreStatus:
             return GetBigSegmentStoreStatus();
+        case Command::RegisterFlagChangeListener:
+            if (!params.registerFlagChangeListener) {
+                return tl::make_unexpected(
+                    "registerFlagChangeListener params must be set");
+            }
+            return RegisterFlagChangeListener(
+                *params.registerFlagChangeListener);
+        case Command::UnregisterListener:
+            if (!params.unregisterListener) {
+                return tl::make_unexpected(
+                    "unregisterListener params must be set");
+            }
+            return UnregisterListener(*params.unregisterListener);
     }
     return tl::make_unexpected("unrecognized command");
 }

contract-tests/server-contract-tests/src/contract_test_hook.cpp

@@ -1,22 +1,11 @@
 #include "contract_test_hook.hpp"
+#include "http_callback.hpp"
 
 #include <launchdarkly/detail/serialization/json_primitives.hpp>
 #include <launchdarkly/detail/serialization/json_value.hpp>
 #include <launchdarkly/serialization/json_context.hpp>
 #include <launchdarkly/serialization/json_evaluation_reason.hpp>
 
-#include <boost/asio/connect.hpp>
-#include <boost/asio/ip/tcp.hpp>
-#include <boost/asio/post.hpp>
-#include <boost/beast/core.hpp>
-#include <boost/beast/http.hpp>
-#include <boost/url.hpp>
-
-namespace beast = boost::beast;
-namespace http = beast::http;
-namespace net = boost::asio;
-using tcp = net::ip::tcp;
-
 using namespace launchdarkly::server_side::hooks;
 
 ContractTestHook::ContractTestHook(boost::asio::any_io_executor executor,
@@ -55,52 +44,7 @@
 
 void ContractTestHook::PostCallback(std::string const& stage,
                                     nlohmann::json const& payload) {
-    auto uri_result = boost::urls::parse_uri(config_.callbackUri);
-    if (!uri_result) {
-        return;
-    }
-
-    auto uri = *uri_result;
-    std::string host(uri.host());
-    std::string port = uri.has_port() ? std::string(uri.port()) : "80";
-    std::string target(uri.path());
-    if (target.empty()) {
-        target = "/";
-    }
-
-    // Send synchronously so callbacks from sequential hook invocations arrive
-    // at the harness in the order the SDK invoked them.
-    beast::error_code ec;
-    tcp::resolver resolver(executor_);
-    auto results = resolver.resolve(host, port, ec);
-    if (ec) {
-        return;
-    }
-
-    beast::tcp_stream stream(executor_);
-    stream.connect(results, ec);
-    if (ec) {
-        return;
-    }
-
-    http::request<http::string_body> req{http::verb::post, target, 11};
-    req.set(http::field::host, host);
-    req.set(http::field::user_agent, "cpp-server-sdk-contract-tests");
-    req.set(http::field::content_type, "application/json");
-    req.body() = payload.dump();
-    req.prepare_payload();
-
-    http::write(stream, req, ec);
-    if (ec) {
-        return;
-    }
-
-    beast::flat_buffer buffer;
-    http::response<http::string_body> res;
-    http::read(stream, buffer, res, ec);
-
-    beast::error_code ignore_ec;
-    stream.socket().shutdown(tcp::socket::shutdown_both, ignore_ec);
+    PostJsonCallback(executor_, config_.callbackUri, payload);
 }
 
 EvaluationSeriesData ContractTestHook::BeforeEvaluation(

contract-tests/server-contract-tests/src/entity_manager.cpp

@@ -1,4 +1,4 @@
 #include "entity_manager.hpp"
 #include "contract_test_big_segment_store.hpp"
 #include "contract_test_hook.hpp"
 
@@ -91,7 +91,7 @@
 }
 #endif
 
 config::builders::DataSystemBuilder::FDv2 BuildFDv2(
     ConfigDataSystemParams const& cfg,
     config::builders::EndpointsBuilder* endpoints) {
     auto fdv2 = config::builders::DataSystemBuilder::FDv2::Custom();
@@ -424,7 +424,7 @@
     auto init = client->StartAsync();
     init.wait_for(waitForClient);
 
-    entities_.try_emplace(id, std::move(client));
+    entities_.try_emplace(id, std::move(client), executor_);
 
     return id;
 }

contract-tests/server-contract-tests/src/http_callback.cpp

@@ -0,0 +1,61 @@
+#include "http_callback.hpp"
+
+#include <boost/asio/connect.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/beast/http.hpp>
+#include <boost/url.hpp>
+
+namespace beast = boost::beast;
+namespace http = beast::http;
+namespace net = boost::asio;
+using tcp = net::ip::tcp;
+
+void PostJsonCallback(boost::asio::any_io_executor executor,
+                      std::string const& callback_uri,
+                      nlohmann::json const& payload) {
+    auto uri_result = boost::urls::parse_uri(callback_uri);
+    if (!uri_result) {
+        return;
+    }
+
+    auto uri = *uri_result;
+    std::string host(uri.host());
+    std::string port = uri.has_port() ? std::string(uri.port()) : "80";
+    std::string target(uri.path());
+    if (target.empty()) {
+        target = "/";
+    }
+
+    beast::error_code ec;
+    tcp::resolver resolver(executor);
+    auto results = resolver.resolve(host, port, ec);
+    if (ec) {
+        return;
+    }
+
+    beast::tcp_stream stream(executor);
+    stream.connect(results, ec);
+    if (ec) {
+        return;
+    }
+
+    http::request<http::string_body> req{http::verb::post, target, 11};
+    req.set(http::field::host, host);
+    req.set(http::field::user_agent, "cpp-server-sdk-contract-tests");
+    req.set(http::field::content_type, "application/json");
+    req.body() = payload.dump();
+    req.prepare_payload();
+
+    http::write(stream, req, ec);
+    if (ec) {
+        return;
+    }
+
+    beast::flat_buffer buffer;
+    http::response<http::string_body> res;
+    http::read(stream, buffer, res, ec);
+
+    beast::error_code ignore_ec;
+    stream.socket().shutdown(tcp::socket::shutdown_both, ignore_ec);
+}

contract-tests/server-contract-tests/src/main.cpp

@@ -1,4 +1,4 @@
 #include "server.hpp"
 
 #include <launchdarkly/logging/console_backend.hpp>
 
@@ -54,6 +54,7 @@
         srv.add_capability("instance-id");
         srv.add_capability("fdv1-fallback");
         srv.add_capability("big-segments");
+        srv.add_capability("flag-change-listeners");
 #ifdef LD_REDIS_SUPPORT_ENABLED
         srv.add_capability("persistent-data-store-redis");
 #endif