diff --git a/changelog/changelog b/changelog/changelog index 3eaa760..37833b7 100644 --- a/changelog/changelog +++ b/changelog/changelog @@ -2,9 +2,19 @@ ## Bug fixes +- [#16](https://github.com/openDAQ/LTStreamingModulesModern/pull/16) Resolve hidden domain (time) signals referenced via the `relatedSignals` metadata element, fixing devices that connected successfully but delivered no data (signals had no domain signal). +- [#16](https://github.com/openDAQ/LTStreamingModulesModern/pull/16) Retry initial metadata-fetch subscribes via a sweep timer, fixing signals that silently never appeared on devices that drop concurrent command-interface requests. +- [#16](https://github.com/openDAQ/LTStreamingModulesModern/pull/16) Defer publishing a signal until its domain signal is published, so the domain link is not lost when metadata arrives out of order; signals whose domain signal never publishes are published without the link after a timeout, with a warning. +- [#18](https://github.com/openDAQ/LTStreamingModulesModern/pull/18) Handle subscribe/unsubscribe requests on the I/O thread to avoid a data race on the signal map. +- [#18](https://github.com/openDAQ/LTStreamingModulesModern/pull/18) Guard signal publication so exceptions can no longer escape into the Boost.Asio I/O thread and terminate the process. +- [#19](https://github.com/openDAQ/LTStreamingModulesModern/pull/19) Hand the initial-fetch subscription over to an immediately subscribing application, fixing data that stopped on devices that reorder back-to-back unsubscribe/subscribe pairs. + ## Documenation ## Misc -- [#2](https://github.com/openDAQ/LTStreamingModulesModern/pull/2) Import files from openDAQ SDK +- [#16](https://github.com/openDAQ/LTStreamingModulesModern/pull/16) Bump ws-streaming to v3.1.1, fixing warnings-as-errors builds (GCC `-Werror=reorder`, MSVC `/WX` C5038). + - [#1](https://github.com/openDAQ/LTStreamingModulesModern/pull/1) Project initial version. +- [#2](https://github.com/openDAQ/LTStreamingModulesModern/pull/2) Import files from openDAQ SDK +- [#8](https://github.com/openDAQ/LTStreamingModulesModern/pull/8) Bumped to 4.0.0 — major version raised to reflect the module's move to a standalone release cycle, decoupled from openDAQ core versioning. diff --git a/module_version b/module_version index f83d792..f0f011f 100644 --- a/module_version +++ b/module_version @@ -1 +1 @@ -4.1.0dev \ No newline at end of file +4.1.0dev diff --git a/modules/websocket_streaming_client_module/tests/CMakeLists.txt b/modules/websocket_streaming_client_module/tests/CMakeLists.txt index d56cac7..e24aa51 100644 --- a/modules/websocket_streaming_client_module/tests/CMakeLists.txt +++ b/modules/websocket_streaming_client_module/tests/CMakeLists.txt @@ -2,6 +2,7 @@ set(MODULE_NAME ws_stream_cl_module) set(TEST_APP test_${MODULE_NAME}) set(TEST_SOURCES test_websocket_streaming_client_module.cpp + test_device_compatibility.cpp test_app.cpp ) diff --git a/modules/websocket_streaming_client_module/tests/test_device_compatibility.cpp b/modules/websocket_streaming_client_module/tests/test_device_compatibility.cpp new file mode 100644 index 0000000..e3c0cf4 --- /dev/null +++ b/modules/websocket_streaming_client_module/tests/test_device_compatibility.cpp @@ -0,0 +1,662 @@ +/* + * End-to-end tests for the streaming client against a fake LT peer that mimics devices which + * do not advertise their time signals ("hidden" domain signals referenced via "relatedSignals" + * with an abstract table id) and which may drop or reorder command-interface requests. + * + * The fake peer accepts a WebSocket upgrade on a raw TCP socket (the client only checks the + * HTTP status line) and then speaks the LT streaming protocol through the ws-streaming + * library's own low-level peer class, with full control over metadata content, ordering and + * request handling. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include + +#include +#include + +#include +#include +#include +#include + +using namespace daq; +using namespace std::chrono_literals; + +namespace +{ + +class FakeLtPeer +{ + public: + + struct Options + { + bool timeMetadataBeforeValue = true; // send the time signal's metadata before the value signal's + std::chrono::milliseconds timeMetadataDelay{0}; // extra delay before the time signal's metadata + bool withholdTimeMetadata = false; // announce the time signal but never send its metadata + unsigned dropSubscribeRequests = 0; // ignore this many leading value-signal subscribe requests + bool streamData = false; // stream value-signal data while it is subscribed + bool outOfOrderUnsubscribe = false; // defer a value-signal unsubscribe behind the next request + }; + + explicit FakeLtPeer(Options options) + : options(options) + , acceptor(ioc, boost::asio::ip::tcp::endpoint( + boost::asio::ip::make_address("127.0.0.1"), 0)) + , timer(ioc) + , dataTimer(ioc) + { + acceptor.async_accept( + [this](const boost::system::error_code& ec, boost::asio::ip::tcp::socket socket) + { + if (!ec) + handleAccept(std::move(socket)); + }); + + thread = std::thread([this] { ioc.run(); }); + } + + ~FakeLtPeer() + { + ioc.stop(); + thread.join(); + } + + std::uint16_t port() const + { + return acceptor.local_endpoint().port(); + } + + unsigned subscribeRequestCount() + { + std::scoped_lock lock(mutex); + return valueSubscribeRequests; + } + + unsigned timeSubscribeRequestCount() + { + std::scoped_lock lock(mutex); + return timeSubscribeRequests; + } + + unsigned valueUnsubscribeRequestCount() + { + std::scoped_lock lock(mutex); + return valueUnsubscribeRequests; + } + + // Unilaterally ends the value-signal subscription, like a device dropping it on its own + void unsubscribeValueSignal() + { + boost::asio::post(ioc, + [this] + { + if (!peer) + return; + valueSubscribed = false; + peer->send_metadata(valueSigno, "unsubscribe", nlohmann::json::object()); + }); + } + + // Advertises the previously hidden time signal in a second 'available' announcement + void advertiseTimeSignal() + { + boost::asio::post(ioc, + [this] + { + if (peer) + peer->send_metadata(0, "available", {{ "signalIds", { timeSignalId } }}); + }); + } + + private: + + void handleAccept(boost::asio::ip::tcp::socket socket) + { + auto sock = std::make_shared(std::move(socket)); + auto buffer = std::make_shared(); + + // consume the client's HTTP upgrade request; the client only checks the status line + boost::asio::async_read_until(*sock, *buffer, "\r\n\r\n", + [this, sock, buffer](const boost::system::error_code& ec, std::size_t) + { + if (ec) + return; + + auto response = std::make_shared( + "HTTP/1.1 101 Switching Protocols\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + "Sec-WebSocket-Accept: fake\r\n" + "\r\n"); + + boost::asio::async_write(*sock, boost::asio::buffer(*response), + [this, sock, response](const boost::system::error_code& ec, std::size_t) + { + if (!ec) + startPeer(std::move(*sock)); + }); + }); + } + + void startPeer(boost::asio::ip::tcp::socket socket) + { + peer = std::make_shared(std::move(socket), false); + + onMetadata = peer->on_metadata_received.connect( + [this](unsigned signo, const std::string& method, const nlohmann::json& params) + { + handleMetadata(signo, method, params); + }); + + peer->run(); + + peer->send_metadata(0, "apiVersion", {{ "version", "1.0.0" }}); + peer->send_metadata(0, "init", { + { "streamId", "FAKE" }, + { "commandInterfaces", { { "jsonrpc", nlohmann::json::object() } } }, + }); + peer->send_metadata(0, "available", {{ "signalIds", { valueSignalId } }}); + } + + void handleMetadata(unsigned /*signo*/, const std::string& method, const nlohmann::json& params) + { + if (method != "request" || !params.is_object()) + return; + + const auto id = params.value("id", nullptr); + const std::string rpcMethod = params.value("method", std::string()); + + std::string signalId; + if (params.contains("params") && params["params"].is_array() + && !params["params"].empty() && params["params"][0].is_string()) + signalId = params["params"][0]; + + if (rpcMethod == "FAKE.subscribe" && signalId == valueSignalId) + { + { + std::scoped_lock lock(mutex); + ++valueSubscribeRequests; + if (valueSubscribeRequests <= options.dropSubscribeRequests) + return; // simulate a dropped request: no response, no effect + } + + // a subscribe processed while still subscribed is rejected, then the deferred unsubscribe runs + if (valueSubscribed && options.outOfOrderUnsubscribe) + { + respondError(id); + flushPendingValueUnsubscribe(); + return; + } + + valueSubscribed = true; + respond(id, true); + sendValueSignalFamily(); + + if (options.streamData) + startStreamingData(); + } + + else if (rpcMethod == "FAKE.subscribe" && signalId == timeSignalId) + { + std::scoped_lock lock(mutex); + ++timeSubscribeRequests; + } + + else if (rpcMethod == "FAKE.unsubscribe") + { + if (signalId == valueSignalId) + { + if (options.outOfOrderUnsubscribe) + pendingValueUnsubscribeId = id; // sit on it until the next request + else + executeValueUnsubscribe(id); + return; + } + + respond(id, true); + } + } + + void executeValueUnsubscribe(const nlohmann::json& id) + { + { + std::scoped_lock lock(mutex); + ++valueUnsubscribeRequests; + } + + valueSubscribed = false; + respond(id, true); + peer->send_metadata(valueSigno, "unsubscribe", nlohmann::json::object()); + } + + void flushPendingValueUnsubscribe() + { + if (!pendingValueUnsubscribeId) + return; + + executeValueUnsubscribe(*pendingValueUnsubscribeId); + pendingValueUnsubscribeId.reset(); + } + + void respond(const nlohmann::json& id, bool result) + { + peer->send_metadata(0, "response", { + { "jsonrpc", "2.0" }, + { "id", id }, + { "result", result }, + }); + } + + void respondError(const nlohmann::json& id) + { + peer->send_metadata(0, "response", { + { "jsonrpc", "2.0" }, + { "id", id }, + { "error", { { "code", -32602 }, { "message", "already subscribed" } } }, + }); + } + + void startStreamingData() + { + // give the linear time table its start point, then pump value samples periodically + peer->send_data(timeSigno, boost::asio::buffer(&timeStart, sizeof(timeStart))); + sendValueData(); + } + + void sendValueData() + { + if (!valueSubscribed) + return; + + peer->send_data(valueSigno, boost::asio::buffer(valueSamples)); + + dataTimer.expires_after(20ms); + dataTimer.async_wait( + [this](const boost::system::error_code& ec) + { + if (!ec) + sendValueData(); + }); + } + + void sendValueSignalFamily() + { + // subscribing the value signal implicitly announces its hidden time signal + peer->send_metadata(valueSigno, "subscribe", {{ "signalId", valueSignalId }}); + peer->send_metadata(timeSigno, "subscribe", {{ "signalId", timeSignalId }}); + + if (options.timeMetadataBeforeValue) + { + sendTimeMetadata(); + sendValueMetadata(); + } + else + { + sendValueMetadata(); + + if (options.timeMetadataDelay.count() > 0) + { + timer.expires_after(options.timeMetadataDelay); + timer.async_wait( + [this](const boost::system::error_code& ec) + { + if (!ec) + sendTimeMetadata(); + }); + } + else + { + sendTimeMetadata(); + } + } + } + + void sendValueMetadata() + { + peer->send_metadata(valueSigno, "signal", { + { "tableId", tableId }, + { "relatedSignals", { + { { "type", "time" }, { "signalId", timeSignalId } }, + } }, + { "definition", { + { "name", valueSignalId }, + { "rule", "explicit" }, + { "dataType", "real32" }, + } }, + }); + } + + void sendTimeMetadata() + { + if (options.withholdTimeMetadata) + return; + + peer->send_metadata(timeSigno, "signal", { + { "tableId", tableId }, + { "definition", { + { "name", timeSignalId }, + { "rule", "linear" }, + { "linear", { { "delta", 1 } } }, + { "dataType", "uint64" }, + { "resolution", { { "num", 1 }, { "denom", 1000 } } }, + } }, + }); + } + + const std::string tableId = "CH1"; + const std::string valueSignalId = "CH1.value"; + const std::string timeSignalId = "CH1.time"; + static constexpr unsigned valueSigno = 1; + static constexpr unsigned timeSigno = 2; + + Options options; + + boost::asio::io_context ioc{1}; + boost::asio::ip::tcp::acceptor acceptor; + boost::asio::steady_timer timer; + boost::asio::steady_timer dataTimer; + std::thread thread; + + // sent by reference from the asynchronous send_data(), so they must outlive the calls + const wss::detail::streaming_protocol::linear_payload timeStart{0, 0}; + const std::vector valueSamples = std::vector(100, 1.0f); + + bool valueSubscribed = false; // only touched on the ioc thread + std::optional pendingValueUnsubscribeId; + + std::shared_ptr peer; + boost::signals2::scoped_connection onMetadata; + + std::mutex mutex; + unsigned valueSubscribeRequests = 0; + unsigned timeSubscribeRequests = 0; + unsigned valueUnsubscribeRequests = 0; +}; + +// An Instance owns the device so that teardown runs the device's removal path; +// creating a device directly from the module would leak it (and fail the leak listener) +InstancePtr createClientInstance() +{ + auto instance = Instance("[[none]]"); + + ModulePtr module; + createModule(&module, instance.getContext()); + instance.getModuleManager().addModule(module); + + return instance; +} + +DevicePtr connectDevice(const InstancePtr& instance, std::uint16_t port) +{ + return instance.addDevice("daq.lt://127.0.0.1:" + std::to_string(port) + "/"); +} + +// Polls until the device exposes the expected number of signals or the timeout elapses. +ListPtr waitForSignals( + const DevicePtr& device, + size_t expectedCount, + std::chrono::milliseconds timeout) +{ + const auto deadline = std::chrono::steady_clock::now() + timeout; + + ListPtr signals; + do + { + signals = device.getSignals(search::Recursive(search::Any())); + if (signals.getCount() >= expectedCount) + return signals; + std::this_thread::sleep_for(50ms); + } while (std::chrono::steady_clock::now() < deadline); + + return signals; +} + +SignalPtr findSignalByName(const ListPtr& signals, const std::string& name) +{ + for (const auto& signal : signals) + if (signal.getDescriptor().assigned() && signal.getDescriptor().getName() == name) + return signal; + return nullptr; +} + +// Bundles the connected client so tests keep the owning Instance alive +struct ClientSetup +{ + InstancePtr instance; + DevicePtr device; + ListPtr signals; +}; + +ClientSetup connectAndWaitForSignals( + const FakeLtPeer& peer, + size_t expectedCount = 2, + std::chrono::milliseconds timeout = 5s) +{ + ClientSetup setup; + setup.instance = createClientInstance(); + setup.device = connectDevice(setup.instance, peer.port()); + setup.signals = waitForSignals(setup.device, expectedCount, timeout); + return setup; +} + +// Builds a Float64/Int64 stream reader on the signal, subscribing it +auto buildStreamReader(const SignalPtr& signal) +{ + return daq::StreamReaderBuilder() + .setSignal(signal) + .setValueReadType(daq::SampleType::Float64) + .setDomainReadType(daq::SampleType::Int64) + .setSkipEvents(true) + .build(); +} + +} // namespace + +using DeviceCompatibilityTest = testing::Test; + +TEST_F(DeviceCompatibilityTest, HiddenDomainSignalIsLinked) +{ + FakeLtPeer peer({}); + auto [instance, device, signals] = connectAndWaitForSignals(peer); + ASSERT_EQ(signals.getCount(), 2u); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + auto timeSignal = findSignalByName(signals, "CH1.time"); + ASSERT_TRUE(valueSignal.assigned()); + ASSERT_TRUE(timeSignal.assigned()); + + ASSERT_TRUE(valueSignal.getDomainSignal().assigned()); + ASSERT_EQ(valueSignal.getDomainSignal(), timeSignal); +} + +TEST_F(DeviceCompatibilityTest, DomainMetadataArrivingLateIsStillLinked) +{ + FakeLtPeer::Options options; + options.timeMetadataBeforeValue = false; + options.timeMetadataDelay = 300ms; + + FakeLtPeer peer(options); + auto [instance, device, signals] = connectAndWaitForSignals(peer); + ASSERT_EQ(signals.getCount(), 2u); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + ASSERT_TRUE(valueSignal.assigned()); + + // the value signal must have been deferred until the time signal published + ASSERT_TRUE(valueSignal.getDomainSignal().assigned()); +} + +TEST_F(DeviceCompatibilityTest, DroppedSubscribeRequestIsRetried) +{ + FakeLtPeer::Options options; + options.dropSubscribeRequests = 1; + + FakeLtPeer peer(options); + + // the sweep timer retries after 1.5 s + 100 ms; allow generous margin + auto [instance, device, signals] = connectAndWaitForSignals(peer, 2, 10s); + ASSERT_EQ(signals.getCount(), 2u); + ASSERT_GE(peer.subscribeRequestCount(), 2u); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + ASSERT_TRUE(valueSignal.assigned()); + ASSERT_TRUE(valueSignal.getDomainSignal().assigned()); +} + +TEST_F(DeviceCompatibilityTest, ReadvertisedHiddenDomainSignalStartsNoNewFetch) +{ + FakeLtPeer peer({}); + auto [instance, device, signals] = connectAndWaitForSignals(peer); + ASSERT_EQ(signals.getCount(), 2u); + + // the device now advertises the already-published hidden time signal + peer.advertiseTimeSignal(); + std::this_thread::sleep_for(500ms); + + // no duplicate signal, the domain link is intact, and no fetch subscribe was sent for it + signals = device.getSignals(search::Recursive(search::Any())); + ASSERT_EQ(signals.getCount(), 2u); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + ASSERT_TRUE(valueSignal.assigned()); + ASSERT_TRUE(valueSignal.getDomainSignal().assigned()); + + ASSERT_EQ(peer.timeSubscribeRequestCount(), 0u); +} + +TEST_F(DeviceCompatibilityTest, ImmediateSubscribeTakesOverFetchSubscription) +{ + FakeLtPeer peer({}); + auto [instance, device, signals] = connectAndWaitForSignals(peer); + ASSERT_EQ(signals.getCount(), 2u); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + ASSERT_TRUE(valueSignal.assigned()); + + // subscribe immediately after the signal appeared, like an auto-subscribing application + auto mirrored = valueSignal.asPtr(); + std::promise ackPromise; + auto ackFuture = ackPromise.get_future(); + mirrored.getOnSubscribeComplete() += + [&ackPromise](MirroredSignalConfigPtr&, SubscriptionEventArgsPtr&) { ackPromise.set_value(); }; + + auto reader = buildStreamReader(valueSignal); + + // the takeover must acknowledge the subscription without any wire traffic + ASSERT_EQ(ackFuture.wait_for(3s), std::future_status::ready); + + // give the sweep time to have (wrongly) released the fetch subscription + std::this_thread::sleep_for(4s); + + EXPECT_EQ(peer.subscribeRequestCount(), 1u); // only the initial fetch subscribed + EXPECT_EQ(peer.valueUnsubscribeRequestCount(), 0u); // never released: taken over +} + +// Without the takeover, a device swapping the release unsubscribe with the app subscribe stops the data +TEST_F(DeviceCompatibilityTest, DataKeepsFlowingWhenDeviceReordersUnsubscribeAndSubscribe) +{ + FakeLtPeer::Options options; + options.streamData = true; + options.outOfOrderUnsubscribe = true; + FakeLtPeer peer(options); + + auto [instance, device, signals] = connectAndWaitForSignals(peer); + ASSERT_EQ(signals.getCount(), 2u); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + ASSERT_TRUE(valueSignal.assigned()); + + // subscribe immediately after the signal appeared, like an auto-subscribing application + auto reader = buildStreamReader(valueSignal); + + // wait past the sweep window and discard the startup burst (the failure mode goes silent after it) + std::this_thread::sleep_for(4s); + + if (SizeT count = reader.getAvailableCount(); count > 0) + { + std::vector values(count); + std::vector domain(count); + reader.readWithDomain(values.data(), domain.data(), &count); + } + + std::this_thread::sleep_for(500ms); + EXPECT_GT(reader.getAvailableCount(), 0u); +} + +// A remote unsubscribe must end the held fetch subscription; a takeover of it would never get data +TEST_F(DeviceCompatibilityTest, SubscribeAfterRemoteUnsubscribeSendsNewRequest) +{ + FakeLtPeer::Options options; + options.streamData = true; + FakeLtPeer peer(options); + + auto [instance, device, signals] = connectAndWaitForSignals(peer); + ASSERT_EQ(signals.getCount(), 2u); + + // the device drops the subscription on its own while it is held for takeover + peer.unsubscribeValueSignal(); + std::this_thread::sleep_for(500ms); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + ASSERT_TRUE(valueSignal.assigned()); + + auto reader = buildStreamReader(valueSignal); + + std::this_thread::sleep_for(1s); + EXPECT_EQ(peer.subscribeRequestCount(), 2u); // a real second subscribe request was sent + EXPECT_GT(reader.getAvailableCount(), 0u); // and data flows again +} + +TEST_F(DeviceCompatibilityTest, UnusedFetchSubscriptionIsReleasedBySweep) +{ + FakeLtPeer peer({}); + auto [instance, device, signals] = connectAndWaitForSignals(peer); + ASSERT_EQ(signals.getCount(), 2u); + + // nobody subscribes: the sweep must release the held fetch subscription + std::this_thread::sleep_for(4s); + + EXPECT_EQ(peer.subscribeRequestCount(), 1u); + EXPECT_EQ(peer.valueUnsubscribeRequestCount(), 1u); +} + +TEST_F(DeviceCompatibilityTest, SignalPublishesWithoutDomainWhenMetadataNeverArrives) +{ + FakeLtPeer::Options options; + options.timeMetadataBeforeValue = false; + options.withholdTimeMetadata = true; + + FakeLtPeer peer(options); + + // deferral gives up after two sweep periods (~3 s); allow generous margin + auto [instance, device, signals] = connectAndWaitForSignals(peer, 1, 10s); + ASSERT_EQ(signals.getCount(), 1u); + + auto valueSignal = findSignalByName(signals, "CH1.value"); + ASSERT_TRUE(valueSignal.assigned()); + ASSERT_FALSE(valueSignal.getDomainSignal().assigned()); +} diff --git a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h index becc226..d8df97a 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h @@ -25,6 +25,7 @@ #include #include +#include #include #include #include @@ -47,12 +48,13 @@ BEGIN_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING * This object uses the ws-streaming library to establish a WebSocket streaming connection to a * remote peer. * - * openDAQ requires signal objects to have a valid descriptor, even if they are not connected. In - * contrast, the WebSocket Streaming protocol only provides signal metadata after a signal has - * been subscribed. To solve this, the streaming object does not immediately register signals with - * openDAQ when they become available. Instead, it does an initial subscribe. When the signal's - * metadata is received, the signal is then unsubscribed, and registered with openDAQ using - * addToAvailableSignals() now that its metadata is known and a valid descriptor can be created. + * openDAQ signals need a valid descriptor, but the WebSocket Streaming protocol only provides + * metadata once a signal is subscribed. New signals are therefore first subscribed to fetch + * their metadata, and registered via addToAvailableSignals() once a descriptor can be built. + * The fetch subscription is briefly kept afterwards so an immediate application subscribe can + * take it over without wire traffic (devices may process a back-to-back unsubscribe/subscribe + * pair out of order); a sweep timer releases it if unused. A takeover replays the cached + * descriptor to openDAQ. * * Once registered with openDAQ, the onAddSignal() and onRemoveSignal() functions are implemented * to manage the subscription state of each known signal. When data is received for an active @@ -152,6 +154,15 @@ class WsStreaming : public Streaming const boost::system::error_code& ec, wss::connection_ptr connection); + /*! @brief I/O-thread implementation of onSubscribeSignal(). */ + void subscribeRemoteSignal(const std::string& signalId); + + /*! @brief I/O-thread implementation of onUnsubscribeSignal(). */ + void unsubscribeRemoteSignal(const std::string& signalId); + + /*! @brief Creates a tracking entry for a remote signal and connects its event slots. */ + std::shared_ptr createSignalEntry(wss::remote_signal_ptr signal); + void onRemoteSignalAvailable(wss::remote_signal_ptr signal); void onRemoteSignalSubscribed(std::weak_ptr weakEntry); @@ -169,6 +180,23 @@ class WsStreaming : public Streaming void onRemoteSignalUnavailable(wss::remote_signal_ptr signal); + /*! @brief Finds a signal's domain entry by table ID or via "relatedSignals", discovering hidden domain signals on demand. */ + std::shared_ptr resolveDomainEntry( + const std::shared_ptr& entry); + + /*! @brief Registers a signal with openDAQ, marks its initial-fetch subscription as held for takeover and publishes signals deferred on it. */ + void publishSignalEntry(const std::shared_ptr& entry); + + /*! @brief Pushes the entry's cached descriptor into openDAQ as descriptor-changed events, propagating to signals that use it as their domain. */ + void emitDescriptorChangedEvents(const std::shared_ptr& entry); + + /*! @brief Checks whether any signal is still awaiting initial metadata or deferred on an unpublished domain signal. */ + bool anyInitialFetchPending() const; + + void armInitialFetchSweep(); + void onInitialFetchSweep(const boost::system::error_code& ec); + void onInitialFetchResubscribe(const boost::system::error_code& ec); + boost::asio::io_context ioContext; std::thread thread; @@ -180,6 +208,10 @@ class WsStreaming : public Streaming std::map> signals; + /** Re-subscribes signals whose initial metadata fetch was dropped (some devices drop concurrent requests). */ + boost::asio::steady_timer initialFetchTimer; + bool initialFetchSweepArmed = false; + std::promise promise; }; diff --git a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_remote_signal_entry.h b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_remote_signal_entry.h index ab5c1d0..faac45e 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_remote_signal_entry.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming_remote_signal_entry.h @@ -28,6 +28,15 @@ BEGIN_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING struct WsStreamingRemoteSignalEntry { + /** The lifecycle of the initial metadata-fetch subscription (see the WsStreaming class doc). */ + enum class FetchState + { + None, /**< No fetch subscription exists: never fetched, given up, released or taken over. */ + Fetching, /**< A fetch wire subscription is active, awaiting the signal's metadata. */ + AwaitingRetry, /**< The sweep unsubscribed a fetch that yielded no metadata; a retry subscribe is due. */ + Held, /**< Published, but the fetch subscription is kept for an application subscribe to take over. */ + }; + /** * The ws-streaming library's remote signal object. This object is created by ws-streaming * when the remote peer advertises a signal as 'available'. It is released (but not @@ -63,6 +72,10 @@ struct WsStreamingRemoteSignalEntry bool isPublished = false; bool isSubscribed = false; + + FetchState fetchState = FetchState::None; /**< State of the initial metadata-fetch subscription. */ + unsigned fetchAttempts = 0; /**< Subscribe requests sent by the initial metadata fetch. */ + unsigned sweeps = 0; /**< Sweep passes in the current wait (deferral before publication, hold after); the wait ends at 2. */ }; END_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING diff --git a/shared/libraries/websocket_streaming/src/ws_streaming.cpp b/shared/libraries/websocket_streaming/src/ws_streaming.cpp index 5c83460..a5c0c73 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming.cpp @@ -15,6 +15,7 @@ */ #include +#include #include #include #include @@ -42,6 +43,13 @@ using namespace std::placeholders; BEGIN_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING +using FetchState = WsStreamingRemoteSignalEntry::FetchState; + +// Initial-fetch retry tuning, measured against a real device (burst metadata <= ~1.1 s; >= 50 ms unsub/sub gap keeps ordering). +static constexpr unsigned INITIAL_FETCH_MAX_ATTEMPTS = 3; +static constexpr std::chrono::milliseconds INITIAL_FETCH_TIMEOUT{1500}; +static constexpr std::chrono::milliseconds INITIAL_FETCH_RETRY_DELAY{100}; + namespace { bool isTlsRejection(bool isSecureChannel, const boost::system::error_code& ec) @@ -87,6 +95,7 @@ WsStreaming::WsStreaming( : Streaming(connectionString, context, true) , ioContext{1} , wsClient(ioContext.get_executor()) + , initialFetchTimer(ioContext) { // NOTE! The 'port' property is not used there. The formed 'connectionString' // must contain the port number. @@ -319,6 +328,20 @@ void WsStreaming::onRemoveSignal(const MirroredSignalConfigPtr& signal) } void WsStreaming::onSubscribeSignal(const StringPtr& signalId) +{ + // called on application threads; the signals map may only be touched on the I/O thread + boost::asio::post(ioContext, + [this, id = signalId.toStdString()] { subscribeRemoteSignal(id); }); +} + +void WsStreaming::onUnsubscribeSignal(const StringPtr& signalId) +{ + // called on application threads; the signals map may only be touched on the I/O thread + boost::asio::post(ioContext, + [this, id = signalId.toStdString()] { unsubscribeRemoteSignal(id); }); +} + +void WsStreaming::subscribeRemoteSignal(const std::string& signalId) { LOG_I("Asked to subscribe signal {}", signalId); @@ -339,6 +362,19 @@ void WsStreaming::onSubscribeSignal(const StringPtr& signalId) return; } + // take over the wire subscription still held by the initial fetch: zero wire traffic + if (signalIt->second->fetchState == FetchState::Held) + { + LOG_I("Found signal, taking over the initial-fetch subscription"); + signalIt->second->fetchState = FetchState::None; + signalIt->second->isSubscribed = true; + triggerSubscribeAck(signalId, true); + + if (signalIt->second->descriptor.assigned()) + emitDescriptorChangedEvents(signalIt->second); + return; + } + LOG_I("Found signal, subscribing"); signalIt->second->ptr->subscribe(); signalIt->second->isSubscribed = true; @@ -350,7 +386,7 @@ void WsStreaming::onSubscribeSignal(const StringPtr& signalId) } } -void WsStreaming::onUnsubscribeSignal(const StringPtr& signalId) +void WsStreaming::unsubscribeRemoteSignal(const std::string& signalId) { LOG_I("Asked to unsubscribe signal {}", signalId); @@ -391,9 +427,11 @@ void WsStreaming::onConnected( std::bind(&WsStreaming::onRemoteSignalUnavailable, this, _1)); } -void WsStreaming::onRemoteSignalAvailable(wss::remote_signal_ptr signal) +std::shared_ptr WsStreaming::createSignalEntry(wss::remote_signal_ptr signal) { - LOG_I("Signal available: {}", signal->id()); + // an entry may already exist if a signal discovered as a hidden domain is later advertised + if (auto it = signals.find(signal->id()); it != signals.end()) + return it->second; auto entry = std::make_shared(); entry->ptr = signal; @@ -405,11 +443,128 @@ void WsStreaming::onRemoteSignalAvailable(wss::remote_signal_ptr signal) entry->onDataReceived = signal->on_data_received .connect(std::bind(&WsStreaming::onRemoteSignalDataReceived, this, weakEntry, _1, _2, _3, _4)); entry->onUnsubscribed = signal->on_unsubscribed .connect(std::bind(&WsStreaming::onRemoteSignalUnsubscribed, this, weakEntry)); - signals[signal->id()] = std::move(entry); + signals[signal->id()] = entry; + + return entry; +} + +void WsStreaming::onRemoteSignalAvailable(wss::remote_signal_ptr signal) +{ + LOG_I("Signal available: {}", signal->id()); + + auto entry = createSignalEntry(signal); + + // a reused entry (a hidden domain signal the device later advertises) needs no new fetch: + // it is already published, has its metadata, or a fetch is already in flight + if (entry->isPublished || entry->descriptor.assigned() || entry->fetchState == FetchState::Fetching) + return; // Do not immediately register the new signal with openDAQ. We need its metadata first so // we can make an openDAQ descriptor. Do an initial subscribe to get that metadata. + entry->fetchState = FetchState::Fetching; + entry->fetchAttempts = 1; signal->subscribe(); + + armInitialFetchSweep(); +} + +void WsStreaming::armInitialFetchSweep() +{ + if (initialFetchSweepArmed) + return; + + initialFetchSweepArmed = true; + initialFetchTimer.expires_after(INITIAL_FETCH_TIMEOUT); + initialFetchTimer.async_wait(std::bind(&WsStreaming::onInitialFetchSweep, this, _1)); +} + +void WsStreaming::onInitialFetchSweep(const boost::system::error_code& ec) +{ + if (ec == boost::asio::error::operation_aborted) + return; + + bool retrying = false; + + for (const auto& [id, entry] : signals) + { + if (entry->isPublished) + { + // release a held fetch subscription only after a full sweep period, clear of app subscribes + if (entry->fetchState == FetchState::Held && ++entry->sweeps >= 2) + { + entry->fetchState = FetchState::None; + entry->ptr->unsubscribe(); + } + continue; + } + + // deferred entry: metadata arrived but the domain signal hasn't published; + // grant at least one full sweep period before dropping the domain link + if (entry->descriptor.assigned()) + { + if (++entry->sweeps >= 2) + publishSignalEntry(entry); + continue; + } + + if (entry->fetchState != FetchState::Fetching) + continue; + + entry->ptr->unsubscribe(); + + if (entry->fetchAttempts >= INITIAL_FETCH_MAX_ATTEMPTS) + { + LOG_W("No metadata received for signal {}; giving up (without a descriptor the signal cannot be added to openDAQ)", id); + entry->fetchState = FetchState::None; + entry->fetchAttempts = 0; + } + else + { + entry->fetchState = FetchState::AwaitingRetry; + ++entry->fetchAttempts; + retrying = true; + } + } + + if (retrying) + { + // stay marked busy so a new arrival cannot re-arm the timer and cancel this resubscribe + initialFetchTimer.expires_after(INITIAL_FETCH_RETRY_DELAY); + initialFetchTimer.async_wait(std::bind(&WsStreaming::onInitialFetchResubscribe, this, _1)); + } + + else + { + initialFetchSweepArmed = false; + + // signals may still be awaiting metadata or deferred on a domain signal + if (anyInitialFetchPending()) + armInitialFetchSweep(); + } +} + +void WsStreaming::onInitialFetchResubscribe(const boost::system::error_code& ec) +{ + if (ec == boost::asio::error::operation_aborted) + return; + + initialFetchSweepArmed = false; + bool pending = false; + + for (const auto& [id, entry] : signals) + { + // an assigned descriptor means metadata already arrived (entry is deferred, not lost) + if (entry->fetchState != FetchState::AwaitingRetry || entry->descriptor.assigned()) + continue; + + entry->fetchState = FetchState::Fetching; + entry->ptr->subscribe(); + pending = true; + } + + // also stay armed for fetches that arrived during the retry delay and deferred entries + if (pending || anyInitialFetchPending()) + armInitialFetchSweep(); } void WsStreaming::onRemoteSignalSubscribed(std::weak_ptr weakEntry) @@ -439,19 +594,15 @@ void WsStreaming::onRemoteSignalMetadataChanged(std::weak_ptrlastPacket = nullptr; entry->descriptor = metadataToDescriptor(entry->ptr->metadata()); + entry->domainEntry = resolveDomainEntry(entry); - std::string tableId = entry->ptr->metadata().table_id(); - - if (auto it = signals.find(tableId); it != signals.end() - && it->second != entry) + if (entry->domainEntry) { - entry->domainEntry = it->second; - LOG_I("Signal {} domain now points to {}", entry->ptr->id(), entry->domainEntry->ptr->id()); + LOG_D("Signal {} domain now points to {}", entry->ptr->id(), entry->domainEntry->ptr->id()); } else { - LOG_I("Signal {} domain now points to nullptr", entry->ptr->id()); - entry->domainEntry = nullptr; + LOG_D("Signal {} domain now points to nullptr", entry->ptr->id()); } } @@ -464,33 +615,149 @@ void WsStreaming::onRemoteSignalMetadataChanged(std::weak_ptrdescriptor.assigned() && entry->isPublished) + emitDescriptorChangedEvents(entry); + if (entry->descriptor.assigned() && !entry->isPublished) { - auto packet = DataDescriptorChangedEventPacket(entry->descriptor, nullptr); - onPacket(entry->ptr->id(), packet); - - // changed signal is time signal - if (entry->ptr->id() == entry->ptr->metadata().table_id()) + // defer until the domain publishes: publishSignalEntry() then publishes this signal too; + // if the domain never publishes, the sweep publishes this signal without the domain link + if (entry->domainEntry && !entry->domainEntry->isPublished) { - packet = DataDescriptorChangedEventPacket(nullptr, entry->descriptor); - for (const auto& [id, dataSignalEntry] : signals) - { - if (dataSignalEntry->ptr->metadata().table_id() == entry->ptr->id() && - dataSignalEntry != entry) - onPacket(dataSignalEntry->ptr->id(), packet); - } + LOG_D("Deferring signal {} until its domain signal {} is published", + entry->ptr->id(), entry->domainEntry->ptr->id()); + armInitialFetchSweep(); + } + else + { + publishSignalEntry(entry); } } - if (entry->descriptor.assigned() && !entry->isPublished) +} + +void WsStreaming::emitDescriptorChangedEvents(const std::shared_ptr& entry) +{ + auto packet = DataDescriptorChangedEventPacket(entry->descriptor, nullptr); + onPacket(entry->ptr->id(), packet); + + // propagate to signals that use this signal as their domain + packet = DataDescriptorChangedEventPacket(nullptr, entry->descriptor); + for (const auto& [id, dataSignalEntry] : signals) + { + if (dataSignalEntry != entry && + dataSignalEntry->domainEntry == entry && + dataSignalEntry->isPublished) + onPacket(dataSignalEntry->ptr->id(), packet); + } +} + +void WsStreaming::publishSignalEntry(const std::shared_ptr& entry) +{ + LOG_I("Signal {} is now ready, publishing it", entry->ptr->id()); + + if (entry->domainEntry && !entry->domainEntry->isPublished) + { + LOG_W("Signal {} is published without a link to its domain signal {}, which was never published", + entry->ptr->id(), entry->domainEntry->ptr->id()); + } + + entry->isPublished = true; + + // a throw from here would escape into the Boost.Asio I/O thread and terminate the process + try { - LOG_I("Signal {} is now ready, publishing it", entry->ptr->id()); - entry->isPublished = true; addToAvailableSignals(entry->ptr->id()); onSignalAvailable( entry->ptr, - entry->domainEntry ? entry->domainEntry->ptr : nullptr, + entry->domainEntry && entry->domainEntry->isPublished ? entry->domainEntry->ptr : nullptr, entry->descriptor); - entry->ptr->unsubscribe(); } + + catch (const std::exception& ex) + { + LOG_E("Failed to register signal {} with openDAQ: {}", entry->ptr->id(), ex.what()); + } + + // keep the fetch subscription so an immediate application subscribe can take it over + if (entry->fetchState == FetchState::Fetching) + { + entry->fetchState = FetchState::Held; + armInitialFetchSweep(); + } + else + entry->fetchState = FetchState::None; + + entry->fetchAttempts = 0; + entry->sweeps = 0; // the counter now times the hold instead of the deferral + + // publish signals that were deferred waiting for this signal as their domain + for (const auto& [id, dependent] : signals) + if (dependent != entry && dependent->domainEntry == entry + && !dependent->isPublished && dependent->descriptor.assigned()) + publishSignalEntry(dependent); +} + +bool WsStreaming::anyInitialFetchPending() const +{ + // deferred entries (descriptor but unpublished) and held fetch subscriptions still need the sweep + for (const auto& [id, entry] : signals) + if (entry->fetchState == FetchState::Held + || (!entry->isPublished && (entry->fetchState == FetchState::Fetching || entry->descriptor.assigned()))) + return true; + + return false; +} + +std::shared_ptr WsStreaming::resolveDomainEntry( + const std::shared_ptr& entry) +{ + std::string tableId = entry->ptr->metadata().table_id(); + + if (tableId.empty() || tableId == entry->ptr->id()) + return nullptr; + + // openDAQ servers advertise domain signals and use the domain signal's ID as the table ID + if (auto it = signals.find(tableId); it != signals.end() && it->second != entry) + return it->second; + + // other devices reference a hidden domain signal via "relatedSignals" + std::string domainSignalId; + const auto& metadataJson = entry->ptr->metadata().json(); + + if (auto relatedIt = metadataJson.find("relatedSignals"); + relatedIt != metadataJson.end() && relatedIt->is_array()) + for (const auto& related : *relatedIt) + if (related.is_object() + && related.value("type", std::string()) == "time" + && related.contains("signalId") + && related["signalId"].is_string()) + { + domainSignalId = related["signalId"]; + break; // first "time" entry wins + } + + if (domainSignalId.empty() || domainSignalId == entry->ptr->id()) + return nullptr; + + if (auto it = signals.find(domainSignalId); it != signals.end() && it->second != entry) + return it->second; + + if (!wsConnection) + return nullptr; + + // the connection knows hidden signals: the peer's acks/metadata create remote signal objects + auto domainSignal = wsConnection->find_remote_signal(domainSignalId); + if (!domainSignal) + return nullptr; + + LOG_D("Discovered hidden domain signal {} of signal {}", domainSignalId, entry->ptr->id()); + + auto domainEntry = createSignalEntry(domainSignal); + + // process its already-received metadata now so it publishes before the referencing signal + // (if the metadata hasn't arrived yet, its later arrival publishes the entry instead) + if (!domainSignal->metadata().json().empty()) + onRemoteSignalMetadataChanged(domainEntry); + + return domainEntry; } void WsStreaming::onRemoteSignalDataReceived( @@ -562,6 +829,13 @@ void WsStreaming::onRemoteSignalUnsubscribed(std::weak_ptrptr->id()); + // a remote unsubscribe ends a held fetch subscription: release it so a later subscribe sends a request + if (entry->fetchState == FetchState::Held) + { + entry->fetchState = FetchState::None; + entry->ptr->unsubscribe(); + } + if (entry->isSubscribed) { entry->isSubscribed = false;