feat: Add FeatureMessage, MessageBus, and a messaging thread

src/CMakeLists.txt

@@ -26,6 +26,8 @@ add_library(dd_native ${DD_LIBRARY_TYPE}
     datadog/impl/attribute/merge.cpp
     datadog/impl/core/context_thread.cpp
     datadog/impl/core/context.cpp
+    datadog/impl/core/message_bus.cpp
+    datadog/impl/core/messaging_thread.cpp
     datadog/impl/core/core.cpp
     datadog/impl/core/feature_read.cpp
     datadog/impl/core/feature_scope.cpp

src/datadog/impl/core/feature_message.hpp

@@ -0,0 +1,37 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#pragma once
+
+#include <variant>
+
+#include "datadog/impl/core/context.hpp"
+
+namespace datadog::impl {
+
+/**
+ * Carries a snapshot of `CoreContext` captured immediately after a
+ * `CoreContextProvider::Update()` call completes. Subscribers can use this snapshot to
+ * react proactively to state changes in other features — the primary use case being
+ * `CrashReporting`, which needs to persist RUM session/view/action IDs every time
+ * `RumFeatureContext` changes so that those IDs are available in a crash report
+ * generated before the next normal launch.
+ *
+ * The snapshot is taken inside `Update()` while the write lock is still held, ensuring
+ * the delivered context is consistent and not partially written.
+ */
+struct ContextChangedMessage {
+  CoreContext context;
+};
+
+/**
+ * Discriminated union of all message types that can be dispatched through the
+ * `MessageBus`. Add new variants here as additional cross-feature communication needs
+ * arise.
+ */
+using FeatureMessage = std::variant<ContextChangedMessage>;
+
+}  // namespace datadog::impl

src/datadog/impl/core/message_bus.cpp

@@ -0,0 +1,18 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#include "datadog/impl/core/message_bus.hpp"
+
+namespace datadog::impl {
+
+MessageBus::MessageBus(std::vector<std::function<void(const FeatureMessage&)>> handlers)
+    : _handlers(std::move(handlers)) {}
+
+bool MessageBus::Send(FeatureMessage msg) { return _queue.Push(std::move(msg)); }
+
+void MessageBus::Stop() { _queue.Stop(); }
+
+}  // namespace datadog::impl

src/datadog/impl/core/message_bus.hpp

@@ -0,0 +1,62 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#pragma once
+
+#include <functional>
+#include <vector>
+
+#include "datadog/impl/core/feature_message.hpp"
+#include "datadog/impl/core/queue.hpp"
+#include "datadog/impl/diagnostics.hpp"
+
+namespace datadog::impl {
+
+/**
+ * Holds the shared state for the messaging subsystem: a thread-safe queue and the
+ * fixed list of subscriber callbacks. The background thread that drains the queue and
+ * invokes the handlers lives outside this class (see `MessagingThreadMain`), following
+ * the same ownership pattern used by `context_thread`, `storage_thread`, and
+ * `upload_thread`.
+ *
+ * Handlers are supplied at construction time and are never modified afterward, so
+ * `MessagingThreadMain` can iterate `_handlers` without any synchronization.
+ *
+ * `Send()` enqueues a message non-blocking from any thread and returns false if the
+ * queue has been stopped. Call `Stop()` to signal shutdown; the messaging thread will
+ * drain any remaining messages before exiting.
+ */
+class MessageBus {
+ public:
+  explicit MessageBus(std::vector<std::function<void(const FeatureMessage&)>> handlers);
+
+  ~MessageBus() = default;
+  MessageBus(const MessageBus&) = delete;
+  MessageBus& operator=(const MessageBus&) = delete;
+  MessageBus(MessageBus&&) = delete;
+  MessageBus& operator=(MessageBus&&) = delete;
+
+  /**
+   * Enqueues `msg` for delivery to all handlers. Non-blocking. Returns false if the
+   * queue has been stopped and the message was therefore dropped.
+   */
+  bool Send(FeatureMessage msg);
+
+  /**
+   * Signals the messaging thread to drain remaining messages and exit. Must be called
+   * before the bus is destroyed; the caller must join the messaging thread after this
+   * returns.
+   */
+  void Stop();
+
+ private:
+  friend void MessagingThreadMain(const DiagnosticLogger&, MessageBus&);
+
+  Queue<FeatureMessage> _queue;
+  std::vector<std::function<void(const FeatureMessage&)>> _handlers;
+};
+
+}  // namespace datadog::impl

src/datadog/impl/core/messaging_thread.cpp

@@ -0,0 +1,23 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#include "datadog/impl/core/messaging_thread.hpp"
+
+namespace datadog::impl {
+
+void MessagingThreadMain(const DiagnosticLogger& diagnostic_logger, MessageBus& bus) {
+  diagnostic_logger.Debug("Messaging thread starting");
+
+  while (auto msg = bus._queue.Pop()) {
+    for (auto& handler : bus._handlers) {
+      handler(*msg);
+    }
+  }
+
+  diagnostic_logger.Debug("Messaging thread finished");
+}
+
+}  // namespace datadog::impl

src/datadog/impl/core/messaging_thread.hpp

@@ -0,0 +1,24 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#pragma once
+
+#include "datadog/impl/core/message_bus.hpp"
+#include "datadog/impl/diagnostics.hpp"
+
+namespace datadog::impl {
+
+/**
+ * Entry point for the messaging thread.
+ *
+ * Drains `bus._queue` until it is stopped, broadcasting each `FeatureMessage` to every
+ * handler in `bus._handlers` in registration order. Because the handler list is fixed
+ * at construction time and never mutated, no additional synchronization is needed
+ * inside this function.
+ */
+void MessagingThreadMain(const DiagnosticLogger& diagnostic_logger, MessageBus& bus);
+
+}  // namespace datadog::impl

tests/CMakeLists.txt

@@ -28,6 +28,8 @@ add_executable(dd_native_tests
     impl/core/block_test.cpp
     impl/core/context_test.cpp
     impl/core/context_thread_test.cpp
+    impl/core/message_bus_test.cpp
+    impl/core/messaging_thread_test.cpp
     impl/core/core_test.cpp
     impl/core/feature_id_test.cpp
     impl/core/feature_read_test.cpp

tests/impl/core/message_bus_test.cpp

@@ -0,0 +1,161 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-Present Datadog, Inc.
+
+#include "datadog/impl/core/message_bus.hpp"
+
+#include <atomic>
+#include <functional>
+#include <thread>
+#include <vector>
+
+#include "datadog/impl/core/feature_message.hpp"
+#include "datadog/impl/core/messaging_thread.hpp"
+
+#include "support/catch.hpp"
+#include "support/context.hpp"
+#include "support/diagnostics.hpp"
+
+using namespace datadog;
+using namespace datadog::impl;
+
+/**
+ * Constructs a ContextChangedMessage carrying a CoreContext whose RUM session_id
+ * first-half bytes encode the given sentinel value, for use as a unique identity marker
+ * in delivery assertions.
+ */
+static ContextChangedMessage MakeMessage(uint64_t sentinel) {
+  CoreContext ctx = MOCK_CONTEXT;
+  ctx.rum.emplace();
+  std::memcpy(ctx.rum->session_id.bytes.data(), &sentinel, sizeof(sentinel));
+  return ContextChangedMessage{std::move(ctx)};
+}
+
+/**
+ * Extracts the sentinel value packed into the first half of the RUM session_id by
+ * MakeMessage.
+ */
+static uint64_t ReadSentinel(const ContextChangedMessage& msg) {
+  uint64_t value = 0;
+  if (msg.context.rum) {
+    std::memcpy(&value, msg.context.rum->session_id.bytes.data(), sizeof(value));
+  }
+  return value;
+}
+
+TEST_CASE("MessageBus", "[unit][core]") {
+  DiagnosticMessageBuffer diagnostics;
+  DiagnosticLogger logger = diagnostics.CreateTestLogger();
+
+  SECTION("M deliver message to single handler W one handler registered") {
+    // Given a bus with one handler that records delivered sentinel values
+    std::vector<uint64_t> received;
+    std::vector<std::function<void(const FeatureMessage&)>> handlers;
+    handlers.push_back([&](const FeatureMessage& msg) {
+      received.push_back(ReadSentinel(std::get<ContextChangedMessage>(msg)));
+    });
+
+    MessageBus bus(std::move(handlers));
+    std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus));
+
+    // When we send a message with a known sentinel and stop the bus
+    REQUIRE(bus.Send(MakeMessage(0xbeef)));
+    bus.Stop();
+    t.join();
+
+    // Then the handler received exactly that message
+    REQUIRE(received.size() == 1);
+    REQUIRE(received[0] == 0xbeef);
+  }
+
+  SECTION("M broadcast to all handlers W multiple handlers registered") {
+    // Given a bus with three handlers, each recording which sentinels it received
+    std::vector<uint64_t> received_a, received_b, received_c;
+    std::vector<std::function<void(const FeatureMessage&)>> handlers;
+    handlers.push_back([&](const FeatureMessage& msg) {
+      received_a.push_back(ReadSentinel(std::get<ContextChangedMessage>(msg)));
+    });
+    handlers.push_back([&](const FeatureMessage& msg) {
+      received_b.push_back(ReadSentinel(std::get<ContextChangedMessage>(msg)));
+    });
+    handlers.push_back([&](const FeatureMessage& msg) {
+      received_c.push_back(ReadSentinel(std::get<ContextChangedMessage>(msg)));
+    });
+
+    MessageBus bus(std::move(handlers));
+    std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus));
+
+    // When we send two messages and stop the bus
+    REQUIRE(bus.Send(MakeMessage(1)));
+    REQUIRE(bus.Send(MakeMessage(2)));
+    bus.Stop();
+    t.join();
+
+    // Then every handler received both messages, in order
+    const std::vector<uint64_t> expected{1, 2};
+    REQUIRE(received_a == expected);
+    REQUIRE(received_b == expected);
+    REQUIRE(received_c == expected);
+  }
+
+  SECTION("M drain all queued messages before messaging thread exits") {
+    // Given a bus with a handler that counts deliveries
+    std::atomic<int> count{0};
+    std::vector<std::function<void(const FeatureMessage&)>> handlers;
+    handlers.push_back([&](const FeatureMessage&) { count.fetch_add(1); });
+
+    MessageBus bus(std::move(handlers));
+    std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus));
+
+    // When we enqueue several messages and then stop
+    const int num_messages = 50;
+    for (int i = 0; i < num_messages; i++) {
+      bus.Send(MakeMessage(static_cast<uint64_t>(i)));
+    }
+    bus.Stop();
+    t.join();
+
+    // Then all messages were delivered before the thread exited
+    REQUIRE(count.load() == num_messages);
+  }
+
+  SECTION("M return false from Send() W bus is stopped") {
+    // Given a bus with no handlers
+    MessageBus bus({});
+    std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus));
+
+    // When the bus is stopped
+    bus.Stop();
+    t.join();
+
+    // Then Send() returns false and the message is dropped
+    REQUIRE_FALSE(bus.Send(MakeMessage(0xdead)));
+  }
+
+  SECTION("M not crash or block other handlers W one handler is a no-op") {
+    // Given a bus whose first handler does nothing (simulating a dead feature) and
+    // whose second handler records delivered sentinels
+    std::vector<uint64_t> received;
+    std::vector<std::function<void(const FeatureMessage&)>> handlers;
+    handlers.push_back([](const FeatureMessage&) {
+      // Deliberately empty — simulates a handler whose owning feature was torn down but
+      // whose lambda was not removed from the bus.
+    });
+    handlers.push_back([&](const FeatureMessage& msg) {
+      received.push_back(ReadSentinel(std::get<ContextChangedMessage>(msg)));
+    });
+
+    MessageBus bus(std::move(handlers));
+    std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus));
+
+    REQUIRE(bus.Send(MakeMessage(0x42)));
+    bus.Stop();
+    t.join();
+
+    // The active handler still received the message
+    REQUIRE(received.size() == 1);
+    REQUIRE(received[0] == 0x42);
+  }
+}