From b5902d0f5562fa6e685216f1e8c97b3770fcc3de Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 14:17:55 -0400 Subject: [PATCH 1/4] feat: Add MessageBus, to be used by the core to propagate messages --- src/CMakeLists.txt | 1 + src/datadog/impl/core/feature_message.hpp | 37 ++++++ src/datadog/impl/core/message_bus.cpp | 44 +++++++ src/datadog/impl/core/message_bus.hpp | 74 +++++++++++ tests/CMakeLists.txt | 1 + tests/impl/core/message_bus_test.cpp | 149 ++++++++++++++++++++++ 6 files changed, 306 insertions(+) create mode 100644 src/datadog/impl/core/feature_message.hpp create mode 100644 src/datadog/impl/core/message_bus.cpp create mode 100644 src/datadog/impl/core/message_bus.hpp create mode 100644 tests/impl/core/message_bus_test.cpp diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 5334b790..4c325615 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -26,6 +26,7 @@ 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/core.cpp datadog/impl/core/feature_read.cpp datadog/impl/core/feature_scope.cpp diff --git a/src/datadog/impl/core/feature_message.hpp b/src/datadog/impl/core/feature_message.hpp new file mode 100644 index 00000000..ea42d966 --- /dev/null +++ b/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 + +#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; + +} // namespace datadog::impl diff --git a/src/datadog/impl/core/message_bus.cpp b/src/datadog/impl/core/message_bus.cpp new file mode 100644 index 00000000..770e453c --- /dev/null +++ b/src/datadog/impl/core/message_bus.cpp @@ -0,0 +1,44 @@ +// 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> handlers, + const DiagnosticLogger& logger +) + : _handlers(std::move(handlers)), + _logger(logger), + _thread(&MessageBus::ThreadMain, this) {} + +MessageBus::~MessageBus() { + // The owner must call Stop() before destruction; if the thread is still joinable here + // the queue was never stopped, which is a programming error. + DATADOG_ASSERT(!_thread.joinable(), "MessageBus destroyed without calling Stop()"); +} + +bool MessageBus::Send(FeatureMessage msg) { return _queue.Push(std::move(msg)); } + +void MessageBus::Stop() { + _queue.Stop(); + _thread.join(); +} + +void MessageBus::ThreadMain() { + _logger.Debug("Message bus thread starting"); + + while (auto msg = _queue.Pop()) { + for (auto& handler : _handlers) { + handler(*msg); + } + } + + _logger.Debug("Message bus thread finished"); +} + +} // namespace datadog::impl diff --git a/src/datadog/impl/core/message_bus.hpp b/src/datadog/impl/core/message_bus.hpp new file mode 100644 index 00000000..ff82c6b1 --- /dev/null +++ b/src/datadog/impl/core/message_bus.hpp @@ -0,0 +1,74 @@ +// 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 +#include +#include + +#include "datadog/impl/core/feature_message.hpp" +#include "datadog/impl/core/queue.hpp" +#include "datadog/impl/diagnostics.hpp" + +namespace datadog::impl { + +/** + * Owns a background thread that delivers `FeatureMessage` values to a fixed set of + * subscriber callbacks. The bus broadcasts every message to all handlers in + * registration order. + * + * Handlers are supplied at construction time and never change after `Start` (which is + * called by the constructor). This means no synchronization is required inside + * `ThreadMain` — the handler list is immutable for the lifetime of the thread. + * + * `Send()` enqueues a message non-blocking from any thread and returns false if the bus + * has already been stopped. `Stop()` drains all remaining queued messages and joins the + * thread before returning, so callers can rely on all in-flight messages having been + * delivered once `Stop()` returns. + */ +class MessageBus { + public: + /** + * Constructs a `MessageBus` and starts the background delivery thread immediately. + * `handlers` is the complete, immutable list of callbacks that will receive every + * message; `logger` is used to emit debug messages at thread start and stop. + */ + explicit MessageBus( + std::vector> handlers, + const DiagnosticLogger& logger + ); + + ~MessageBus(); + + 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 bus + * has been stopped and the message was therefore dropped. + */ + bool Send(FeatureMessage msg); + + /** + * Stops the bus: no further messages will be accepted, all messages already in the + * queue will be delivered, and the background thread will be joined. Must be called + * before the `MessageBus` is destroyed. + */ + void Stop(); + + private: + void ThreadMain(); + + Queue _queue; + std::vector> _handlers; + DiagnosticLogger _logger; + std::thread _thread; +}; + +} // namespace datadog::impl diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index dedc0a98..71343588 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -28,6 +28,7 @@ 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/core_test.cpp impl/core/feature_id_test.cpp impl/core/feature_read_test.cpp diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp new file mode 100644 index 00000000..2e810c39 --- /dev/null +++ b/tests/impl/core/message_bus_test.cpp @@ -0,0 +1,149 @@ +// 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 +#include +#include + +#include "datadog/impl/core/feature_message.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 received; + std::vector> handlers; + handlers.push_back([&](const FeatureMessage& msg) { + received.push_back(ReadSentinel(std::get(msg))); + }); + + MessageBus bus(std::move(handlers), logger); + + // When we send a message with a known sentinel and stop the bus + REQUIRE(bus.Send(MakeMessage(0xbeef))); + bus.Stop(); + + // 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 received_a, received_b, received_c; + std::vector> handlers; + handlers.push_back([&](const FeatureMessage& msg) { + received_a.push_back(ReadSentinel(std::get(msg))); + }); + handlers.push_back([&](const FeatureMessage& msg) { + received_b.push_back(ReadSentinel(std::get(msg))); + }); + handlers.push_back([&](const FeatureMessage& msg) { + received_c.push_back(ReadSentinel(std::get(msg))); + }); + + MessageBus bus(std::move(handlers), logger); + + // When we send two messages and stop the bus + REQUIRE(bus.Send(MakeMessage(1))); + REQUIRE(bus.Send(MakeMessage(2))); + bus.Stop(); + + // Then every handler received both messages, in order + const std::vector expected{1, 2}; + REQUIRE(received_a == expected); + REQUIRE(received_b == expected); + REQUIRE(received_c == expected); + } + + SECTION("M drain all queued messages before Stop() returns") { + // Given a bus with a handler that counts deliveries + std::atomic count{0}; + std::vector> handlers; + handlers.push_back([&](const FeatureMessage&) { count.fetch_add(1); }); + + MessageBus bus(std::move(handlers), logger); + + // 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(i))); + } + bus.Stop(); + + // Then all messages were delivered before Stop() returned + REQUIRE(count.load() == num_messages); + } + + SECTION("M return false from Send() W bus is stopped") { + // Given a bus with no handlers + MessageBus bus({}, logger); + + // When the bus is stopped + bus.Stop(); + + // 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 received; + std::vector> 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(msg))); + }); + + MessageBus bus(std::move(handlers), logger); + + REQUIRE(bus.Send(MakeMessage(0x42))); + bus.Stop(); + + // The active handler still received the message + REQUIRE(received.size() == 1); + REQUIRE(received[0] == 0x42); + } +} From 89cb4d8c3195c8094c1368250d5a5bbe6675bace Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 14:42:45 -0400 Subject: [PATCH 2/4] chore: Refactor messaging thread out of MessageBus This pattern is consistent with how the SDK currently manages all Core-owned background threads: for the sake of clarity, each dedicated background thread gets its own header (e.g. `messaging_thread.hpp`) and its own consistently-named entry point (e.g. `MessagingThreadMain`), and the Core explicitly manages the lifetime of each thread, rather than adding layers of indirection to thread startup/shutdown logic by encapsulating them in other classes. In other words, data (e.g. `MessageBus`) is kept entirely separate from the Core-owned `std::thread` that runs the thread's logic (e.g. `MessagingThreadMain`). If we change this pattern, we should do so consistently for all threads' implementations. --- src/CMakeLists.txt | 1 + src/datadog/impl/core/message_bus.cpp | 32 +-------- src/datadog/impl/core/message_bus.hpp | 50 ++++---------- src/datadog/impl/core/messaging_thread.cpp | 23 +++++++ src/datadog/impl/core/messaging_thread.hpp | 24 +++++++ tests/CMakeLists.txt | 1 + tests/impl/core/message_bus_test.cpp | 34 ++++++--- tests/impl/core/messaging_thread_test.cpp | 80 ++++++++++++++++++++++ 8 files changed, 167 insertions(+), 78 deletions(-) create mode 100644 src/datadog/impl/core/messaging_thread.cpp create mode 100644 src/datadog/impl/core/messaging_thread.hpp create mode 100644 tests/impl/core/messaging_thread_test.cpp diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 4c325615..02021f94 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -27,6 +27,7 @@ add_library(dd_native ${DD_LIBRARY_TYPE} 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 diff --git a/src/datadog/impl/core/message_bus.cpp b/src/datadog/impl/core/message_bus.cpp index 770e453c..675f3170 100644 --- a/src/datadog/impl/core/message_bus.cpp +++ b/src/datadog/impl/core/message_bus.cpp @@ -8,37 +8,9 @@ namespace datadog::impl { -MessageBus::MessageBus( - std::vector> handlers, - const DiagnosticLogger& logger -) - : _handlers(std::move(handlers)), - _logger(logger), - _thread(&MessageBus::ThreadMain, this) {} - -MessageBus::~MessageBus() { - // The owner must call Stop() before destruction; if the thread is still joinable here - // the queue was never stopped, which is a programming error. - DATADOG_ASSERT(!_thread.joinable(), "MessageBus destroyed without calling Stop()"); -} +MessageBus::MessageBus(std::vector> handlers) + : _handlers(std::move(handlers)) {} bool MessageBus::Send(FeatureMessage msg) { return _queue.Push(std::move(msg)); } -void MessageBus::Stop() { - _queue.Stop(); - _thread.join(); -} - -void MessageBus::ThreadMain() { - _logger.Debug("Message bus thread starting"); - - while (auto msg = _queue.Pop()) { - for (auto& handler : _handlers) { - handler(*msg); - } - } - - _logger.Debug("Message bus thread finished"); -} - } // namespace datadog::impl diff --git a/src/datadog/impl/core/message_bus.hpp b/src/datadog/impl/core/message_bus.hpp index ff82c6b1..067dd493 100644 --- a/src/datadog/impl/core/message_bus.hpp +++ b/src/datadog/impl/core/message_bus.hpp @@ -7,68 +7,44 @@ #pragma once #include -#include #include #include "datadog/impl/core/feature_message.hpp" #include "datadog/impl/core/queue.hpp" -#include "datadog/impl/diagnostics.hpp" namespace datadog::impl { /** - * Owns a background thread that delivers `FeatureMessage` values to a fixed set of - * subscriber callbacks. The bus broadcasts every message to all handlers in - * registration order. + * 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 never change after `Start` (which is - * called by the constructor). This means no synchronization is required inside - * `ThreadMain` — the handler list is immutable for the lifetime of the 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 bus - * has already been stopped. `Stop()` drains all remaining queued messages and joins the - * thread before returning, so callers can rely on all in-flight messages having been - * delivered once `Stop()` returns. + * `Send()` enqueues a message non-blocking from any thread and returns false if the + * queue has been stopped (i.e., `_queue.Stop()` has been called). */ class MessageBus { public: - /** - * Constructs a `MessageBus` and starts the background delivery thread immediately. - * `handlers` is the complete, immutable list of callbacks that will receive every - * message; `logger` is used to emit debug messages at thread start and stop. - */ - explicit MessageBus( - std::vector> handlers, - const DiagnosticLogger& logger - ); - - ~MessageBus(); + explicit MessageBus(std::vector> 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 bus - * has been stopped and the message was therefore dropped. + * 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); - /** - * Stops the bus: no further messages will be accepted, all messages already in the - * queue will be delivered, and the background thread will be joined. Must be called - * before the `MessageBus` is destroyed. - */ - void Stop(); - - private: - void ThreadMain(); - Queue _queue; std::vector> _handlers; - DiagnosticLogger _logger; - std::thread _thread; }; } // namespace datadog::impl diff --git a/src/datadog/impl/core/messaging_thread.cpp b/src/datadog/impl/core/messaging_thread.cpp new file mode 100644 index 00000000..741a93b6 --- /dev/null +++ b/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 diff --git a/src/datadog/impl/core/messaging_thread.hpp b/src/datadog/impl/core/messaging_thread.hpp new file mode 100644 index 00000000..d8a60e6c --- /dev/null +++ b/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 diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 71343588..a8fd23f7 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -29,6 +29,7 @@ add_executable(dd_native_tests 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 diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp index 2e810c39..753797ec 100644 --- a/tests/impl/core/message_bus_test.cpp +++ b/tests/impl/core/message_bus_test.cpp @@ -8,9 +8,11 @@ #include #include +#include #include #include "datadog/impl/core/feature_message.hpp" +#include "datadog/impl/core/messaging_thread.hpp" #include "support/catch.hpp" #include "support/context.hpp" @@ -55,11 +57,13 @@ TEST_CASE("MessageBus", "[unit][core]") { received.push_back(ReadSentinel(std::get(msg))); }); - MessageBus bus(std::move(handlers), logger); + 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(); + bus._queue.Stop(); + t.join(); // Then the handler received exactly that message REQUIRE(received.size() == 1); @@ -80,12 +84,14 @@ TEST_CASE("MessageBus", "[unit][core]") { received_c.push_back(ReadSentinel(std::get(msg))); }); - MessageBus bus(std::move(handlers), logger); + 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(); + bus._queue.Stop(); + t.join(); // Then every handler received both messages, in order const std::vector expected{1, 2}; @@ -100,25 +106,29 @@ TEST_CASE("MessageBus", "[unit][core]") { std::vector> handlers; handlers.push_back([&](const FeatureMessage&) { count.fetch_add(1); }); - MessageBus bus(std::move(handlers), logger); + 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(i))); } - bus.Stop(); + bus._queue.Stop(); + t.join(); - // Then all messages were delivered before Stop() returned + // 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({}, logger); + MessageBus bus({}); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); // When the bus is stopped - bus.Stop(); + bus._queue.Stop(); + t.join(); // Then Send() returns false and the message is dropped REQUIRE_FALSE(bus.Send(MakeMessage(0xdead))); @@ -137,10 +147,12 @@ TEST_CASE("MessageBus", "[unit][core]") { received.push_back(ReadSentinel(std::get(msg))); }); - MessageBus bus(std::move(handlers), logger); + MessageBus bus(std::move(handlers)); + std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); REQUIRE(bus.Send(MakeMessage(0x42))); - bus.Stop(); + bus._queue.Stop(); + t.join(); // The active handler still received the message REQUIRE(received.size() == 1); diff --git a/tests/impl/core/messaging_thread_test.cpp b/tests/impl/core/messaging_thread_test.cpp new file mode 100644 index 00000000..f843494b --- /dev/null +++ b/tests/impl/core/messaging_thread_test.cpp @@ -0,0 +1,80 @@ +// 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" + +#include "datadog/impl/core/feature_message.hpp" +#include "datadog/impl/core/message_bus.hpp" + +#include "support/catch.hpp" +#include "support/context.hpp" +#include "support/diagnostics.hpp" + +using namespace datadog; +using namespace datadog::impl; + +// Tests run MessagingThreadMain synchronously by stopping the queue before calling the +// function, avoiding any background-thread synchronization. + +TEST_CASE("MessagingThreadMain", "[unit][core]") { + DiagnosticMessageBuffer diagnostics; + DiagnosticLogger logger = diagnostics.CreateTestLogger(); + + SECTION("M log debug messages on start and stop W run on empty stopped bus") { + // Given an empty bus with no handlers and a stopped queue + MessageBus bus({}); + bus._queue.Stop(); + + // When we run the messaging thread synchronously + MessagingThreadMain(logger, bus); + + // Then exactly the start and stop messages are emitted — nothing else + REQUIRE(diagnostics.debug.size() == 2); + REQUIRE(diagnostics.debug[0] == "Messaging thread starting"); + REQUIRE(diagnostics.debug[1] == "Messaging thread finished"); + REQUIRE(diagnostics.TotalSize() == 2); + } + + SECTION("M deliver enqueued messages to all handlers in order") { + // Given a bus with two handlers that each record which messages they receive + std::vector received_a; + std::vector received_b; + + std::vector> handlers; + handlers.push_back([&](const FeatureMessage& msg) { + const auto& ccm = std::get(msg); + int value = 0; + std::memcpy(&value, ccm.context.rum->session_id.bytes.data(), sizeof(value)); + received_a.push_back(value); + }); + handlers.push_back([&](const FeatureMessage& msg) { + const auto& ccm = std::get(msg); + int value = 0; + std::memcpy(&value, ccm.context.rum->session_id.bytes.data(), sizeof(value)); + received_b.push_back(value); + }); + + MessageBus bus(std::move(handlers)); + + // Enqueue two messages, each carrying a distinct int sentinel + for (int i : {1, 2}) { + CoreContext ctx = MOCK_CONTEXT; + ctx.rum.emplace(); + std::memcpy(ctx.rum->session_id.bytes.data(), &i, sizeof(i)); + bus._queue.Push(ContextChangedMessage{std::move(ctx)}); + } + bus._queue.Stop(); + + // When we run the messaging thread synchronously + MessagingThreadMain(logger, bus); + + // Then both handlers received both messages in order, with no extra diagnostics + REQUIRE(received_a == std::vector{1, 2}); + REQUIRE(received_b == std::vector{1, 2}); + REQUIRE(diagnostics.debug.size() == 2); + REQUIRE(diagnostics.debug.size() == diagnostics.TotalSize()); + } +} From 161feb5a4803275d148d47c7151805d1f0e03cd9 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 14:49:18 -0400 Subject: [PATCH 3/4] chore: Make MessageBus members private --- src/datadog/impl/core/message_bus.cpp | 2 ++ src/datadog/impl/core/message_bus.hpp | 14 +++++++++++++- tests/impl/core/message_bus_test.cpp | 10 +++++----- tests/impl/core/messaging_thread_test.cpp | 6 +++--- 4 files changed, 23 insertions(+), 9 deletions(-) diff --git a/src/datadog/impl/core/message_bus.cpp b/src/datadog/impl/core/message_bus.cpp index 675f3170..30cbc9d0 100644 --- a/src/datadog/impl/core/message_bus.cpp +++ b/src/datadog/impl/core/message_bus.cpp @@ -13,4 +13,6 @@ MessageBus::MessageBus(std::vector> h bool MessageBus::Send(FeatureMessage msg) { return _queue.Push(std::move(msg)); } +void MessageBus::Stop() { _queue.Stop(); } + } // namespace datadog::impl diff --git a/src/datadog/impl/core/message_bus.hpp b/src/datadog/impl/core/message_bus.hpp index 067dd493..49c49c83 100644 --- a/src/datadog/impl/core/message_bus.hpp +++ b/src/datadog/impl/core/message_bus.hpp @@ -11,6 +11,7 @@ #include "datadog/impl/core/feature_message.hpp" #include "datadog/impl/core/queue.hpp" +#include "datadog/impl/diagnostics.hpp" namespace datadog::impl { @@ -25,7 +26,8 @@ namespace datadog::impl { * `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 (i.e., `_queue.Stop()` has been called). + * queue has been stopped. Call `Stop()` to signal shutdown; the messaging thread will + * drain any remaining messages before exiting. */ class MessageBus { public: @@ -43,6 +45,16 @@ class MessageBus { */ 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 _queue; std::vector> _handlers; }; diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp index 753797ec..a898622d 100644 --- a/tests/impl/core/message_bus_test.cpp +++ b/tests/impl/core/message_bus_test.cpp @@ -62,7 +62,7 @@ TEST_CASE("MessageBus", "[unit][core]") { // When we send a message with a known sentinel and stop the bus REQUIRE(bus.Send(MakeMessage(0xbeef))); - bus._queue.Stop(); + bus.Stop(); t.join(); // Then the handler received exactly that message @@ -90,7 +90,7 @@ TEST_CASE("MessageBus", "[unit][core]") { // When we send two messages and stop the bus REQUIRE(bus.Send(MakeMessage(1))); REQUIRE(bus.Send(MakeMessage(2))); - bus._queue.Stop(); + bus.Stop(); t.join(); // Then every handler received both messages, in order @@ -114,7 +114,7 @@ TEST_CASE("MessageBus", "[unit][core]") { for (int i = 0; i < num_messages; i++) { bus.Send(MakeMessage(static_cast(i))); } - bus._queue.Stop(); + bus.Stop(); t.join(); // Then all messages were delivered before the thread exited @@ -127,7 +127,7 @@ TEST_CASE("MessageBus", "[unit][core]") { std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); // When the bus is stopped - bus._queue.Stop(); + bus.Stop(); t.join(); // Then Send() returns false and the message is dropped @@ -151,7 +151,7 @@ TEST_CASE("MessageBus", "[unit][core]") { std::thread t(MessagingThreadMain, std::ref(logger), std::ref(bus)); REQUIRE(bus.Send(MakeMessage(0x42))); - bus._queue.Stop(); + bus.Stop(); t.join(); // The active handler still received the message diff --git a/tests/impl/core/messaging_thread_test.cpp b/tests/impl/core/messaging_thread_test.cpp index f843494b..6194738f 100644 --- a/tests/impl/core/messaging_thread_test.cpp +++ b/tests/impl/core/messaging_thread_test.cpp @@ -26,7 +26,7 @@ TEST_CASE("MessagingThreadMain", "[unit][core]") { SECTION("M log debug messages on start and stop W run on empty stopped bus") { // Given an empty bus with no handlers and a stopped queue MessageBus bus({}); - bus._queue.Stop(); + bus.Stop(); // When we run the messaging thread synchronously MessagingThreadMain(logger, bus); @@ -64,9 +64,9 @@ TEST_CASE("MessagingThreadMain", "[unit][core]") { CoreContext ctx = MOCK_CONTEXT; ctx.rum.emplace(); std::memcpy(ctx.rum->session_id.bytes.data(), &i, sizeof(i)); - bus._queue.Push(ContextChangedMessage{std::move(ctx)}); + bus.Send(ContextChangedMessage{std::move(ctx)}); } - bus._queue.Stop(); + bus.Stop(); // When we run the messaging thread synchronously MessagingThreadMain(logger, bus); From 463b8f67a2279f8b13693f50b43ea4a998f49039 Mon Sep 17 00:00:00 2001 From: Alex Forsythe Date: Wed, 11 Mar 2026 14:53:07 -0400 Subject: [PATCH 4/4] chore: Missing cstring include, clarified test name --- tests/impl/core/message_bus_test.cpp | 2 +- tests/impl/core/messaging_thread_test.cpp | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp index a898622d..26b6248b 100644 --- a/tests/impl/core/message_bus_test.cpp +++ b/tests/impl/core/message_bus_test.cpp @@ -100,7 +100,7 @@ TEST_CASE("MessageBus", "[unit][core]") { REQUIRE(received_c == expected); } - SECTION("M drain all queued messages before Stop() returns") { + SECTION("M drain all queued messages before messaging thread exits") { // Given a bus with a handler that counts deliveries std::atomic count{0}; std::vector> handlers; diff --git a/tests/impl/core/messaging_thread_test.cpp b/tests/impl/core/messaging_thread_test.cpp index 6194738f..eded1ca1 100644 --- a/tests/impl/core/messaging_thread_test.cpp +++ b/tests/impl/core/messaging_thread_test.cpp @@ -6,6 +6,8 @@ #include "datadog/impl/core/messaging_thread.hpp" +#include + #include "datadog/impl/core/feature_message.hpp" #include "datadog/impl/core/message_bus.hpp"