From 05b7315c6a8faa20ae89246857f41d397e0dd184 Mon Sep 17 00:00:00 2001 From: Aliaksandr Adziareika <8034372+alexadereyko@users.noreply.github.com> Date: Fri, 29 May 2026 14:16:26 +0200 Subject: [PATCH 1/6] Bump version to 4.0.0 (#8) * Bump version to 4.0.0 * Update changelog 4.0.0 --- changelog/changelog | 3 ++- module_version | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/changelog/changelog b/changelog/changelog index 3eaa760..4d6baec 100644 --- a/changelog/changelog +++ b/changelog/changelog @@ -6,5 +6,6 @@ ## Misc -- [#2](https://github.com/openDAQ/LTStreamingModulesModern/pull/2) Import files from openDAQ SDK - [#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 c06dab4..0c89fc9 100644 --- a/module_version +++ b/module_version @@ -1 +1 @@ -3.31.0dev \ No newline at end of file +4.0.0 \ No newline at end of file From de47e61572b34f6178b180176275c5ebabb18b1c Mon Sep 17 00:00:00 2001 From: Viacheslav Kalenikov Date: Wed, 12 Aug 2026 14:32:43 +0200 Subject: [PATCH 2/6] ws-streaming ref updating --- external/ws-streaming/CMakeLists.txt | 6 +- .../patches/0001-Fixes-plus.patch | 545 ------------------ 2 files changed, 2 insertions(+), 549 deletions(-) delete mode 100644 external/ws-streaming/patches/0001-Fixes-plus.patch diff --git a/external/ws-streaming/CMakeLists.txt b/external/ws-streaming/CMakeLists.txt index d9a5b0b..03f83a9 100644 --- a/external/ws-streaming/CMakeLists.txt +++ b/external/ws-streaming/CMakeLists.txt @@ -2,10 +2,8 @@ set(WS_STREAMING_INSTALL OFF) opendaq_dependency( NAME ws-streaming - REQUIRED_VERSION 3.0.7 + REQUIRED_VERSION 3.1.0 GIT_REPOSITORY https://github.com/openDAQ/ws-streaming - GIT_REF 9d591b39ec4845a9f65c558411d809800d455a5d + GIT_REF v3.1.0 EXPECT_TARGET ws-streaming::ws-streaming - PATCH_FILES - ${CMAKE_CURRENT_SOURCE_DIR}/patches/0001-Fixes-plus.patch ) diff --git a/external/ws-streaming/patches/0001-Fixes-plus.patch b/external/ws-streaming/patches/0001-Fixes-plus.patch deleted file mode 100644 index 57b413a..0000000 --- a/external/ws-streaming/patches/0001-Fixes-plus.patch +++ /dev/null @@ -1,545 +0,0 @@ -diff --git a/include/ws-streaming/connection.hpp b/include/ws-streaming/connection.hpp -index bf3c0ec..2471f0e 100644 ---- a/include/ws-streaming/connection.hpp -+++ b/include/ws-streaming/connection.hpp -@@ -62,6 +62,7 @@ namespace wss - connection( - boost::asio::ip::tcp::socket&& socket, - bool is_client, -+ std::string local_stream_id, - bool use_tcp_protocol = false); - - /** -diff --git a/include/ws-streaming/detail/peer.hpp b/include/ws-streaming/detail/peer.hpp -index 7c07a2b..8970385 100644 ---- a/include/ws-streaming/detail/peer.hpp -+++ b/include/ws-streaming/detail/peer.hpp -@@ -368,7 +368,7 @@ namespace wss::detail - template - void enqueue( - const ConstBufferSequence& buffers, -- std::size_t size, -+ std::size_t /*size*/, - bool do_shutdown_after) - { - std::size_t bytes_buffered = boost::asio::buffer_copy( -diff --git a/include/ws-streaming/listener.hpp b/include/ws-streaming/listener.hpp -index fc402d3..1d2494a 100644 ---- a/include/ws-streaming/listener.hpp -+++ b/include/ws-streaming/listener.hpp -@@ -105,9 +105,13 @@ namespace wss - - void do_accept() - { -- using namespace std::placeholders; - acceptor.async_accept( -- std::bind(&listener::finish_accept, this->shared_from_this(), _1, _2)); -+ [self_weak = this->weak_from_this()](auto&& ec, auto&& socket) { -+ if (auto self = self_weak.lock()) { -+ self->finish_accept(ec, std::forward(socket)); -+ } -+ } -+ ); - } - - void finish_accept( -diff --git a/src/client.cpp b/src/client.cpp -index bda2e68..031892a 100644 ---- a/src/client.cpp -+++ b/src/client.cpp -@@ -58,9 +58,22 @@ void wss::client::async_connect( - if (response.result() != boost::beast::http::status::switching_protocols) - return handler(boost::beast::http::error::bad_status, {}); - -+ std::string connection_local_stream_id; -+ try -+ { -+ auto remote_endpoint = stream.socket().remote_endpoint(); -+ connection_local_stream_id = remote_endpoint.address().to_string() -+ + ":" + std::to_string(remote_endpoint.port()); -+ } -+ catch (const std::exception& /*e*/) -+ { -+ return; -+ } -+ - auto connection = std::make_shared( - stream.release_socket(), -- true); -+ true, -+ connection_local_stream_id); - - auto data = buffer.data(); - connection->run(data.data(), data.size()); -@@ -99,9 +112,22 @@ void wss::client::async_connect( - if (ec) - return handler(ec, {}); - -+ std::string connection_local_stream_id; -+ try -+ { -+ auto remote_endpoint = socket->remote_endpoint(); -+ connection_local_stream_id = remote_endpoint.address().to_string() -+ + ":" + std::to_string(remote_endpoint.port()); -+ } -+ catch (const std::exception& /*e*/) -+ { -+ return; -+ } -+ - auto connection = std::make_shared( - std::move(*socket), - true, -+ connection_local_stream_id, - true); - - connection->run(); -diff --git a/src/connection.cpp b/src/connection.cpp -index b0a1c52..ff8b9cc 100644 ---- a/src/connection.cpp -+++ b/src/connection.cpp -@@ -28,11 +28,11 @@ using namespace std::placeholders; - wss::connection::connection( - boost::asio::ip::tcp::socket&& socket, - bool is_client, -+ std::string local_stream_id, - bool use_tcp_protocol) - : _is_client{is_client} - , _peer{std::make_shared(std::move(socket), is_client, use_tcp_protocol)} -- , _local_stream_id{_peer->socket().remote_endpoint().address().to_string() -- + ":" + std::to_string(_peer->socket().remote_endpoint().port())} -+ , _local_stream_id{local_stream_id} - { - _command_interfaces["jsonrpc"] = { { "httpMethod", "" } }; - } -@@ -51,9 +51,25 @@ void wss::connection::register_external_command_interface( - - void wss::connection::run() - { -- _on_peer_data_received = _peer->on_data_received.connect(std::bind(&connection::on_peer_data_received, shared_from_this(), _1, _2, _3)); -- _on_peer_metadata_received = _peer->on_metadata_received.connect(std::bind(&connection::on_peer_metadata_received, shared_from_this(), _1, _2, _3)); -- _on_peer_closed = _peer->on_closed.connect(std::bind(&connection::on_peer_closed, shared_from_this(), _1)); -+ auto self_weak = weak_from_this(); -+ -+ _on_peer_data_received = _peer->on_data_received.connect( -+ [self_weak](unsigned signo, const std::uint8_t* data, std::size_t size) { -+ if (auto self = self_weak.lock()) -+ self->on_peer_data_received(signo, data, size); -+ }); -+ -+ _on_peer_metadata_received = _peer->on_metadata_received.connect( -+ [self_weak](unsigned signo, const std::string& method, const nlohmann::json& params) { -+ if (auto self = self_weak.lock()) -+ self->on_peer_metadata_received(signo, method, params); -+ }); -+ -+ _on_peer_closed = _peer->on_closed.connect( -+ [self_weak](const boost::system::error_code& ec) { -+ if (auto self = self_weak.lock()) -+ self->on_peer_closed(ec); -+ }); - - _peer->run(); - -@@ -63,9 +79,28 @@ void wss::connection::run() - - void wss::connection::run(const void *data, std::size_t size) - { -- _on_peer_data_received = _peer->on_data_received.connect(std::bind(&connection::on_peer_data_received, shared_from_this(), _1, _2, _3)); -- _on_peer_metadata_received = _peer->on_metadata_received.connect(std::bind(&connection::on_peer_metadata_received, shared_from_this(), _1, _2, _3)); -- _on_peer_closed = _peer->on_closed.connect(std::bind(&connection::on_peer_closed, shared_from_this(), _1)); -+ auto self_weak = weak_from_this(); -+ -+ _on_peer_data_received = _peer->on_data_received.connect( -+ [self_weak](unsigned signo, const std::uint8_t* data, std::size_t size) -+ { -+ if (auto self = self_weak.lock()) -+ self->on_peer_data_received(signo, data, size); -+ }); -+ -+ _on_peer_metadata_received = _peer->on_metadata_received.connect( -+ [self_weak](unsigned signo, const std::string& method, const nlohmann::json& params) -+ { -+ if (auto self = self_weak.lock()) -+ self->on_peer_metadata_received(signo, method, params); -+ }); -+ -+ _on_peer_closed = _peer->on_closed.connect( -+ [self_weak](const boost::system::error_code& ec) -+ { -+ if (auto self = self_weak.lock()) -+ self->on_peer_closed(ec); -+ }); - - _peer->run(data, size); - -@@ -266,6 +301,7 @@ void wss::connection::on_local_signal_metadata_changed( - } - } - -+ entry.value_index = 0; - _peer->send_metadata(entry.signo, "signal", entry.signal.metadata().json()); - } - -@@ -316,7 +352,7 @@ void wss::connection::on_signal_subscribe_requested( - return; // @todo XXX TODO - - _command_interface_client->async_request(_remote_stream_id + ".subscribe", { signal_id }, -- [](const boost::system::error_code& ec, const nlohmann::json& response) -+ [](const boost::system::error_code& /*ec*/, const nlohmann::json& /*response*/) - { - }); - } -@@ -328,7 +364,7 @@ void wss::connection::on_signal_unsubscribe_requested( - return; // @todo XXX TODO - - _command_interface_client->async_request(_remote_stream_id + ".unsubscribe", { signal_id }, -- [](const boost::system::error_code& ec, const nlohmann::json& response) -+ [](const boost::system::error_code& /*ec*/, const nlohmann::json& /*response*/) - { - }); - } -@@ -615,21 +651,25 @@ bool wss::connection::subscribe( - "signal", - metadata); - -+ auto self_weak = weak_from_this(); -+ auto signal_weak = std::weak_ptr(signal); -+ - signal->on_data_published = signal->signal.on_data_published.connect( -- std::bind( -- &connection::on_local_signal_data_published, -- shared_from_this(), -- signal, -- _1, -- _2, -- _3, -- _4)); -+ [self_weak, signal_weak](std::int64_t domain_value, -+ std::size_t sample_count, -+ const void* data, -+ std::size_t size) { -+ if (auto self = self_weak.lock()) -+ if (auto s = signal_weak.lock()) -+ self->on_local_signal_data_published(s, domain_value, sample_count, data, size); -+ }); - - signal->on_metadata_changed = signal->signal.on_metadata_changed.connect( -- std::bind( -- &connection::on_local_signal_metadata_changed, -- shared_from_this(), -- std::ref(*signal))); -+ [self_weak, signal_weak]() { -+ if (auto self = self_weak.lock()) -+ if (auto s = signal_weak.lock()) -+ self->on_local_signal_metadata_changed(*s); -+ }); - - return true; - } -diff --git a/src/detail/command_interface_client_factory.cpp b/src/detail/command_interface_client_factory.cpp -index 7f8eea0..f101571 100644 ---- a/src/detail/command_interface_client_factory.cpp -+++ b/src/detail/command_interface_client_factory.cpp -@@ -38,9 +38,20 @@ wss::detail::command_interface_client_factory::create_client( - else - port = interfaces["jsonrpc-http"]["port"]; - -+ std::string remote_endpoint_address; -+ try -+ { -+ auto remote_endpoint = peer->socket().remote_endpoint(); -+ remote_endpoint_address = remote_endpoint.address().to_string(); -+ } -+ catch (const std::exception& /*e*/) -+ { -+ return nullptr; -+ } -+ - return std::make_unique( - peer->socket().get_executor(), -- peer->socket().remote_endpoint().address().to_string(), -+ remote_endpoint_address, - port, - interfaces["jsonrpc-http"]["httpMethod"], - interfaces["jsonrpc-http"]["httpPath"], -diff --git a/src/detail/http_client.cpp b/src/detail/http_client.cpp -index a64aa03..211c1f6 100644 ---- a/src/detail/http_client.cpp -+++ b/src/detail/http_client.cpp -@@ -53,11 +53,12 @@ void wss::detail::http_client::async_request( - _resolver.async_resolve( - hostname, - port, -- std::bind( -- &http_client::finish_resolve, -- shared_from_this(), -- std::placeholders::_1, -- std::placeholders::_2)); -+ [self_weak = weak_from_this()](const boost::system::error_code& ec, -+ const boost::asio::ip::tcp::resolver::results_type& results) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_resolve(ec, results); -+ }); - } - - void wss::detail::http_client::cancel() -@@ -75,10 +76,11 @@ void wss::detail::http_client::finish_resolve( - - _stream.async_connect( - results, -- std::bind( -- &http_client::finish_connect, -- shared_from_this(), -- std::placeholders::_1)); -+ [self_weak = weak_from_this()](const boost::system::error_code& ec, auto /*endpoint*/) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_connect(ec); // finish_connect only takes ec -+ }); - } - - void wss::detail::http_client::finish_connect( -@@ -92,10 +94,11 @@ void wss::detail::http_client::finish_connect( - boost::beast::http::async_write( - _stream, - _request, -- std::bind( -- &http_client::finish_write, -- shared_from_this(), -- std::placeholders::_1)); -+ [self_weak = weak_from_this()](const boost::system::error_code& ec, std::size_t /*bytes_transferred*/) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_write(ec); -+ }); - } - - void wss::detail::http_client::finish_write( -@@ -110,10 +113,11 @@ void wss::detail::http_client::finish_write( - _stream, - _buffer, - _response, -- std::bind( -- &http_client::finish_read, -- shared_from_this(), -- std::placeholders::_1)); -+ [self_weak = weak_from_this()](const boost::system::error_code& ec, std::size_t /*bytes_transferred*/) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_read(ec); -+ }); - } - - void wss::detail::http_client::finish_read( -diff --git a/src/detail/http_client_servicer.cpp b/src/detail/http_client_servicer.cpp -index eeb1352..86be5f6 100644 ---- a/src/detail/http_client_servicer.cpp -+++ b/src/detail/http_client_servicer.cpp -@@ -40,9 +40,10 @@ void wss::detail::http_client_servicer::stop() - { - boost::asio::post( - stream.get_executor(), -- [self = shared_from_this()]() -+ [self_weak = weak_from_this()]() - { -- self->close(); -+ if (auto self = self_weak.lock()) -+ self->close(); - }); - } - -@@ -54,11 +55,11 @@ void wss::detail::http_client_servicer::do_read() - stream, - buffer, - req = {}, -- std::bind( -- &http_client_servicer::finish_read, -- shared_from_this(), -- _1, -- _2)); -+ [self_weak = weak_from_this()](const boost::system::error_code& ec, std::size_t bytes_transferred) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_read(ec, bytes_transferred); -+ }); - } - - void wss::detail::http_client_servicer::do_write( -@@ -68,17 +69,16 @@ void wss::detail::http_client_servicer::do_write( - boost::beast::async_write( - stream, - std::move(msg), -- std::bind( -- &http_client_servicer::finish_write, -- shared_from_this(), -- action, -- _1, -- _2)); -+ [self_weak = weak_from_this(), action](const boost::system::error_code& ec, std::size_t bytes_transferred) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_write(action, ec, bytes_transferred); -+ }); - } - - void wss::detail::http_client_servicer::finish_read( - const boost::system::error_code& ec, -- std::size_t bytes_transferred) -+ std::size_t /*bytes_transferred*/) - { - if (ec) - return close(ec); -@@ -197,7 +197,7 @@ void wss::detail::http_client_servicer::finish_read( - void wss::detail::http_client_servicer::finish_write( - response_actions action, - const boost::beast::error_code& ec, -- std::size_t bytes_transferred) -+ std::size_t /*bytes_transferred*/) - { - if (ec) - return close(ec); -diff --git a/src/detail/http_command_interface_client.cpp b/src/detail/http_command_interface_client.cpp -index 1932625..ea1244d 100644 ---- a/src/detail/http_command_interface_client.cpp -+++ b/src/detail/http_command_interface_client.cpp -@@ -66,8 +66,8 @@ void wss::detail::http_command_interface_client::async_request( - [this, handler = std::move(handler), client]( - const boost::system::error_code& ec, - const boost::beast::http::response& response, -- boost::beast::tcp_stream& stream, -- const boost::beast::flat_buffer& buffer) -+ boost::beast::tcp_stream& /*stream*/, -+ const boost::beast::flat_buffer& /*buffer*/) - { - _clients.erase(client); - -@@ -84,7 +84,7 @@ void wss::detail::http_command_interface_client::async_request( - response_json = nlohmann::json::parse(response.body()); - } - -- catch (const nlohmann::json::exception& ex) -+ catch (const nlohmann::json::exception& /*ex*/) - { - } - -diff --git a/src/detail/peer.cpp b/src/detail/peer.cpp -index 0a24fd2..10d8b01 100644 ---- a/src/detail/peer.cpp -+++ b/src/detail/peer.cpp -@@ -25,7 +25,7 @@ using namespace std::placeholders; - - wss::detail::peer::peer( - boost::asio::ip::tcp::socket&& socket, -- bool is_client, -+ bool /*is_client*/, - bool use_tcp_protocol, - std::size_t rx_buffer_size, - std::size_t tx_buffer_size) -@@ -48,9 +48,10 @@ void wss::detail::peer::run(const void *data, std::size_t size) - if (size > _rx_buffer.size()) - boost::asio::post( - _socket.get_executor(), -- [self = shared_from_this()]() -+ [self_weak = weak_from_this()]() - { -- self->close(boost::asio::error::no_buffer_space); -+ if (auto self = self_weak.lock()) -+ self->close(boost::asio::error::no_buffer_space); - }); - - std::memcpy( -@@ -62,9 +63,10 @@ void wss::detail::peer::run(const void *data, std::size_t size) - - boost::asio::post( - _socket.get_executor(), -- [self = shared_from_this()]() -+ [self_weak = weak_from_this()]() - { -- self->process_buffer(); -+ if (auto self = self_weak.lock()) -+ self->process_buffer(); - }); - } - -@@ -115,20 +117,22 @@ void wss::detail::peer::do_wait_rx() - { - _socket.async_wait( - boost::asio::socket_base::wait_read, -- std::bind( -- &peer::finish_wait_rx, -- shared_from_this(), -- _1)); -+ [self_weak = weak_from_this()](const boost::system::error_code& ec) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_wait_rx(ec); -+ }); - } - - void wss::detail::peer::do_wait_tx() - { - _socket.async_wait( - boost::asio::socket_base::wait_write, -- std::bind( -- &peer::finish_wait_tx, -- shared_from_this(), -- _1)); -+ [self_weak = weak_from_this()](const boost::system::error_code& ec) -+ { -+ if (auto self = self_weak.lock()) -+ self->finish_wait_tx(ec); -+ }); - - _waiting_tx = true; - } -diff --git a/src/server.cpp b/src/server.cpp -index d915db1..a743df7 100644 ---- a/src/server.cpp -+++ b/src/server.cpp -@@ -116,7 +116,7 @@ void wss::server::on_listener_accept( - } - - nlohmann::json wss::server::on_servicer_command_interface_request( -- const std::shared_ptr& servicer, -+ const std::shared_ptr& /*servicer*/, - const std::string& method, - const nlohmann::json& params) - { -@@ -139,12 +139,24 @@ nlohmann::json wss::server::on_servicer_command_interface_request( - } - - void wss::server::on_servicer_websocket_upgrade( -- const std::shared_ptr& servicer, -+ const std::shared_ptr& /*servicer*/, - boost::asio::ip::tcp::socket& socket) - { -+ std::string connection_local_stream_id; -+ try -+ { -+ auto remote_endpoint = socket.remote_endpoint(); -+ connection_local_stream_id = remote_endpoint.address().to_string() -+ + ":" + std::to_string(remote_endpoint.port()); -+ } -+ catch (const std::exception& /*e*/) -+ { -+ return; -+ } - auto connection = std::make_shared( - std::move(socket), -- false); -+ false, -+ connection_local_stream_id); - - if (_command_interface_port) - connection->register_external_command_interface( -@@ -189,7 +201,7 @@ void wss::server::on_servicer_websocket_upgrade( - - void wss::server::on_servicer_closed( - const std::shared_ptr& servicer, -- const boost::system::error_code& ec) -+ const boost::system::error_code& /*ec*/) - { - _sessions.remove_if([&](const client_entry& entry) - { From ebdea31b9c0cae795311e9b548bb6b1a317979d0 Mon Sep 17 00:00:00 2001 From: Viacheslav Kalenikov Date: Wed, 12 Aug 2026 14:33:51 +0200 Subject: [PATCH 3/6] Bump version to 4.0.1 --- module_version | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/module_version b/module_version index fcdb2e1..1454f6e 100644 --- a/module_version +++ b/module_version @@ -1 +1 @@ -4.0.0 +4.0.1 From 9ab5eca737cf34105d258092384c0a08fd7ccbbd Mon Sep 17 00:00:00 2001 From: Jaka Mohorko <96818661+JakaMohorko@users.noreply.github.com> Date: Mon, 17 Aug 2026 13:15:38 +0200 Subject: [PATCH 4/6] [FIX] Missing data -> hidden time signals + missing subscription replies (#16) --- changelog/changelog | 8 + external/ws-streaming/CMakeLists.txt | 4 +- module_version | 2 +- .../tests/CMakeLists.txt | 2 + .../tests/test_hidden_domain_signals.cpp | 440 ++++++++++++++++++ .../websocket_streaming/ws_streaming.h | 28 ++ .../ws_streaming_remote_signal_entry.h | 5 + .../websocket_streaming/src/ws_streaming.cpp | 289 +++++++++++- 8 files changed, 751 insertions(+), 27 deletions(-) create mode 100644 modules/websocket_streaming_client_module/tests/test_hidden_domain_signals.cpp diff --git a/changelog/changelog b/changelog/changelog index 4d6baec..5cb304a 100644 --- a/changelog/changelog +++ b/changelog/changelog @@ -2,10 +2,18 @@ ## 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. + ## Documenation ## Misc +- [#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/external/ws-streaming/CMakeLists.txt b/external/ws-streaming/CMakeLists.txt index 03f83a9..db65596 100644 --- a/external/ws-streaming/CMakeLists.txt +++ b/external/ws-streaming/CMakeLists.txt @@ -2,8 +2,8 @@ set(WS_STREAMING_INSTALL OFF) opendaq_dependency( NAME ws-streaming - REQUIRED_VERSION 3.1.0 + REQUIRED_VERSION 3.1.2 GIT_REPOSITORY https://github.com/openDAQ/ws-streaming - GIT_REF v3.1.0 + GIT_REF v3.1.2 EXPECT_TARGET ws-streaming::ws-streaming ) diff --git a/module_version b/module_version index 1454f6e..4d54dad 100644 --- a/module_version +++ b/module_version @@ -1 +1 @@ -4.0.1 +4.0.2 diff --git a/modules/websocket_streaming_client_module/tests/CMakeLists.txt b/modules/websocket_streaming_client_module/tests/CMakeLists.txt index 29b2796..90d0279 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_hidden_domain_signals.cpp test_app.cpp ) @@ -10,6 +11,7 @@ add_executable(${TEST_APP} ${TEST_SOURCES} target_link_libraries(${TEST_APP} PRIVATE daq::opendaq_test_utils gtest ${OPENDAQ_SDK_TARGET_NAMESPACE}::${MODULE_NAME} + ws-streaming::ws-streaming ) add_test(NAME ${TEST_APP} diff --git a/modules/websocket_streaming_client_module/tests/test_hidden_domain_signals.cpp b/modules/websocket_streaming_client_module/tests/test_hidden_domain_signals.cpp new file mode 100644 index 0000000..fdbad3f --- /dev/null +++ b/modules/websocket_streaming_client_module/tests/test_hidden_domain_signals.cpp @@ -0,0 +1,440 @@ +/* + * 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 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 + +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 + }; + + 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) + { + 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; + } + + // 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 + } + + respond(id, true); + sendValueSignalFamily(); + } + + else if (rpcMethod == "FAKE.subscribe" && signalId == timeSignalId) + { + std::scoped_lock lock(mutex); + ++timeSubscribeRequests; + } + + else if (rpcMethod == "FAKE.unsubscribe") + { + respond(id, true); + if (signalId == valueSignalId) + peer->send_metadata(valueSigno, "unsubscribe", nlohmann::json::object()); + } + } + + void respond(const nlohmann::json& id, bool result) + { + peer->send_metadata(0, "response", { + { "jsonrpc", "2.0" }, + { "id", id }, + { "result", result }, + }); + } + + 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; + std::thread thread; + + std::shared_ptr peer; + boost::signals2::scoped_connection onMetadata; + + std::mutex mutex; + unsigned valueSubscribeRequests = 0; + unsigned timeSubscribeRequests = 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; +} + +} // namespace + +using HiddenDomainSignalsTest = testing::Test; + +TEST_F(HiddenDomainSignalsTest, HiddenDomainSignalIsLinked) +{ + FakeLtPeer peer({}); + auto instance = createClientInstance(); + auto device = connectDevice(instance, peer.port()); + + auto signals = waitForSignals(device, 2, 5s); + 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(HiddenDomainSignalsTest, DomainMetadataArrivingLateIsStillLinked) +{ + FakeLtPeer::Options options; + options.timeMetadataBeforeValue = false; + options.timeMetadataDelay = 300ms; + + FakeLtPeer peer(options); + auto instance = createClientInstance(); + auto device = connectDevice(instance, peer.port()); + + auto signals = waitForSignals(device, 2, 5s); + 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(HiddenDomainSignalsTest, DroppedSubscribeRequestIsRetried) +{ + FakeLtPeer::Options options; + options.dropSubscribeRequests = 1; + + FakeLtPeer peer(options); + auto instance = createClientInstance(); + auto device = connectDevice(instance, peer.port()); + + // the sweep timer retries after 1.5 s + 100 ms; allow generous margin + auto signals = waitForSignals(device, 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(HiddenDomainSignalsTest, ReadvertisedHiddenDomainSignalStartsNoNewFetch) +{ + FakeLtPeer peer({}); + auto instance = createClientInstance(); + auto device = connectDevice(instance, peer.port()); + + auto signals = waitForSignals(device, 2, 5s); + 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(HiddenDomainSignalsTest, SignalPublishesWithoutDomainWhenMetadataNeverArrives) +{ + FakeLtPeer::Options options; + options.timeMetadataBeforeValue = false; + options.withholdTimeMetadata = true; + + FakeLtPeer peer(options); + auto instance = createClientInstance(); + auto device = connectDevice(instance, peer.port()); + + // deferral gives up after two sweep periods (~3 s); allow generous margin + auto signals = waitForSignals(device, 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 8150c0c..8c6ebb5 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 @@ -140,6 +141,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); @@ -157,6 +167,20 @@ 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, releases its initial-fetch subscription and publishes signals deferred on it. */ + void publishSignalEntry(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; @@ -165,6 +189,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..e44d1ec 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 @@ -63,6 +63,11 @@ struct WsStreamingRemoteSignalEntry bool isPublished = false; bool isSubscribed = false; + + bool isHiddenDomain = false; /**< Discovered as a hidden domain signal via "relatedSignals", not advertised as 'available'. */ + unsigned initialFetchAttempts = 0; /**< Subscribe requests sent by the initial metadata fetch; 0 if never fetched or given up on. */ + bool initialFetchActive = false; /**< An initial metadata-fetch subscription is currently active for this signal. */ + unsigned deferredSweeps = 0; /**< Sweep passes seen while deferred on an unpublished domain signal; at 2 the signal is published without the link. */ }; 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 5f2b5a2..b312d0c 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming.cpp @@ -15,12 +15,15 @@ */ #include +#include #include #include #include #include #include +#include +#include #include #include @@ -38,6 +41,11 @@ using namespace std::placeholders; BEGIN_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING +// 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}; + StreamingTypePtr WsStreaming::createType() { return StreamingTypeBuilder() @@ -55,6 +63,7 @@ WsStreaming::WsStreaming( : Streaming(connectionString, context, true) , ioContext{1} , wsClient(ioContext.get_executor()) + , initialFetchTimer(ioContext) { // The ws-streaming library wants a URL like ws://1.2.3.4:7418/foo. // So we simply need to replace the daq.lt:// prefix with ws://. @@ -113,6 +122,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); @@ -144,7 +167,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); @@ -185,9 +208,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; @@ -199,11 +224,121 @@ 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->initialFetchActive) + 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->initialFetchAttempts = 1; + entry->initialFetchActive = true; 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) + 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->deferredSweeps >= 2) + publishSignalEntry(entry); + continue; + } + + if (!entry->initialFetchActive) + continue; + + entry->initialFetchActive = false; + entry->ptr->unsubscribe(); + + if (entry->initialFetchAttempts >= 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->initialFetchAttempts = 0; + } + else + { + ++entry->initialFetchAttempts; + 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) + { + // attempts >= 2 selects signals the sweep unsubscribed (fresh are at 1, given-up at 0); + // an assigned descriptor means metadata already arrived (entry is deferred, not lost) + if (entry->isPublished || entry->initialFetchActive + || entry->descriptor.assigned() || entry->initialFetchAttempts < 2) + continue; + + entry->initialFetchActive = true; + 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) @@ -233,19 +368,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()); } } @@ -262,29 +393,139 @@ void WsStreaming::onRemoteSignalMetadataChanged(std::weak_ptrdescriptor, nullptr); onPacket(entry->ptr->id(), packet); - // changed signal is time signal - if (entry->ptr->id() == entry->ptr->metadata().table_id()) + // propagate to signals that use this signal as their domain + packet = DataDescriptorChangedEventPacket(nullptr, entry->descriptor); + for (const auto& [id, dataSignalEntry] : signals) { - 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); - } + if (dataSignalEntry != entry && + dataSignalEntry->domainEntry == entry && + dataSignalEntry->isPublished) + onPacket(dataSignalEntry->ptr->id(), packet); } } if (entry->descriptor.assigned() && !entry->isPublished) { - LOG_I("Signal {} is now ready, publishing it", entry->ptr->id()); - entry->isPublished = true; + // 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) + { + LOG_D("Deferring signal {} until its domain signal {} is published", + entry->ptr->id(), entry->domainEntry->ptr->id()); + armInitialFetchSweep(); + } + else + { + publishSignalEntry(entry); + } + } +} + +void WsStreaming::publishSignalEntry(const std::shared_ptr& entry) +{ + LOG_I("Signal {} is now ready, publishing it{}", + entry->ptr->id(), entry->isHiddenDomain ? " (hidden domain signal)" : ""); + + 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 + { addToAvailableSignals(entry->ptr->id()); onSignalAvailable( entry->ptr, - entry->domainEntry ? entry->domainEntry->ptr : nullptr, + entry->domainEntry && entry->domainEntry->isPublished ? entry->domainEntry->ptr : nullptr, entry->descriptor); + } + + catch (const std::exception& ex) + { + LOG_E("Failed to register signal {} with openDAQ: {}", entry->ptr->id(), ex.what()); + } + + // release the initial metadata-fetch subscription, if one is active + if (entry->initialFetchActive) + { + entry->initialFetchActive = false; entry->ptr->unsubscribe(); } + entry->initialFetchAttempts = 0; + + // 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 +{ + // an entry with a descriptor but no publication is deferred, waiting for its domain signal + for (const auto& [id, entry] : signals) + if (!entry->isPublished && (entry->initialFetchActive || 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); + domainEntry->isHiddenDomain = true; + + // 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( From 9a8156e8dd83df0c05a36cd9436c724203dd5db0 Mon Sep 17 00:00:00 2001 From: Jaka Mohorko <96818661+JakaMohorko@users.noreply.github.com> Date: Thu, 20 Aug 2026 14:49:04 +0200 Subject: [PATCH 5/6] Hand the initial-fetch subscription over to immediately subscribing applications (#19) --- changelog/changelog | 1 + module_version | 2 +- .../tests/CMakeLists.txt | 2 +- ...nals.cpp => test_device_compatibility.cpp} | 276 ++++++++++++++++-- .../websocket_streaming/ws_streaming.h | 14 +- .../ws_streaming_remote_signal_entry.h | 16 +- .../websocket_streaming/src/ws_streaming.cpp | 75 +++-- 7 files changed, 323 insertions(+), 63 deletions(-) rename modules/websocket_streaming_client_module/tests/{test_hidden_domain_signals.cpp => test_device_compatibility.cpp} (59%) diff --git a/changelog/changelog b/changelog/changelog index 5cb304a..37833b7 100644 --- a/changelog/changelog +++ b/changelog/changelog @@ -7,6 +7,7 @@ - [#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 diff --git a/module_version b/module_version index 4d54dad..c4e41f9 100644 --- a/module_version +++ b/module_version @@ -1 +1 @@ -4.0.2 +4.0.3 diff --git a/modules/websocket_streaming_client_module/tests/CMakeLists.txt b/modules/websocket_streaming_client_module/tests/CMakeLists.txt index 90d0279..c0aee7c 100644 --- a/modules/websocket_streaming_client_module/tests/CMakeLists.txt +++ b/modules/websocket_streaming_client_module/tests/CMakeLists.txt @@ -2,7 +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_hidden_domain_signals.cpp + test_device_compatibility.cpp test_app.cpp ) diff --git a/modules/websocket_streaming_client_module/tests/test_hidden_domain_signals.cpp b/modules/websocket_streaming_client_module/tests/test_device_compatibility.cpp similarity index 59% rename from modules/websocket_streaming_client_module/tests/test_hidden_domain_signals.cpp rename to modules/websocket_streaming_client_module/tests/test_device_compatibility.cpp index fdbad3f..e3c0cf4 100644 --- a/modules/websocket_streaming_client_module/tests/test_hidden_domain_signals.cpp +++ b/modules/websocket_streaming_client_module/tests/test_device_compatibility.cpp @@ -1,7 +1,7 @@ /* * 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 command-interface requests. + * 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 @@ -10,12 +10,16 @@ */ #include +#include #include +#include #include #include #include +#include #include #include +#include #include #include @@ -28,6 +32,7 @@ #include #include +#include #include #include @@ -53,6 +58,8 @@ class FakeLtPeer 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) @@ -60,6 +67,7 @@ class FakeLtPeer , 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) @@ -94,6 +102,25 @@ class FakeLtPeer 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() { @@ -177,8 +204,20 @@ class FakeLtPeer 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) @@ -189,12 +228,40 @@ class FakeLtPeer else if (rpcMethod == "FAKE.unsubscribe") { - respond(id, true); if (signalId == valueSignalId) - peer->send_metadata(valueSigno, "unsubscribe", nlohmann::json::object()); + { + 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", { @@ -204,6 +271,38 @@ class FakeLtPeer }); } + 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 @@ -279,14 +378,23 @@ class FakeLtPeer 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; @@ -335,17 +443,45 @@ SignalPtr findSignalByName(const ListPtr& signals, const std::string& n 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 HiddenDomainSignalsTest = testing::Test; +using DeviceCompatibilityTest = testing::Test; -TEST_F(HiddenDomainSignalsTest, HiddenDomainSignalIsLinked) +TEST_F(DeviceCompatibilityTest, HiddenDomainSignalIsLinked) { FakeLtPeer peer({}); - auto instance = createClientInstance(); - auto device = connectDevice(instance, peer.port()); - - auto signals = waitForSignals(device, 2, 5s); + auto [instance, device, signals] = connectAndWaitForSignals(peer); ASSERT_EQ(signals.getCount(), 2u); auto valueSignal = findSignalByName(signals, "CH1.value"); @@ -357,17 +493,14 @@ TEST_F(HiddenDomainSignalsTest, HiddenDomainSignalIsLinked) ASSERT_EQ(valueSignal.getDomainSignal(), timeSignal); } -TEST_F(HiddenDomainSignalsTest, DomainMetadataArrivingLateIsStillLinked) +TEST_F(DeviceCompatibilityTest, DomainMetadataArrivingLateIsStillLinked) { FakeLtPeer::Options options; options.timeMetadataBeforeValue = false; options.timeMetadataDelay = 300ms; FakeLtPeer peer(options); - auto instance = createClientInstance(); - auto device = connectDevice(instance, peer.port()); - - auto signals = waitForSignals(device, 2, 5s); + auto [instance, device, signals] = connectAndWaitForSignals(peer); ASSERT_EQ(signals.getCount(), 2u); auto valueSignal = findSignalByName(signals, "CH1.value"); @@ -377,17 +510,15 @@ TEST_F(HiddenDomainSignalsTest, DomainMetadataArrivingLateIsStillLinked) ASSERT_TRUE(valueSignal.getDomainSignal().assigned()); } -TEST_F(HiddenDomainSignalsTest, DroppedSubscribeRequestIsRetried) +TEST_F(DeviceCompatibilityTest, DroppedSubscribeRequestIsRetried) { FakeLtPeer::Options options; options.dropSubscribeRequests = 1; FakeLtPeer peer(options); - auto instance = createClientInstance(); - auto device = connectDevice(instance, peer.port()); // the sweep timer retries after 1.5 s + 100 ms; allow generous margin - auto signals = waitForSignals(device, 2, 10s); + auto [instance, device, signals] = connectAndWaitForSignals(peer, 2, 10s); ASSERT_EQ(signals.getCount(), 2u); ASSERT_GE(peer.subscribeRequestCount(), 2u); @@ -396,13 +527,10 @@ TEST_F(HiddenDomainSignalsTest, DroppedSubscribeRequestIsRetried) ASSERT_TRUE(valueSignal.getDomainSignal().assigned()); } -TEST_F(HiddenDomainSignalsTest, ReadvertisedHiddenDomainSignalStartsNoNewFetch) +TEST_F(DeviceCompatibilityTest, ReadvertisedHiddenDomainSignalStartsNoNewFetch) { FakeLtPeer peer({}); - auto instance = createClientInstance(); - auto device = connectDevice(instance, peer.port()); - - auto signals = waitForSignals(device, 2, 5s); + auto [instance, device, signals] = connectAndWaitForSignals(peer); ASSERT_EQ(signals.getCount(), 2u); // the device now advertises the already-published hidden time signal @@ -420,18 +548,112 @@ TEST_F(HiddenDomainSignalsTest, ReadvertisedHiddenDomainSignalStartsNoNewFetch) ASSERT_EQ(peer.timeSubscribeRequestCount(), 0u); } -TEST_F(HiddenDomainSignalsTest, SignalPublishesWithoutDomainWhenMetadataNeverArrives) +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); - auto instance = createClientInstance(); - auto device = connectDevice(instance, peer.port()); // deferral gives up after two sweep periods (~3 s); allow generous margin - auto signals = waitForSignals(device, 1, 10s); + auto [instance, device, signals] = connectAndWaitForSignals(peer, 1, 10s); ASSERT_EQ(signals.getCount(), 1u); auto valueSignal = findSignalByName(signals, "CH1.value"); 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 8c6ebb5..09edcb4 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h @@ -48,12 +48,12 @@ 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. * * 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 @@ -171,7 +171,7 @@ class WsStreaming : public Streaming std::shared_ptr resolveDomainEntry( const std::shared_ptr& entry); - /*! @brief Registers a signal with openDAQ, releases its initial-fetch subscription and publishes signals deferred on it. */ + /*! @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 Checks whether any signal is still awaiting initial metadata or deferred on an unpublished domain signal. */ 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 e44d1ec..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 @@ -64,10 +73,9 @@ struct WsStreamingRemoteSignalEntry bool isPublished = false; bool isSubscribed = false; - bool isHiddenDomain = false; /**< Discovered as a hidden domain signal via "relatedSignals", not advertised as 'available'. */ - unsigned initialFetchAttempts = 0; /**< Subscribe requests sent by the initial metadata fetch; 0 if never fetched or given up on. */ - bool initialFetchActive = false; /**< An initial metadata-fetch subscription is currently active for this signal. */ - unsigned deferredSweeps = 0; /**< Sweep passes seen while deferred on an unpublished domain signal; at 2 the signal is published without the link. */ + 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 b312d0c..4230cca 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming.cpp @@ -41,6 +41,8 @@ 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}; @@ -156,6 +158,16 @@ void WsStreaming::subscribeRemoteSignal(const std::string& 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); + return; + } + LOG_I("Found signal, subscribing"); signalIt->second->ptr->subscribe(); signalIt->second->isSubscribed = true; @@ -237,13 +249,13 @@ void WsStreaming::onRemoteSignalAvailable(wss::remote_signal_ptr 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->initialFetchActive) + 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->initialFetchAttempts = 1; - entry->initialFetchActive = true; + entry->fetchState = FetchState::Fetching; + entry->fetchAttempts = 1; signal->subscribe(); armInitialFetchSweep(); @@ -269,31 +281,40 @@ void WsStreaming::onInitialFetchSweep(const boost::system::error_code& ec) 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->deferredSweeps >= 2) + if (++entry->sweeps >= 2) publishSignalEntry(entry); continue; } - if (!entry->initialFetchActive) + if (entry->fetchState != FetchState::Fetching) continue; - entry->initialFetchActive = false; entry->ptr->unsubscribe(); - if (entry->initialFetchAttempts >= INITIAL_FETCH_MAX_ATTEMPTS) + 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->initialFetchAttempts = 0; + entry->fetchState = FetchState::None; + entry->fetchAttempts = 0; } else { - ++entry->initialFetchAttempts; + entry->fetchState = FetchState::AwaitingRetry; + ++entry->fetchAttempts; retrying = true; } } @@ -325,13 +346,11 @@ void WsStreaming::onInitialFetchResubscribe(const boost::system::error_code& ec) for (const auto& [id, entry] : signals) { - // attempts >= 2 selects signals the sweep unsubscribed (fresh are at 1, given-up at 0); // an assigned descriptor means metadata already arrived (entry is deferred, not lost) - if (entry->isPublished || entry->initialFetchActive - || entry->descriptor.assigned() || entry->initialFetchAttempts < 2) + if (entry->fetchState != FetchState::AwaitingRetry || entry->descriptor.assigned()) continue; - entry->initialFetchActive = true; + entry->fetchState = FetchState::Fetching; entry->ptr->subscribe(); pending = true; } @@ -422,8 +441,7 @@ void WsStreaming::onRemoteSignalMetadataChanged(std::weak_ptr& entry) { - LOG_I("Signal {} is now ready, publishing it{}", - entry->ptr->id(), entry->isHiddenDomain ? " (hidden domain signal)" : ""); + LOG_I("Signal {} is now ready, publishing it", entry->ptr->id()); if (entry->domainEntry && !entry->domainEntry->isPublished) { @@ -448,13 +466,17 @@ void WsStreaming::publishSignalEntry(const std::shared_ptrptr->id(), ex.what()); } - // release the initial metadata-fetch subscription, if one is active - if (entry->initialFetchActive) + // keep the fetch subscription so an immediate application subscribe can take it over + if (entry->fetchState == FetchState::Fetching) { - entry->initialFetchActive = false; - entry->ptr->unsubscribe(); + entry->fetchState = FetchState::Held; + armInitialFetchSweep(); } - entry->initialFetchAttempts = 0; + 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) @@ -465,9 +487,10 @@ void WsStreaming::publishSignalEntry(const std::shared_ptrisPublished && (entry->initialFetchActive || entry->descriptor.assigned())) + if (entry->fetchState == FetchState::Held + || (!entry->isPublished && (entry->fetchState == FetchState::Fetching || entry->descriptor.assigned()))) return true; return false; @@ -518,7 +541,6 @@ std::shared_ptr WsStreaming::resolveDomainEntry( LOG_D("Discovered hidden domain signal {} of signal {}", domainSignalId, entry->ptr->id()); auto domainEntry = createSignalEntry(domainSignal); - domainEntry->isHiddenDomain = true; // 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) @@ -597,6 +619,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; From feabcf7a2483c1f86ed79a2ee3d7f8168269fd21 Mon Sep 17 00:00:00 2001 From: Jaka Mohorko <96818661+JakaMohorko@users.noreply.github.com> Date: Thu, 20 Aug 2026 22:06:47 +0200 Subject: [PATCH 6/6] Replay the cached descriptor when a subscribe takes over the initial-fetch subscription (#21) --- .../websocket_streaming/ws_streaming.h | 6 +++- .../websocket_streaming/src/ws_streaming.cpp | 34 +++++++++++-------- 2 files changed, 25 insertions(+), 15 deletions(-) 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 af0bd62..d8df97a 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h @@ -53,7 +53,8 @@ BEGIN_NAMESPACE_OPENDAQ_WEBSOCKET_STREAMING * 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. + * 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 @@ -186,6 +187,9 @@ class WsStreaming : public Streaming /*! @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; diff --git a/shared/libraries/websocket_streaming/src/ws_streaming.cpp b/shared/libraries/websocket_streaming/src/ws_streaming.cpp index 9ce66e2..a5c0c73 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming.cpp @@ -369,6 +369,9 @@ void WsStreaming::subscribeRemoteSignal(const std::string& signalId) signalIt->second->fetchState = FetchState::None; signalIt->second->isSubscribed = true; triggerSubscribeAck(signalId, true); + + if (signalIt->second->descriptor.assigned()) + emitDescriptorChangedEvents(signalIt->second); return; } @@ -612,20 +615,7 @@ void WsStreaming::onRemoteSignalMetadataChanged(std::weak_ptrdescriptor.assigned() && entry->isPublished) - { - 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); - } - } + emitDescriptorChangedEvents(entry); if (entry->descriptor.assigned() && !entry->isPublished) { // defer until the domain publishes: publishSignalEntry() then publishes this signal too; @@ -643,6 +633,22 @@ void WsStreaming::onRemoteSignalMetadataChanged(std::weak_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());