From a1f1ad56795b558060491dd5451c25ada9d9f950 Mon Sep 17 00:00:00 2001 From: Jaka Mohorko Date: Thu, 20 Aug 2026 21:31:55 +0200 Subject: [PATCH] Replay the cached descriptor when a subscribe takes over the initial-fetch subscription --- .../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 09edcb4..430f613 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 @@ -174,6 +175,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 4230cca..2c5b97e 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming.cpp @@ -165,6 +165,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; } @@ -408,20 +411,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; @@ -439,6 +429,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());