feat: Add MessageBus to core, add MakeMessageHandler to Feature

src/datadog/impl/core/context.cpp

@@ -12,6 +12,7 @@
 #include "datadog/uuid.hpp"
 
 #include "datadog/impl/assert.hpp"
+#include "datadog/impl/core/message_bus.hpp"
 #include "datadog/impl/core/version.hpp"
 
 namespace datadog::impl {
@@ -187,11 +188,24 @@ CoreContext CoreContextProvider::Get() const {
 }
 
 void CoreContextProvider::Update(const std::function<void(CoreContext&)>& callback) {
-  // Acquire an exclusive write lock, then allow the callback to mutate the context
-  std::unique_lock lock(_mutex);
-  callback(_context);
+  // Acquire an exclusive write lock, mutate the context, and capture a snapshot. The
+  // lock is released before Send() to minimize contention on the message bus.
+  CoreContext snapshot = [&] {
+    std::unique_lock lock(_mutex);
+    callback(_context);
+    return _context;
+  }();
+
+  // CoreContext has been mutated (and we have a valid snapshot): notify all registered
+  // message-handlers of the change, so that Features can perform work (on the messaging
+  // thread) in response to the updated context
+  if (_message_bus) {
+    _message_bus->Send(ContextChangedMessage{std::move(snapshot)});
+  }
 }
 
+void CoreContextProvider::SetMessageBus(MessageBus* bus) { _message_bus = bus; }
+
 const HttpContext& CoreContextProvider::GetHttpContext() const {
   std::shared_lock lock(_mutex);
   return *_context.http;

src/datadog/impl/core/context.hpp

@@ -24,6 +24,8 @@ struct DeviceInfo;
 
 namespace datadog::impl {
 
+class MessageBus;
+
 /**
  * SDK configuration details that influence how HTTP requests are built. Immutable for
  * the lifetime of the Core.
@@ -159,6 +161,7 @@ class CoreContextProvider {
  private:
   CoreContext _context;
   mutable std::shared_mutex _mutex;
+  MessageBus* _message_bus{nullptr};
 
  public:
   explicit CoreContextProvider(const CoreContext& context);
@@ -171,13 +174,25 @@ class CoreContextProvider {
   /**
    * Mutates the current CoreContext value, by invoking the provided callback after
    * obtaining exclusive write access.
+   *
+   * After mutation, dispatches a `ContextChangedMessage` on the `MessageBus`, carrying
+   * an immutable snapshot of the updated context to any Features that have registered
+   * an interest in receiving notifications of context changes.
    */
   void Update(const std::function<void(CoreContext&)>& callback);
 
   /**
    * Synchronously returns an immutable reference to the HttpContext value.
    */
   const HttpContext& GetHttpContext() const;
+
+  /**
+   * Registers the `MessageBus` that `Update()` will dispatch `ContextChangedMessage`
+   * values to. Pass `nullptr` to detach the bus (e.g. during shutdown). Called by
+   * `Core::Start()` before the context thread launches; no synchronization is needed
+   * because the context thread is not yet running at that point.
+   */
+  void SetMessageBus(MessageBus* bus);
 };
 
 }  // namespace datadog::impl

src/datadog/impl/core/core.cpp

@@ -13,6 +13,7 @@
 
 #include "datadog/impl/assert.hpp"
 #include "datadog/impl/core/context_thread.hpp"
+#include "datadog/impl/core/messaging_thread.hpp"
 #include "datadog/impl/core/storage_thread.hpp"
 #include "datadog/impl/core/types.hpp"
 #include "datadog/impl/core/version.hpp"
@@ -266,6 +267,7 @@ bool Core::RegisterFeature(const std::shared_ptr<Feature>& impl) {
       std::move(*event_read_directory),
       std::move(upload_state)
   );
+
   _diagnostic_logger.Debug(
       "Feature registered",
       {{"feature", name}, {"feature_id", static_cast<int64_t>(id)}}
@@ -320,8 +322,29 @@ bool Core::Start() {
       std::ref(_features)
   );
 
-  // Initialize a thread-safe queue for functions that will execute on the context
-  // thread
+  // Iterate through all registered features and call MakeMessageHandler(), allowing
+  // features to register their intent to be notified when messages are sent on the
+  // message bus
+  std::vector<std::function<void(const FeatureMessage&)>> handlers;
+  handlers.reserve(_features.size());
+  for (const auto& f : _features) {
+    if (auto h = f.impl->MakeMessageHandler()) {
+      handlers.push_back(std::move(*h));
+    }
+  }
+
+  // Create the message bus, which will queue all core-to-feature and feature-to-feature
+  // messages that need to be handled by interested features
+  DATADOG_ASSERT(!_message_bus, "_message_bus already exists on Start()");
+  _message_bus = std::make_unique<MessageBus>(std::move(handlers));
+
+  // Install the MessageBus into the CoreContextProvider so that it can send
+  // ContextChangedMessage in response to updates: context thread doesn't exist yet, so
+  // no synchronization is required here
+  _context_provider->SetMessageBus(_message_bus.get());
+
+  // Initialize a thread-safe queue for feature-submitted functions that will execute on
+  // the context thread
   DATADOG_ASSERT(!_context_queue, "_context_queue already exists on Start()");
   _context_queue = std::make_unique<Queue<std::function<void()>>>();
 
@@ -383,6 +406,18 @@ bool Core::Start() {
         )
     );
   }
