diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 5334b790..02021f94 100644 --- a/src/CMakeLists.txt +++ b/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 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..30cbc9d0 --- /dev/null +++ b/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> handlers) + : _handlers(std::move(handlers)) {} + +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 new file mode 100644 index 00000000..49c49c83 --- /dev/null +++ b/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 +#include + +#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> 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 _queue; + std::vector> _handlers; +}; + +} // 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 dedc0a98..a8fd23f7 100644 --- a/tests/CMakeLists.txt +++ b/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 diff --git a/tests/impl/core/message_bus_test.cpp b/tests/impl/core/message_bus_test.cpp new file mode 100644 index 00000000..26b6248b --- /dev/null +++ b/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 +#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" +#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)); + 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 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)); + 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 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 count{0}; + std::vector> 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(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 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)); + 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); + } +} diff --git a/tests/impl/core/messaging_thread_test.cpp b/tests/impl/core/messaging_thread_test.cpp new file mode 100644 index 00000000..eded1ca1 --- /dev/null +++ b/tests/impl/core/messaging_thread_test.cpp @@ -0,0 +1,82 @@ +// 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 + +#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.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.Send(ContextChangedMessage{std::move(ctx)}); + } + bus.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()); + } +}