feat: implement event delivery (#29)

* feat:  implement asynchronous event delivery

Author: cwaldren-ld

libs/client-sdk/include/launchdarkly/client_side/api.hpp

@@ -12,9 +12,9 @@
 #include "config/client.hpp"
 #include "context.hpp"
 #include "error.hpp"
-#include "events/event_processor.hpp"
 #include "launchdarkly/client_side/data_source.hpp"
 #include "launchdarkly/client_side/data_sources/detail/data_source_status_manager.hpp"
+#include "launchdarkly/client_side/event_processor.hpp"
 #include "launchdarkly/client_side/flag_manager/detail/flag_manager.hpp"
 #include "launchdarkly/client_side/flag_manager/detail/flag_updater.hpp"
 #include "logger.hpp"
@@ -25,6 +25,11 @@ class Client {
    public:
     Client(Config config, Context context);
 
+    Client(Client&&) = delete;
+    Client(Client const&) = delete;
+    Client& operator=(Client) = delete;
+    Client& operator=(Client&& other) = delete;
+
     using FlagKey = std::string;
     [[nodiscard]] std::unordered_map<FlagKey, Value> AllFlags() const;
 
@@ -72,7 +77,7 @@ class Client {
     std::thread thread_;
     boost::asio::io_context ioc_;
     Context context_;
-    std::unique_ptr<events::IEventProcessor> event_processor_;
+    std::unique_ptr<IEventProcessor> event_processor_;
     std::unique_ptr<IDataSource> data_source_;
     std::thread run_thread_;
 };

libs/common/include/events/event_processor.hpp renamed to libs/client-sdk/include/launchdarkly/client_side/event_processor.hpp

@@ -2,7 +2,7 @@
 
 #include "events/events.hpp"
 
-namespace launchdarkly::events {
+namespace launchdarkly::client_side {
 
 class IEventProcessor {
    public:
@@ -12,7 +12,7 @@ class IEventProcessor {
      * capacity.
      * @param event InputEvent to deliver.
      */
-    virtual void AsyncSend(InputEvent event) = 0;
+    virtual void AsyncSend(events::InputEvent event) = 0;
     /**
      * Asynchronously flush's the processor's events, returning as soon as
      * possible. Flushing may be a no-op if a flush is ongoing.
@@ -34,4 +34,4 @@ class IEventProcessor {
     IEventProcessor() = default;
 };
 
-}  // namespace launchdarkly::events
+}  // namespace launchdarkly::client_side

libs/client-sdk/include/launchdarkly/client_side/event_processor/detail/event_processor.hpp

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <boost/asio/any_io_executor.hpp>
+#include "config/client.hpp"
+#include "config/detail/sdks.hpp"
+#include "events/detail/asio_event_processor.hpp"
+#include "launchdarkly/client_side/event_processor.hpp"
+#include "logger.hpp"
+
+namespace launchdarkly::client_side::detail {
+
+class EventProcessor : public IEventProcessor {
+   public:
+    EventProcessor(boost::asio::any_io_executor const& io,
+                   Config const& config,
+                   Logger& logger);
+    void AsyncSend(events::InputEvent event) override;
+    void AsyncFlush() override;
+    void AsyncClose() override;
+
+   private:
+    events::detail::AsioEventProcessor<SDK> impl_;
+};
+
+}  // namespace launchdarkly::client_side::detail

libs/client-sdk/include/launchdarkly/client_side/event_processor/detail/null_event_processor.hpp

@@ -0,0 +1,14 @@
+#pragma once
+
+#include "launchdarkly/client_side/event_processor.hpp"
+
+namespace launchdarkly::client_side::detail {
+
+class NullEventProcessor : public IEventProcessor {
+   public:
+    NullEventProcessor() = default;
+    void AsyncSend(events::InputEvent event) override;
+    void AsyncFlush() override;
+    void AsyncClose() override;
+};
+}  // namespace launchdarkly::client_side::detail

libs/client-sdk/src/CMakeLists.txt

@@ -6,6 +6,8 @@ file(GLOB HEADER_LIST CONFIGURE_DEPENDS
         "${LaunchDarklyCPPClient_SOURCE_DIR}/include/launchdarkly/client_side/data_sources/detail/*.hpp"
         "${LaunchDarklyCPPClient_SOURCE_DIR}/include/launchdarkly/client_side/flag_manager/*.hpp"
         "${LaunchDarklyCPPClient_SOURCE_DIR}/include/launchdarkly/client_side/flag_manager/detail/*.hpp"
+        "${LaunchDarklyCPPClient_SOURCE_DIR}/include/launchdarkly/client_side/event_processor/*.hpp"
+        "${LaunchDarklyCPPClient_SOURCE_DIR}/include/launchdarkly/client_side/event_processor/detail/*.hpp"
         )
 
 # Automatic library: static or dynamic based on user config.
@@ -23,6 +25,8 @@ add_library(${LIBNAME}
         flag_manager/flag_change_event.cpp
         data_sources/data_source_status.cpp
         data_sources/data_source_status_manager.cpp
+        event_processor/event_processor.cpp
+        event_processor/null_event_processor.cpp
         boost_signal_connection.cpp
         )
 

libs/client-sdk/src/api.cpp

@@ -3,9 +3,10 @@
 #include <optional>
 #include <utility>
 
-#include "events/detail/asio_event_processor.hpp"
 #include "launchdarkly/client_side/data_sources/detail/polling_data_source.hpp"
 #include "launchdarkly/client_side/data_sources/detail/streaming_data_source.hpp"
+#include "launchdarkly/client_side/event_processor/detail/event_processor.hpp"
+#include "launchdarkly/client_side/event_processor/detail/null_event_processor.hpp"
 
 namespace launchdarkly::client_side {
 
@@ -14,7 +15,7 @@ using launchdarkly::client_side::data_sources::DataSourceStatus;
 static std::unique_ptr<IDataSource> MakeDataSource(
     Config const& config,
     Context const& context,
-    boost::asio::any_io_executor executor,
+    boost::asio::any_io_executor const& executor,
     flag_manager::detail::FlagUpdater& flag_updater,
     data_sources::detail::DataSourceStatusManager& status_manager,
     Logger& logger) {
@@ -32,13 +33,7 @@ static std::unique_ptr<IDataSource> MakeDataSource(
 Client::Client(Config config, Context context)
     : logger_(config.Logger()),
       context_(std::move(context)),
-      event_processor_(
-          std::make_unique<launchdarkly::events::detail::AsioEventProcessor>(
-              ioc_.get_executor(),
-              config.Events(),
-              config.ServiceEndpoints(),
-              config.SdkKey(),
-              logger_)),
+      event_processor_(nullptr),
       flag_updater_(flag_manager_),
       data_source_(MakeDataSource(config,
                                   context_,
@@ -47,6 +42,15 @@ Client::Client(Config config, Context context)
                                   status_manager_,
                                   logger_)),
       initialized_(false) {
+    if (config.Events().Enabled()) {
+        event_processor_ = std::make_unique<detail::EventProcessor>(
+            ioc_.get_executor(), config, logger_);
+    } else {
+        event_processor_ = std::make_unique<detail::NullEventProcessor>();
+    }
+
+    data_source_->Start();
+
     status_manager_.OnDataSourceStatusChange([this](auto status) {
         if (status.State() == DataSourceStatus::DataSourceState::kValid ||
             status.State() == DataSourceStatus::DataSourceState::kShutdown ||

libs/client-sdk/src/event_processor/event_processor.cpp

@@ -0,0 +1,22 @@
+#include "launchdarkly/client_side/event_processor/detail/event_processor.hpp"
+
+namespace launchdarkly::client_side::detail {
+
+EventProcessor::EventProcessor(boost::asio::any_io_executor const& io,
+                               Config const& config,
+                               Logger& logger)
+    : impl_(io, config, logger) {}
+
+void EventProcessor::AsyncSend(launchdarkly::events::InputEvent event) {
+    impl_.AsyncSend(std::move(event));
+}
+
+void EventProcessor::AsyncFlush() {
+    impl_.AsyncFlush();
+}
+
+void EventProcessor::AsyncClose() {
+    impl_.AsyncClose();
+}
+
+}  // namespace launchdarkly::client_side::detail

libs/client-sdk/src/event_processor/null_event_processor.cpp

@@ -0,0 +1,10 @@
+#include "launchdarkly/client_side/event_processor/detail/null_event_processor.hpp"
+
+namespace launchdarkly::client_side::detail {
+
+void NullEventProcessor::AsyncSend(events::InputEvent event) {}
+
+void NullEventProcessor::AsyncFlush() {}
+
+void NullEventProcessor::AsyncClose() {}
+}  // namespace launchdarkly::client_side::detail

libs/common/include/config/client.hpp

@@ -16,7 +16,10 @@ using AppInfoBuilder = config::detail::builders::AppInfoBuilder;
 using EndpointsBuilder = config::detail::builders::EndpointsBuilder<SDK>;
 using ConfigBuilder = config::detail::builders::ConfigBuilder<SDK>;
 using EventsBuilder = config::detail::builders::EventsBuilder<SDK>;
+using HttpPropertiesBuilder =
+    config::detail::builders::HttpPropertiesBuilder<SDK>;
 using DataSourceBuilder = config::detail::builders::DataSourceBuilder<SDK>;
+
 using Config = config::detail::Config<SDK>;
 
 }  // namespace launchdarkly::client_side

libs/common/include/config/detail/builders/events_builder.hpp

@@ -34,6 +34,20 @@ class EventsBuilder {
      */
     EventsBuilder();
 
+    /**
+     * Specify if event-sending should be enabled or not. By default,
+     * events are enabled.
+     * @param enabled True to enable.
+     * @return Reference to this builder.
+     */
+    EventsBuilder& Enabled(bool enabled);
+
+    /**
+     * Alias for Enabled(false).
+     * @return Reference to this builder.
+     */
+    EventsBuilder& Disable();
+
     /**
      * Sets the capacity of the event processor. When more events are generated
      * within the processor's flush interval than this value, events will be

libs/common/include/config/detail/built/events.hpp

@@ -19,6 +19,8 @@ class Events final {
     friend class builders::EventsBuilder;
     /**
      * Constructs configuration for the event subsystem.
+     * @param enabled If event-sending is enabled. If false, no events will be
+     * sent to LaunchDarkly.
      * @param capacity How many events can queue in memory before new events
      * are dropped.
      * @param flush_interval How often events are automatically flushed to
@@ -31,12 +33,22 @@ class Events final {
      * AllAttributesPrivate is false.
      * @param security Whether a plaintext or encrypted client should be used
      * for event delivery.
+     * @param flush_workers How many workers to use for concurrent event
+     * delivery.
      */
-    Events(std::size_t capacity,
+    Events(bool enabled,
+           std::size_t capacity,
            std::chrono::milliseconds flush_interval,
            std::string path,
            bool all_attributes_private,
-           AttributeReference::SetType private_attrs);
+           AttributeReference::SetType private_attrs,
+           std::chrono::milliseconds delivery_retry_delay,
+           std::size_t flush_workers);
+
+    /**
+     * Returns true if event-sending is enabled.
+     */
+    [[nodiscard]] bool Enabled() const;
 
     /**
      * Capacity of the event processor.
@@ -48,6 +60,12 @@ class Events final {
      */
     [[nodiscard]] std::chrono::milliseconds FlushInterval() const;
 
+    /*
+     * If an event payload fails to be delivered and can be retried, how long
+     * to wait before retrying.
+     */
+    [[nodiscard]] std::chrono::milliseconds DeliveryRetryDelay() const;
+
     /**
      * Path component of the LaunchDarkly event delivery endpoint.
      */
@@ -63,12 +81,20 @@ class Events final {
      */
     [[nodiscard]] AttributeReference::SetType const& PrivateAttributes() const;
 
+    /**
+     * Number of flush workers used for concurrent event delivery.
+     */
+    [[nodiscard]] std::size_t FlushWorkers() const;
+
    private:
+    bool enabled_;
     std::size_t capacity_;
     std::chrono::milliseconds flush_interval_;
     std::string path_;
     bool all_attributes_private_;
     AttributeReference::SetType private_attributes_;
+    std::chrono::milliseconds delivery_retry_delay_;
+    std::size_t flush_workers_;
 };
 
 bool operator==(Events const& lhs, Events const& rhs);

libs/common/include/config/detail/defaults.hpp

@@ -35,8 +35,14 @@ struct Defaults<ClientSDK> {
     }
 
     static auto Events() -> built::Events {
-        return {100, std::chrono::seconds(30), "/mobile", false,
-                AttributeReference::SetType()};
+        return {true,
+                100,
+                std::chrono::seconds(30),
+                "/mobile",
+                false,
+                AttributeReference::SetType(),
+                std::chrono::seconds(1),
+                5};
     }
 
     static auto HttpProperties() -> built::HttpProperties {
@@ -74,8 +80,14 @@ struct Defaults<ServerSDK> {
     }
 
     static auto Events() -> built::Events {
-        return {10000, std::chrono::seconds(5), "/bulk", false,
-                AttributeReference::SetType()};
+        return {true,
+                10000,
+                std::chrono::seconds(5),
+                "/bulk",
+                false,
+                AttributeReference::SetType(),
+                std::chrono::seconds(1),
+                5};
     }
 
     static auto HttpProperties() -> built::HttpProperties {

libs/common/include/config/server.hpp

@@ -16,6 +16,8 @@ using AppInfoBuilder = config::detail::builders::AppInfoBuilder;
 using EndpointsBuilder = config::detail::builders::EndpointsBuilder<SDK>;
 using ConfigBuilder = config::detail::builders::ConfigBuilder<SDK>;
 using EventsBuilder = config::detail::builders::EventsBuilder<SDK>;
+using HttpPropertiesBuilder =
+    config::detail::builders::HttpPropertiesBuilder<SDK>;
 using DataSourceBuilder = config::detail::builders::DataSourceBuilder<SDK>;
 
 using Config = config::detail::Config<SDK>;