+
+  // Start the messaging thread only after all features have completed OnCoreStarted().
+  // This ensures that any ContextChangedMessages dispatched during startup (e.g. from
+  // a feature calling UpdateContext in its Start()) are not delivered to a feature's
+  // handler before that feature's own OnCoreStarted() has run. Messages enqueued
+  // during the loop above are buffered in the queue and will be drained once this
+  // thread starts.
+  DATADOG_ASSERT(!_message_bus_thread, "_message_bus_thread already exists on Start()");
+  _message_bus_thread = std::thread(
+      MessagingThreadMain, std::ref(_diagnostic_logger), std::ref(*_message_bus)
+  );
+
   return true;
 }
 
@@ -422,14 +457,31 @@ void Core::Stop() {
   }
   _context_thread.reset();
 
+  // Stop the messaging thread after the context thread exits: the context thread is
+  // the primary producer of messages (via Update()), so draining it first prevents new
+  // messages from arriving after we signal the messaging thread to stop. After joining,
+  // detach the bus from the context provider so any stray Update() calls (which should
+  // not happen at this point) do not reference the destroyed bus.
+  DATADOG_ASSERT(_message_bus, "_message_bus is invalid on Stop");
+  DATADOG_ASSERT(
+      _message_bus_thread && _message_bus_thread->joinable(),
+      "_message_bus_thread is non-joinable on Stop"
+  );
+  _message_bus->Stop();
+  _diagnostic_logger.Debug("Joining on messaging thread");
+  _message_bus_thread->join();
+  _message_bus_thread.reset();
+  _context_provider->SetMessageBus(nullptr);
+
   // Notify each registered feature that the core has stopped, now that no more
   // context-thread functions enqueued by those features may be running
   for (const auto& feature : _features) {
     feature.impl->OnCoreStopping();
   }
 
-  // Destroy the context queue, now that no more features exist
+  // Destroy the context queue and message bus, now that no more features exist
   _context_queue.reset();
+  _message_bus.reset();
 
   // Stop all queue processing, then block until the consumer thread drains the queue
   // and exits, at which point it's safe to release the queue

src/datadog/impl/core/core.hpp

@@ -20,6 +20,7 @@
 #include "datadog/impl/core/block.hpp"
 #include "datadog/impl/core/context.hpp"
 #include "datadog/impl/core/feature.hpp"
+#include "datadog/impl/core/message_bus.hpp"
 #include "datadog/impl/core/queue.hpp"
 #include "datadog/impl/core/storage_queue.hpp"
 #include "datadog/impl/core/storage_write.hpp"
@@ -389,6 +390,9 @@ class Core {
   std::unique_ptr<Queue<std::function<void()>>> _context_queue;
   std::optional<std::thread> _context_thread;
 
+  std::unique_ptr<MessageBus> _message_bus;
+  std::optional<std::thread> _message_bus_thread;
+
   std::unique_ptr<UploadScheduler> _upload_scheduler;
   std::optional<std::thread> _upload_thread;
 

src/datadog/impl/core/feature.hpp

@@ -18,6 +18,7 @@
 #include "datadog/impl/core/block.hpp"
 #include "datadog/impl/core/context.hpp"
 #include "datadog/impl/core/feature_id.hpp"
+#include "datadog/impl/core/feature_message.hpp"
 #include "datadog/impl/core/feature_scope.hpp"
 #include "datadog/impl/core/tlv.hpp"
 #include "datadog/impl/platform/filesystem.hpp"
@@ -136,6 +137,23 @@ class Feature : public std::enable_shared_from_this<Feature> {
       const HttpContext& context, class BatchReader& reader
   ) = 0;
 
+  /**
+   * Called by `Core::Start()` each time the SDK starts up. If the feature wants to
+   * react to messages dispatched on the `MessageBus`, it should override this method
+   * and return a handler function: typically a lambda that captures `weak_from_this()`
+   * so that message delivery is silently skipped if the feature has already been
+   * destroyed.
+   *
+   * Implementations must be idempotent, as this function called on every `Start()`.
+   *
+   * The default implementation returns `std::nullopt`, meaning the feature opts out of
+   * message handling entirely.
+   */
+  virtual std::optional<std::function<void(const FeatureMessage&)>>
+  MakeMessageHandler() {
+    return std::nullopt;
+  }
+
  protected:
   std::optional<FeatureScope> _scope;
 };

tests/impl/core/core_test.cpp

@@ -6,8 +6,11 @@
 
 #include "datadog/impl/core/core.hpp"
 
+#include <atomic>
 #include <catch2/catch_test_macros.hpp>
 
+#include "datadog/impl/core/feature_message.hpp"
+
 #include "mock/clock.hpp"
 #include "mock/feature.hpp"
 #include "mock/filesystem.hpp"
@@ -288,3 +291,58 @@ TEST_CASE("Core Lifecycle", "[unit]") {
     REQUIRE(feature->num_stop_calls == 1);
   }
 }
+
+/**
+ * Feature that counts received ContextChangedMessages via MakeMessageHandler(), and
+ * triggers a context update (causing one such message to be dispatched) each time the
+ * core starts.
+ */
+class MessageHandlerFeature : public MockFeature {
+ public:
+  std::atomic<int> messages_received{0};
+
+  MessageHandlerFeature()
+      : MockFeature(CreateFeatureId("MHFT"), "message_handler_test") {}
+
+  std::optional<std::function<void(const FeatureMessage&)>>
+  MakeMessageHandler() override {
+    auto weak = weak_from_this();
+    return [weak](const FeatureMessage&) {
+      if (auto self = std::static_pointer_cast<MessageHandlerFeature>(weak.lock())) {
+        self->messages_received.fetch_add(1);
+      }
+    };
+  }
+
+  void Start() override {
+    MockFeature::Start();
+    // Trigger a context update so that a ContextChangedMessage is dispatched to
+    // all registered handlers on each start
+    _scope->UpdateContext([](CoreContext& ctx) { ctx.rum.emplace(); });
+  }
+};
+
+TEST_CASE("Core Messaging", "[unit]") {
+  SECTION(
+      "M deliver messages to handler after restart W core is stopped and "
+      "started again"
+  ) {
+    impl::Core core = _make_core();
+    REQUIRE(core.Init());
+
+    auto feature = std::make_shared<MessageHandlerFeature>();
+    REQUIRE(core.RegisterFeature(feature));
+
+    // First run: start triggers one context update, stop drains both the context
+    // thread and the messaging thread before returning
+    REQUIRE(core.Start());
+    core.Stop();
+    REQUIRE(feature->messages_received.load() == 1);
+
+    // Second run: the handler must still be registered after the restart; the
+    // same context update should deliver a second message
+    REQUIRE(core.Start());
+    core.Stop();
+    REQUIRE(feature->messages_received.load() == 2);
+  }
+}