From 3808a66c1eb786cb289cb44551dfab13c635f5c8 Mon Sep 17 00:00:00 2001 From: trigaux Date: Wed, 16 Sep 2026 17:00:31 -0400 Subject: [PATCH 1/2] UPdate to use the new streambuffer flow. --- dependencies/libquicr | 2 +- src/client_manager.cc | 183 +++++++++++++++++-- src/client_manager.h | 36 ++++ src/fetch_handler.cc | 64 ++----- src/fetch_handler.h | 7 +- src/peering/peer_session.cc | 71 ++++---- src/peering/peer_session.h | 9 +- src/publish_namespace_handler.cc | 8 - src/subscribe_handler.cc | 297 +++---------------------------- src/subscribe_handler.h | 18 +- 10 files changed, 287 insertions(+), 408 deletions(-) diff --git a/dependencies/libquicr b/dependencies/libquicr index 0b0841f..c88c58c 160000 --- a/dependencies/libquicr +++ b/dependencies/libquicr @@ -1 +1 @@ -Subproject commit 0b0841f4912d88c8227806daaccef115fb4b1aa2 +Subproject commit c88c58c73ce853807858002f79e8ab9feeda1e73 diff --git a/src/client_manager.cc b/src/client_manager.cc index 37c158b..1bf4674 100644 --- a/src/client_manager.cc +++ b/src/client_manager.cc @@ -14,7 +14,10 @@ #include "subscribe_handler.h" #include "transport_helpers.h" +#include "peering/messages/data_header.h" + #include +#include namespace laps { ClientManager::ClientManager(State& state, @@ -1705,32 +1708,186 @@ namespace laps { std::shared_ptr> data) { const auto it = state_.pub_subscribes.find({ track_full_name_hash, 0 }); - if (it == state_.pub_subscribes.end()) { + if (it == state_.pub_subscribes.end() || !it->second) { return; } - if (stream_id.has_value()) { - if (is_new_stream) { - quicr::InitialStreamData initial_buffer; - initial_buffer.buffer.Push(*data); - initial_buffer.source_buffers.push_back(std::move(data)); - it->second->StreamDataRecv(*stream_id, std::move(initial_buffer)); - } else { - it->second->StreamDataRecv(*stream_id, std::move(data)); - } - } else { + if (!stream_id.has_value()) { it->second->DgramDataRecv(std::move(data)); + return; } + + if (is_new_stream) { + peer_rx_streams_.erase({ track_full_name_hash, *stream_id }); + } + + auto& stream = peer_rx_streams_[{ track_full_name_hash, *stream_id }]; + if (!stream.header_initialized) { + stream.buffer.InitAny(); + stream.header_initialized = true; + } + + stream.buffer.Push(std::span(*data)); + TryParsePeerStream(stream, *it->second); } void ClientManager::PeerStreamClosed(std::uint64_t track_full_name_hash, uint64_t stream_id, bool reset) { const auto it = state_.pub_subscribes.find({ track_full_name_hash, 0 }); - if (it == state_.pub_subscribes.end()) { + const auto stream_it = peer_rx_streams_.find({ track_full_name_hash, stream_id }); + if (stream_it == peer_rx_streams_.end()) { return; } - it->second->StreamClosed(stream_id, reset); + if (it != state_.pub_subscribes.end() && it->second) { + it->second->SubgroupEnded(stream_it->second.group_id, stream_it->second.subgroup_id, reset); + } + + peer_rx_streams_.erase(stream_it); + } + + void ClientManager::TryParsePeerStream(PeerRxStreamState& stream, SubscribeTrackHandler& handler) + { + auto& s_hdr = stream.buffer.GetAny(); + if (!(stream.buffer >> s_hdr)) { + return; + } + + if (s_hdr.priority.has_value()) { + handler.SetPriority(*s_hdr.priority); + } + + while (!stream.buffer.Empty()) { + if (!stream.buffer.AnyHasValueB()) { + stream.buffer.InitAnyB(); + } + + auto& obj = stream.buffer.GetAnyB(); + obj.properties.emplace(*s_hdr.properties); + if (!(stream.buffer >> obj)) { + return; + } + + handler.subscribe_track_metrics_.objects_received++; + + const bool same_subgroup = stream.next_object_id.has_value() && stream.group_id == s_hdr.group_id && + stream.subgroup_id == s_hdr.subgroup_id.value_or(stream.subgroup_id); + stream.next_object_id = same_subgroup ? *stream.next_object_id + obj.object_delta : obj.object_delta; + + if (!s_hdr.subgroup_id.has_value()) { + if (obj.properties->subgroup_id_mode != quicr::messages::SubgroupIdType::kSetFromFirstObject) { + throw quicr::messages::ProtocolViolationException("Subgoup ID mismatch"); + } + s_hdr.subgroup_id = stream.next_object_id; + } + + stream.group_id = s_hdr.group_id; + stream.subgroup_id = *s_hdr.subgroup_id; + + std::optional stream_mode; + if (!same_subgroup && s_hdr.properties.has_value()) { + stream_mode.emplace(*s_hdr.properties); + } + + handler.ObjectReceived({ s_hdr.group_id, + *stream.next_object_id, + *s_hdr.subgroup_id, + obj.payload.size(), + obj.object_status, + s_hdr.priority, + std::nullopt, + quicr::TrackMode::kStream, + obj.extensions, + obj.immutable_extensions }, + obj.payload, + stream_mode); + + *stream.next_object_id += 1; + stream.buffer.ResetAnyB(); + } + } + + void ClientManager::SendObjectToPeers(uint64_t track_alias, + uint8_t priority, + uint32_t ttl, + bool is_datagram, + const quicr::ObjectHeaders& object_headers, + quicr::BytesSpan data, + std::optional stream_mode) + { + auto bytes = std::make_shared>(); + peering::DataType d_type = peering::DataType::kDatagram; + + if (is_datagram) { + quicr::messages::ObjectDatagram msg; + msg.group_id = object_headers.group_id; + msg.object_id = object_headers.object_id; + msg.priority = object_headers.priority.value_or(priority); + msg.track_alias = track_alias; + msg.extensions = object_headers.extensions; + msg.immutable_extensions = object_headers.immutable_extensions; + msg.payload.assign(data.begin(), data.end()); + *bytes << msg; + } else { + auto& tx = peer_tx_subgroups_[{ track_alias, { object_headers.group_id, object_headers.subgroup_id } }]; + const bool is_new_stream = !tx.last_object_id.has_value(); + d_type = is_new_stream ? peering::DataType::kNewStream : peering::DataType::kExistingStream; + + if (stream_mode.has_value()) { + tx.properties.emplace(*stream_mode); + } + + const auto properties = tx.properties.value_or( + stream_mode.value_or(quicr::messages::StreamHeaderProperties(false, + quicr::messages::SubgroupIdType::kExplicit, + false, + false, + true))); + + uint64_t object_id_delta = object_headers.object_id; + if (tx.last_object_id.has_value()) { + object_id_delta = object_headers.object_id >= *tx.last_object_id + ? object_headers.object_id - *tx.last_object_id + : object_headers.object_id; + if (object_id_delta) { + object_id_delta--; + } + } + + if (is_new_stream) { + quicr::messages::StreamHeaderSubGroup subgroup_hdr; + subgroup_hdr.properties.emplace(properties); + subgroup_hdr.group_id = object_headers.group_id; + if (properties.subgroup_id_mode == quicr::messages::SubgroupIdType::kExplicit) { + subgroup_hdr.subgroup_id = object_headers.subgroup_id; + } + if (!properties.default_priority) { + subgroup_hdr.priority = object_headers.priority.value_or(priority); + } + subgroup_hdr.track_alias = track_alias; + *bytes << subgroup_hdr; + } + + quicr::messages::StreamSubGroupObject object; + object.object_delta = object_id_delta; + object.object_status = object_headers.status; + object.properties.emplace(properties); + object.extensions = object_headers.extensions; + object.immutable_extensions = object_headers.immutable_extensions; + object.payload.assign(data.begin(), data.end()); + *bytes << object; + + tx.last_object_id = object_headers.object_id; + } + + peer_manager_.ClientDataRecv( + track_alias, priority, ttl, d_type, object_headers.group_id, object_headers.subgroup_id, std::move(bytes)); + } + + void ClientManager::PeerSubgroupEnded(uint64_t track_alias, uint64_t group_id, uint64_t subgroup_id, bool reset) + { + peer_tx_subgroups_.erase({ track_alias, { group_id, subgroup_id } }); + peer_manager_.EndSubgroup(track_alias, group_id, subgroup_id, reset); } void ClientManager::MetricsSampled(const std::shared_ptr& session, diff --git a/src/client_manager.h b/src/client_manager.h index 5332c1c..98bdeaa 100644 --- a/src/client_manager.h +++ b/src/client_manager.h @@ -7,6 +7,8 @@ #include "track_ranking.h" #include #include +#include +#include #include #include #include @@ -16,6 +18,7 @@ #include #include #include +#include #include namespace laps { @@ -157,6 +160,32 @@ namespace laps { friend class peering::PeerManager; private: + struct PeerRxStreamState + { + quicr::StreamBuffer buffer; + uint64_t group_id{ 0 }; + uint64_t subgroup_id{ 0 }; + std::optional next_object_id; + bool header_initialized{ false }; + }; + + struct PeerTxSubgroupState + { + std::optional last_object_id; + std::optional properties; + }; + + void SendObjectToPeers(uint64_t track_alias, + uint8_t priority, + uint32_t ttl, + bool is_datagram, + const quicr::ObjectHeaders& object_headers, + quicr::BytesSpan data, + std::optional stream_mode); + + void PeerSubgroupEnded(uint64_t track_alias, uint64_t group_id, uint64_t subgroup_id, bool reset); + + void TryParsePeerStream(PeerRxStreamState& stream, SubscribeTrackHandler& handler); /** * @brief Connection handle of the session that received a request, or zero if it is already gone */ @@ -312,6 +341,13 @@ namespace laps { // Key is track namespace hash std::unordered_map> track_rankings_; + /// Inbound peer subgroup parse state, keyed by track full name hash and peer stream id + std::map, PeerRxStreamState> peer_rx_streams_; + + /// Outbound peer subgroup serialization state, keyed by track alias, group id, and subgroup id + std::map>, PeerTxSubgroupState> + peer_tx_subgroups_; + friend class SubscribeTrackHandler; friend class PublishTrackHandler; friend class FetchTrackHandler; diff --git a/src/fetch_handler.cc b/src/fetch_handler.cc index 815eea3..5337fd8 100644 --- a/src/fetch_handler.cc +++ b/src/fetch_handler.cc @@ -4,7 +4,6 @@ #include "fetch_handler.h" #include "config.h" #include -#include namespace laps { FetchTrackHandler::FetchTrackHandler(const std::shared_ptr publish_fetch_handler, @@ -18,64 +17,21 @@ namespace laps { { } - void FetchTrackHandler::StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) + void FetchTrackHandler::ObjectReceived(const quicr::ObjectHeaders& object_headers, + quicr::BytesSpan data, + std::optional stream_mode) { - subscribe_track_metrics_.bytes_received += initial_buffer.buffer.Size(); - - auto [stream_it, inserted] = - streams_.try_emplace(stream_id, StreamContext{ .buffer = std::move(initial_buffer.buffer) }); - if (!inserted) { - SPDLOG_ERROR("StreamDataRecv got new stream for existing Stream ID {}", stream_id); - return; - } - - stream_it->second.buffer.InitAny(); - TryForwardInitialStreamData(stream_id, stream_it->second); - } - - void FetchTrackHandler::StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) - { - subscribe_track_metrics_.bytes_received += data->size(); - - if (initial_stream_data_forwarded_) { - publish_fetch_handler_->ForwardPublishedData(false, 0, 0, std::move(data)); - return; - } - - const auto stream_it = streams_.find(stream_id); - if (stream_it == streams_.end()) { - SPDLOG_ERROR("StreamDataRecv had no stream for expected Stream ID {}", stream_id); - return; - } - - stream_it->second.buffer.Push(*data); - TryForwardInitialStreamData(stream_id, stream_it->second); - } - - void FetchTrackHandler::TryForwardInitialStreamData(uint64_t stream_id, StreamContext& stream) - { - auto& f_hdr = stream.buffer.GetAny(); - if (!(stream.buffer >> f_hdr)) { + if (!initial_stream_data_forwarded_) { + publish_fetch_handler_->PublishObject(object_headers, data, stream_mode); + initial_stream_data_forwarded_ = true; return; } - SPDLOG_DEBUG("Fetch header added in rid: {} out rid: {} remaining data size: {}", - f_hdr.request_id, - *publish_fetch_handler_->GetRequestId(), - stream.buffer.Size()); - - f_hdr.request_id = *publish_fetch_handler_->GetRequestId(); - auto bytes = std::make_shared(); - *bytes << f_hdr; - - const auto remaining_data = stream.buffer.Data(); - if (!remaining_data.empty()) { - bytes->insert(bytes->end(), remaining_data.begin(), remaining_data.end()); + auto bytes = std::make_shared>(data.begin(), data.end()); + if (publish_fetch_handler_->ForwardPublishedData(false, 0, 0, std::move(bytes)) != + quicr::PublishTrackHandler::PublishObjectStatus::kOk) { + publish_fetch_handler_->PublishObject(object_headers, data, stream_mode); } - - publish_fetch_handler_->ForwardPublishedData(true, 0, 0, std::move(bytes)); - initial_stream_data_forwarded_ = true; - streams_.erase(stream_id); } void FetchTrackHandler::StatusChanged(Status status) diff --git a/src/fetch_handler.h b/src/fetch_handler.h index 1955136..9eaa906 100644 --- a/src/fetch_handler.h +++ b/src/fetch_handler.h @@ -32,12 +32,11 @@ namespace laps { } void StatusChanged(Status status) override; - void StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) override; - void StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) override; + void ObjectReceived(const quicr::ObjectHeaders& object_headers, + quicr::BytesSpan data, + std::optional stream_mode = std::nullopt) override; private: - void TryForwardInitialStreamData(uint64_t stream_id, StreamContext& stream); - bool initial_stream_data_forwarded_{ false }; std::shared_ptr publish_fetch_handler_; }; diff --git a/src/peering/peer_session.cc b/src/peering/peer_session.cc index 47d54a3..c2d86f7 100644 --- a/src/peering/peer_session.cc +++ b/src/peering/peer_session.cc @@ -8,6 +8,10 @@ #include "peering/messages/node_info.h" #include "peering/messages/subscribe_info.h" +// Stream is internal to libquicr. PeerSession implements Connection::Delegate and +// must drain rx_data directly because peering still carries raw subgroup bytes. +#include "../../dependencies/libquicr/src/stream.h" + #include #include @@ -554,47 +558,51 @@ namespace laps::peering { return true; } - void PeerSession::OnRecvStream(std::uint64_t stream_id, - const std::shared_ptr& rx_ctx, - const std::shared_ptr& stream, - const bool is_bidir) + bool PeerSession::OnRecvStream(const std::shared_ptr& stream) { - if (!IsUsable() || rx_ctx == nullptr) { - return; + if (!IsUsable() || stream == nullptr) { + return false; } + const auto stream_id = stream->GetStreamId(); + const bool is_bidir = stream->IsBidirectional(); + /* * Control messages arrive on the one bidirectional stream, which the inbound side of the session * only learns about here. Both sides answer on it, so it is adopted either way. */ - if (is_bidir && stream != nullptr) { + if (is_bidir) { control_stream_ = stream; control_stream_id_ = stream_id; } - for (int i = 0; i < kReadLoopMaxPerStream; i++) { - if (rx_ctx->data_queue.Empty()) { - break; + std::shared_ptr> data; + { + std::lock_guard _(stream->rx_mutex); + if (stream->rx_data.Empty()) { + return false; } - auto data_opt = rx_ctx->data_queue.Pop(); - if (not data_opt.has_value()) { - break; + const auto view = stream->rx_data.Data(); + if (!is_bidir) { + const auto header_it = rx_stream_headers_.find(stream_id); + if (header_it == rx_stream_headers_.end() && (view.empty() || view.size() < view.front())) { + return false; + } } - const auto& data = data_opt.value(); - - // Get common header - if (is_bidir) { // control - controL_msg_buffer_.insert(controL_msg_buffer_.end(), data->begin(), data->end()); - - ProcessControlMessage(); + data = std::make_shared>(view.begin(), view.end()); + stream->rx_data.Pop(view.size()); + } - } else if (!ProcessReceivedData(stream_id, std::move(data))) { - i = 59; - continue; // Try once more - } + if (is_bidir) { + controL_msg_buffer_.insert(controL_msg_buffer_.end(), data->begin(), data->end()); + ProcessControlMessage(); + } else { + ProcessReceivedData(stream_id, std::move(data)); } + + return stream->RxDataSize() > 0; } void PeerSession::OnRecvDgram() @@ -618,9 +626,9 @@ namespace laps::peering { SPDLOG_LOGGER_TRACE(LOGGER, "Received dgram sns_id: {} track_full_name: {} data size: {}", - data_object.sns_id, - data_object.track_full_name_hash, - data_object.data.size()); + data_header.sns_id, + data_header.track_full_name_hash, + data->size()); } } @@ -630,10 +638,13 @@ namespace laps::peering { metrics_.srtt_us = quic_connection_metrics.srtt_us.avg; } - void PeerSession::OnStreamClosed(std::uint64_t stream_id, - [[maybe_unused]] std::shared_ptr rx_context, - quicr::StreamClosedFlag flag) + void PeerSession::OnStreamClosed(const std::shared_ptr& stream, quicr::StreamClosedFlag flag) { + if (stream == nullptr) { + return; + } + + const auto stream_id = stream->GetStreamId(); const auto conn_id = GetSessionId(); if (control_stream_ != nullptr && control_stream_id_.has_value() && *control_stream_id_ == stream_id) { diff --git a/src/peering/peer_session.h b/src/peering/peer_session.h index 8071756..826fd2f 100644 --- a/src/peering/peer_session.h +++ b/src/peering/peer_session.h @@ -176,10 +176,7 @@ namespace laps::peering { */ void OnConnectionStatus(quicr::Connection::Status status) override; - void OnRecvStream(std::uint64_t stream_id, - const std::shared_ptr& rx_ctx, - const std::shared_ptr& stream, - bool is_bidir = false) override; + bool OnRecvStream(const std::shared_ptr& stream) override; void OnRecvDgram() override; @@ -193,9 +190,7 @@ namespace laps::peering { { } - void OnStreamClosed(std::uint64_t stream_id, - std::shared_ptr rx_context, - quicr::StreamClosedFlag flag) override; + void OnStreamClosed(const std::shared_ptr& stream, quicr::StreamClosedFlag flag) override; // --------------------------------------- diff --git a/src/publish_namespace_handler.cc b/src/publish_namespace_handler.cc index 3974733..03bc7fd 100644 --- a/src/publish_namespace_handler.cc +++ b/src/publish_namespace_handler.cc @@ -39,18 +39,10 @@ laps::PublishNamespaceHandler::PublishObject(std::uint64_t track_full_name_hash, if (const auto pub_it = handlers_.find(track_full_name_hash); pub_it != handlers_.end()) { - auto handler = dynamic_pointer_cast(pub_it->second); - - // Use pipeline forwarding now that first object has been sent - if (handler->SentFirstObject(object_headers.group_id, object_headers.subgroup_id)) { - return quicr::PublishTrackHandler::PublishObjectStatus::kOk; - } - if (object_headers.track_mode.has_value()) { pub_it->second->SetDefaultTrackMode(object_headers.track_mode.value()); } - // Not using pipeline forwarding, publish complete object return pub_it->second->PublishObject(object_headers, data, stream_mode); } diff --git a/src/subscribe_handler.cc b/src/subscribe_handler.cc index d466858..6546156 100644 --- a/src/subscribe_handler.cc +++ b/src/subscribe_handler.cc @@ -174,9 +174,10 @@ namespace laps { is_datagram_ = *object_headers.track_mode == quicr::TrackMode::kDatagram; } - auto self_connection_handle = GetConnectionId(); + if (!GetTrackAlias().has_value()) { + return; + } - // Update tracked properties UpdateTrackedProperties(object_headers.extensions, object_headers.immutable_extensions); if (pending_new_group_request_id_.has_value() && @@ -186,8 +187,7 @@ namespace laps { CacheObject object{ object_headers, { data.begin(), data.end() } }; - // Cache Object - if (!server_.config_.disable_cache) { + if (!server_.config_.disable_cache && GetTrackAlias().has_value()) { if (server_.cache_.count(GetTrackAlias().value()) == 0) { server_.cache_.insert( std::make_pair(GetTrackAlias().value(), @@ -205,272 +205,34 @@ namespace laps { } try { - // Fanout object to subscribe namespaces - for (const auto [_, conn_subs] : sub_namespaces_) { + for (const auto& [_, conn_subs] : sub_namespaces_) { for (const auto& [_, handler] : conn_subs) { handler->PublishObject(GetTrackAlias().value(), object_headers, data, stream_mode); } } - // Fanout object to subscribers for (auto& [conn_handle, pub_handler] : subscribers_) { - if (conn_handle == 0 || (!is_datagram_ && pub_handler->SentFirstObject(object_headers.group_id, - object_headers.subgroup_id))) { + if (conn_handle == 0) { continue; } pub_handler->SetDefaultTrackMode(is_datagram_ ? quicr::TrackMode::kDatagram : quicr::TrackMode::kStream); - if (object.headers.group_id >= pub_handler->start_location_.group && - object.headers.object_id >= pub_handler->start_location_.object) { + if (object_headers.group_id >= pub_handler->start_location_.group && + object_headers.object_id >= pub_handler->start_location_.object) { pub_handler->PublishObject(object_headers, data, stream_mode); } } - } catch (const std::exception& e) { - SPDLOG_ERROR("Caught exception trying to publish. (error={})", e.what()); - } - } - - void SubscribeTrackHandler::StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) - { - is_datagram_ = false; - - auto [stream_it, inserted] = - streams_.try_emplace(stream_id, StreamContext{ .buffer = std::move(initial_buffer.buffer) }); - if (!inserted) { - SPDLOG_ERROR("StreamDataRecv got new stream for existing Stream ID {}", stream_id); - return; - } - - stream_it->second.buffer.InitAny(); - pending_source_buffers_.emplace(stream_id, std::move(initial_buffer.source_buffers)); - TryProcessStreamData(stream_id, stream_it->second); - } - - void SubscribeTrackHandler::StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) - { - is_datagram_ = false; - - const auto stream_it = streams_.find(stream_id); - if (stream_it == streams_.end()) { - SPDLOG_ERROR("StreamDataRecv had no stream for expected Stream ID {}", stream_id); - return; - } - - stream_it->second.buffer.Push(*data); - - const auto pending_it = pending_source_buffers_.find(stream_id); - if (pending_it != pending_source_buffers_.end()) { - pending_it->second.push_back(std::move(data)); - } else { - ForwardReceivedData( - false, stream_it->second.current_group_id, stream_it->second.current_subgroup_id, std::move(data)); - } - - TryProcessStreamData(stream_id, stream_it->second); - } - - void SubscribeTrackHandler::TryProcessStreamData(uint64_t stream_id, StreamContext& stream) - { - auto& s_hdr = stream.buffer.GetAny(); - const auto pending_it = pending_source_buffers_.find(stream_id); - if (pending_it != pending_source_buffers_.end()) { - if (not(stream.buffer >> s_hdr)) { - SPDLOG_DEBUG("Not enough data to process new stream headers yet"); - return; - } - - if (s_hdr.priority.has_value()) { - SetPriority(*s_hdr.priority); - } - - // Adapt publisher track alias to normalized track alias used by subscribers. - if (GetReceivedTrackAlias().value_or(0) != GetTrackAlias().value() || pending_it->second.empty()) { - s_hdr.track_alias = GetTrackAlias().value(); - - auto updated_data = std::make_shared>(); - *updated_data << s_hdr; - - const auto remaining_data = stream.buffer.Data(); - updated_data->insert(updated_data->end(), remaining_data.begin(), remaining_data.end()); - - ForwardReceivedData(true, s_hdr.group_id, s_hdr.subgroup_id.value_or(0), std::move(updated_data)); - } else { - bool is_start = true; - for (auto& source_buffer : pending_it->second) { - ForwardReceivedData( - is_start, s_hdr.group_id, s_hdr.subgroup_id.value_or(0), std::move(source_buffer)); - is_start = false; - } - } - - stream.current_group_id = s_hdr.group_id; - stream.current_subgroup_id = s_hdr.subgroup_id.value_or(0); - pending_source_buffers_.erase(pending_it); - } - - while (not stream.buffer.Empty()) { - if (not stream.buffer.AnyHasValueB()) { - stream.buffer.InitAnyB(); - } - - auto& obj = stream.buffer.GetAnyB(); - obj.properties.emplace(*s_hdr.properties); - if (not(stream.buffer >> obj)) { - return; - } - - subscribe_track_metrics_.objects_received++; - const bool is_first_object = not stream.next_object_id.has_value(); - - if (stream.next_object_id.has_value()) { - if (stream.current_group_id != s_hdr.group_id || stream.current_subgroup_id != s_hdr.subgroup_id) { - stream.next_object_id = obj.object_delta; - } else { - *stream.next_object_id += obj.object_delta; - } - } else { - stream.next_object_id = obj.object_delta; - } - - if (!s_hdr.subgroup_id.has_value()) { - if (obj.properties->subgroup_id_mode != quicr::messages::SubgroupIdType::kSetFromFirstObject) { - throw quicr::messages::ProtocolViolationException("Subgoup ID mismatch"); - } - // Set the subgroup ID from the first object ID. - s_hdr.subgroup_id = stream.next_object_id; - } - - stream.current_group_id = s_hdr.group_id; - stream.current_subgroup_id = s_hdr.subgroup_id.value(); - - ObjectReceived({ s_hdr.group_id, - stream.next_object_id.value(), - s_hdr.subgroup_id.value(), - obj.payload.size(), - obj.object_status, - s_hdr.priority, - std::nullopt, - quicr::TrackMode::kStream, - obj.extensions, - obj.immutable_extensions }, - obj.payload, - s_hdr.properties); - - *stream.next_object_id += 1; - stream.buffer.ResetAnyB(); - - if (is_first_object) { - const auto remaining_data = stream.buffer.Data(); - if (!remaining_data.empty()) { - SPDLOG_DEBUG("Bytes remaining being forwarded: {}", stream.buffer.Size()); - ForwardReceivedData( - false, - s_hdr.group_id, - s_hdr.subgroup_id.value_or(0), - std::make_shared>(remaining_data.begin(), remaining_data.end()), - false); - } - } - } - } - - void SubscribeTrackHandler::DgramDataRecv(std::shared_ptr> data) - { - is_datagram_ = true; - - dgram_buffer_.Clear(); - dgram_buffer_.Push(*data); - - quicr::messages::ObjectDatagram msg; - if (dgram_buffer_ >> msg) { - if (subscribers_.contains(0)) { // Is peering subscribed - server_.peer_manager_.ClientDataRecv( - msg.track_alias, - GetPriority(), - GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(), - peering::DataType::kDatagram, - msg.group_id, - 0, - data); - } - - subscribe_track_metrics_.objects_received++; - subscribe_track_metrics_.bytes_received += msg.payload.size(); - ObjectReceived( - { - msg.group_id, - msg.object_id, - 0, // datagrams don't have subgroups - msg.payload.size(), - quicr::ObjectStatus::kAvailable, - msg.priority, - std::nullopt, - quicr::TrackMode::kDatagram, - msg.extensions, - }, - msg.payload); - } - } - - void SubscribeTrackHandler::ForwardReceivedData(bool is_new_stream, - uint64_t group_id, - uint64_t subgroup_id, - std::shared_ptr> data, - bool forward_to_peers) - { - auto self_connection_handle = GetConnectionId(); - - auto track_alias = GetTrackAlias(); - if (!track_alias.has_value()) { - SPDLOG_DEBUG("Data without valid track alias"); - return; - } - - peering::DataType d_type; - - if (is_datagram_) { - d_type = peering::DataType::kDatagram; - } else { - d_type = peering::DataType::kExistingStream; - - if (is_new_stream) { - d_type = peering::DataType::kNewStream; - - if (GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count() == 0) { - // Use default if delivery timeout is not set - SetDeliveryTimeout(std::chrono::milliseconds(server_.config_.object_ttl_)); - } - } - } - - // Fanout object to subscribe namespaces - for (const auto [_, conn_subs] : sub_namespaces_) { - for (const auto& [_, handler] : conn_subs) { - handler->ForwardPublishedData(GetTrackAlias().value(), is_new_stream, group_id, subgroup_id, data); - } - } - // Fanout object to subscribers - for (auto& [conn_handle, pub_handler] : subscribers_) { - if (conn_handle == 0) { // from peer - if (forward_to_peers) { - server_.peer_manager_.ClientDataRecv( - *track_alias, - GetPriority(), - GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(), - d_type, - group_id, - subgroup_id, - data); - } - continue; - } - - if (pub_handler->SentFirstObject(group_id, subgroup_id)) { - pub_handler->ForwardPublishedData(is_new_stream, group_id, subgroup_id, data); + if (!is_from_peer_ && subscribers_.contains(0)) { + const auto ttl = + GetDeliveryTimeout().value_or(std::chrono::milliseconds(server_.config_.object_ttl_)).count(); + server_.SendObjectToPeers( + GetTrackAlias().value(), GetPriority(), ttl, is_datagram_, object_headers, data, stream_mode); } + } catch (const std::exception& e) { + SPDLOG_ERROR("Caught exception trying to publish. (error={})", e.what()); } } @@ -531,38 +293,25 @@ namespace laps { } } - void SubscribeTrackHandler::StreamClosed(std::uint64_t stream_id, bool use_reset) + void SubscribeTrackHandler::SubgroupEnded(std::uint64_t group_id, std::uint64_t subgroup_id, bool reset) { - pending_source_buffers_.erase(stream_id); - - auto stream_it = streams_.find(stream_id); - if (stream_it == streams_.end()) { - return; + if (!is_from_peer_ && GetTrackAlias().has_value() && subscribers_.contains(0)) { + server_.PeerSubgroupEnded(GetTrackAlias().value(), group_id, subgroup_id, reset); } for (auto& [conn_handle, pub_handler] : subscribers_) { if (conn_handle == 0) { - // Notify peering manager - if (GetTrackAlias().has_value()) { - server_.peer_manager_.EndSubgroup(GetTrackAlias().value(), - stream_it->second.current_group_id, - stream_it->second.current_subgroup_id, - use_reset); - } - } else { - pub_handler->EndSubgroup( - stream_it->second.current_group_id, stream_it->second.current_subgroup_id, !use_reset); + continue; } + + pub_handler->EndSubgroup(group_id, subgroup_id, !reset); } - for (const auto [_, conn_subs] : sub_namespaces_) { + for (const auto& [_, conn_subs] : sub_namespaces_) { for (const auto& [_, handler] : conn_subs) { - handler->EndSubgroup( - stream_it->second.current_group_id, stream_it->second.current_subgroup_id, !use_reset); + handler->EndSubgroup(group_id, subgroup_id, !reset); } } - - streams_.erase(stream_it); } void SubscribeTrackHandler::SetFromPeer() diff --git a/src/subscribe_handler.h b/src/subscribe_handler.h index 3974666..c7d428d 100644 --- a/src/subscribe_handler.h +++ b/src/subscribe_handler.h @@ -27,14 +27,10 @@ namespace laps { ~SubscribeTrackHandler(); - void StreamClosed(std::uint64_t stream_id, bool use_reset) override; - - void StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) override; - void StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) override; - void DgramDataRecv(std::shared_ptr> data) override; void ObjectReceived(const quicr::ObjectHeaders& object_headers, quicr::BytesSpan data, std::optional stream_mode = std::nullopt) override; + void SubgroupEnded(std::uint64_t group_id, std::uint64_t subgroup_id, bool reset) override; void MetricsSampled(const quicr::SubscribeTrackMetrics& metrics) override; void StatusChanged(Status status) override; @@ -106,14 +102,6 @@ namespace laps { std::size_t SubscriberCount() const { return subscribers_.size(); } private: - void TryProcessStreamData(uint64_t stream_id, StreamContext& stream); - - void ForwardReceivedData(bool is_new_stream, - uint64_t group_id, - uint64_t subgroup_id, - std::shared_ptr> data, - bool forward_to_peers = true); - void UpdateTrackedProperties(std::optional extensions, std::optional immutable_extensions); @@ -123,10 +111,6 @@ namespace laps { bool is_datagram_{ false }; bool is_from_peer_{ false }; // Indicates that the subscribe handler was created by peer manager for recv data - // Original receive buffers retained until the subgroup header is complete so they can be forwarded without - // copying when the track alias is unchanged. - std::map>>> pending_source_buffers_; - /** * @brief Map of subscribers that have subscribed to this content * From 21d993b0d18adaad8542398bdc8118b6ef8ea397 Mon Sep 17 00:00:00 2001 From: trigaux Date: Thu, 17 Sep 2026 11:59:18 -0400 Subject: [PATCH 2/2] Add back forwarding. --- dependencies/libquicr | 2 +- src/client_manager.cc | 67 +++------- src/client_manager.h | 4 +- src/publish_namespace_handler.cc | 14 +++ src/publish_namespace_handler.h | 2 + src/subscribe_handler.cc | 209 +++++++++++++++++++++++++++++-- src/subscribe_handler.h | 24 +++- 7 files changed, 261 insertions(+), 61 deletions(-) diff --git a/dependencies/libquicr b/dependencies/libquicr index c88c58c..8707282 160000 --- a/dependencies/libquicr +++ b/dependencies/libquicr @@ -1 +1 @@ -Subproject commit c88c58c73ce853807858002f79e8ab9feeda1e73 +Subproject commit 8707282cf3e98bff10b892df3fdc1b3678b7c82f diff --git a/src/client_manager.cc b/src/client_manager.cc index 1bf4674..b9d27c9 100644 --- a/src/client_manager.cc +++ b/src/client_manager.cc @@ -1722,13 +1722,18 @@ namespace laps { } auto& stream = peer_rx_streams_[{ track_full_name_hash, *stream_id }]; + if (stream.started) { + it->second->StreamBytesForwarded(stream.group_id, stream.subgroup_id, std::move(data)); + return; + } + if (!stream.header_initialized) { stream.buffer.InitAny(); stream.header_initialized = true; } stream.buffer.Push(std::span(*data)); - TryParsePeerStream(stream, *it->second); + TryStartPeerStream(stream, *it->second); } void ClientManager::PeerStreamClosed(std::uint64_t track_full_name_hash, uint64_t stream_id, bool reset) @@ -1746,64 +1751,30 @@ namespace laps { peer_rx_streams_.erase(stream_it); } - void ClientManager::TryParsePeerStream(PeerRxStreamState& stream, SubscribeTrackHandler& handler) + void ClientManager::TryStartPeerStream(PeerRxStreamState& stream, SubscribeTrackHandler& handler) { auto& s_hdr = stream.buffer.GetAny(); if (!(stream.buffer >> s_hdr)) { return; } - if (s_hdr.priority.has_value()) { - handler.SetPriority(*s_hdr.priority); - } - - while (!stream.buffer.Empty()) { - if (!stream.buffer.AnyHasValueB()) { - stream.buffer.InitAnyB(); - } - - auto& obj = stream.buffer.GetAnyB(); - obj.properties.emplace(*s_hdr.properties); - if (!(stream.buffer >> obj)) { + auto subgroup_id = s_hdr.subgroup_id; + if (!subgroup_id.has_value()) { + subgroup_id = stream.buffer.DecodeUintV(false); + if (!subgroup_id.has_value()) { return; } + } - handler.subscribe_track_metrics_.objects_received++; - - const bool same_subgroup = stream.next_object_id.has_value() && stream.group_id == s_hdr.group_id && - stream.subgroup_id == s_hdr.subgroup_id.value_or(stream.subgroup_id); - stream.next_object_id = same_subgroup ? *stream.next_object_id + obj.object_delta : obj.object_delta; - - if (!s_hdr.subgroup_id.has_value()) { - if (obj.properties->subgroup_id_mode != quicr::messages::SubgroupIdType::kSetFromFirstObject) { - throw quicr::messages::ProtocolViolationException("Subgoup ID mismatch"); - } - s_hdr.subgroup_id = stream.next_object_id; - } - - stream.group_id = s_hdr.group_id; - stream.subgroup_id = *s_hdr.subgroup_id; + stream.group_id = s_hdr.group_id; + stream.subgroup_id = *subgroup_id; + stream.started = true; - std::optional stream_mode; - if (!same_subgroup && s_hdr.properties.has_value()) { - stream_mode.emplace(*s_hdr.properties); - } + handler.SubgroupStarted(s_hdr.group_id, *subgroup_id, s_hdr.priority, *s_hdr.properties); - handler.ObjectReceived({ s_hdr.group_id, - *stream.next_object_id, - *s_hdr.subgroup_id, - obj.payload.size(), - obj.object_status, - s_hdr.priority, - std::nullopt, - quicr::TrackMode::kStream, - obj.extensions, - obj.immutable_extensions }, - obj.payload, - stream_mode); - - *stream.next_object_id += 1; - stream.buffer.ResetAnyB(); + auto remaining = stream.buffer.TakeAll(); + if (!remaining.empty()) { + handler.StreamBytesForwarded(stream.group_id, stream.subgroup_id, std::move(remaining)); } } diff --git a/src/client_manager.h b/src/client_manager.h index 98bdeaa..1774df1 100644 --- a/src/client_manager.h +++ b/src/client_manager.h @@ -165,8 +165,8 @@ namespace laps { quicr::StreamBuffer buffer; uint64_t group_id{ 0 }; uint64_t subgroup_id{ 0 }; - std::optional next_object_id; bool header_initialized{ false }; + bool started{ false }; }; struct PeerTxSubgroupState @@ -185,7 +185,7 @@ namespace laps { void PeerSubgroupEnded(uint64_t track_alias, uint64_t group_id, uint64_t subgroup_id, bool reset); - void TryParsePeerStream(PeerRxStreamState& stream, SubscribeTrackHandler& handler); + void TryStartPeerStream(PeerRxStreamState& stream, SubscribeTrackHandler& handler); /** * @brief Connection handle of the session that received a request, or zero if it is already gone */ diff --git a/src/publish_namespace_handler.cc b/src/publish_namespace_handler.cc index 03bc7fd..6f97bce 100644 --- a/src/publish_namespace_handler.cc +++ b/src/publish_namespace_handler.cc @@ -75,6 +75,20 @@ laps::PublishNamespaceHandler::ForwardPublishedData(std::uint64_t track_full_nam return quicr::PublishTrackHandler::PublishObjectStatus::kOk; } +bool +laps::PublishNamespaceHandler::SentFirstObject(std::uint64_t track_full_name_hash, + uint64_t group_id, + uint64_t subgroup_id) const +{ + const auto pub_it = handlers_.find(track_full_name_hash); + if (pub_it == handlers_.end()) { + return false; + } + + const auto handler = std::dynamic_pointer_cast(pub_it->second); + return handler && handler->SentFirstObject(group_id, subgroup_id); +} + void laps::PublishNamespaceHandler::UpdateTrackRanking( std::span> ordered_tracks) diff --git a/src/publish_namespace_handler.h b/src/publish_namespace_handler.h index fb37f05..329dbef 100644 --- a/src/publish_namespace_handler.h +++ b/src/publish_namespace_handler.h @@ -43,6 +43,8 @@ namespace laps { uint64_t subgroup_id, std::shared_ptr> data) override; + bool SentFirstObject(std::uint64_t track_full_name_hash, uint64_t group_id, uint64_t subgroup_id) const; + void PublishTrack(std::shared_ptr handler) override; static auto Create(const quicr::TrackNamespace& prefix, std::weak_ptr tick_service) diff --git a/src/subscribe_handler.cc b/src/subscribe_handler.cc index 6546156..d1e8f22 100644 --- a/src/subscribe_handler.cc +++ b/src/subscribe_handler.cc @@ -6,7 +6,10 @@ #include "publish_handler.h" #include "publish_namespace_handler.h" -#include +#include "peering/messages/data_header.h" + +#include +#include #include namespace laps { @@ -16,12 +19,12 @@ namespace laps { ClientManager& server, std::weak_ptr tick_service, bool is_publisher_initiated) - : quicr::SubscribeTrackHandler(full_track_name, - priority, - group_order, - std::monostate{}, - std::nullopt, - is_publisher_initiated) + : quicr::ForwardingSubscribeTrackHandler(full_track_name, + priority, + group_order, + std::monostate{}, + std::nullopt, + is_publisher_initiated) , server_(server) , tick_service_(std::move(tick_service)) { @@ -166,6 +169,185 @@ namespace laps { } } + void SubscribeTrackHandler::SubgroupStarted(std::uint64_t group_id, + std::uint64_t subgroup_id, + std::optional priority, + quicr::messages::StreamHeaderProperties properties) + { + is_datagram_ = false; + + if (priority.has_value()) { + SetPriority(*priority); + } + + auto& stream = forwarded_subgroups_[{ group_id, subgroup_id }]; + stream.properties.emplace(properties); + stream.priority = priority; + stream.parse_buffer.InitAnyB(); + + if (!GetTrackAlias().has_value()) { + return; + } + + quicr::messages::StreamHeaderSubGroup hdr; + hdr.properties.emplace(properties); + hdr.track_alias = GetTrackAlias().value(); + hdr.group_id = group_id; + if (properties.subgroup_id_mode == quicr::messages::SubgroupIdType::kExplicit) { + hdr.subgroup_id = subgroup_id; + } + if (!properties.default_priority) { + hdr.priority = priority; + } + + auto header_bytes = std::make_shared>(); + *header_bytes << hdr; + + if (GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count() == 0) { + SetDeliveryTimeout(std::chrono::milliseconds(server_.config_.object_ttl_)); + } + + const auto ttl = GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(); + + for (const auto& [_, conn_subs] : sub_namespaces_) { + for (const auto& [_, handler] : conn_subs) { + handler->ForwardPublishedData(*GetTrackAlias(), true, group_id, subgroup_id, header_bytes); + } + } + + for (auto& [conn_handle, pub_handler] : subscribers_) { + if (conn_handle == 0) { + if (!is_from_peer_) { + server_.peer_manager_.ClientDataRecv(*GetTrackAlias(), + GetPriority(), + ttl, + peering::DataType::kNewStream, + group_id, + subgroup_id, + header_bytes); + } + continue; + } + + if (pub_handler->SentFirstObject(group_id, subgroup_id)) { + pub_handler->ForwardPublishedData(true, group_id, subgroup_id, header_bytes); + } + } + } + + void SubscribeTrackHandler::StreamBytesForwarded(std::uint64_t group_id, + std::uint64_t subgroup_id, + quicr::Bytes&& data) + { + StreamBytesForwarded(group_id, subgroup_id, std::make_shared(std::move(data))); + } + + void SubscribeTrackHandler::StreamBytesForwarded(std::uint64_t group_id, + std::uint64_t subgroup_id, + std::shared_ptr> bytes) + { + is_datagram_ = false; + + if (!bytes || bytes->empty()) { + return; + } + + if (const auto track_alias = GetTrackAlias(); track_alias.has_value()) { + const auto ttl = GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(); + + for (const auto& [_, conn_subs] : sub_namespaces_) { + for (const auto& [_, handler] : conn_subs) { + handler->ForwardPublishedData(*track_alias, false, group_id, subgroup_id, bytes); + } + } + + for (auto& [conn_handle, pub_handler] : subscribers_) { + if (conn_handle == 0) { + if (!is_from_peer_) { + server_.peer_manager_.ClientDataRecv(*track_alias, + GetPriority(), + ttl, + peering::DataType::kExistingStream, + group_id, + subgroup_id, + bytes); + } + continue; + } + + if (pub_handler->SentFirstObject(group_id, subgroup_id)) { + pub_handler->ForwardPublishedData(false, group_id, subgroup_id, bytes); + } + } + } + + auto stream_it = forwarded_subgroups_.find({ group_id, subgroup_id }); + if (stream_it == forwarded_subgroups_.end() || !stream_it->second.properties.has_value()) { + return; + } + + auto& stream = stream_it->second; + stream.parse_buffer.Push(std::span(*bytes)); + + while (!stream.parse_buffer.Empty()) { + if (!stream.parse_buffer.AnyHasValueB()) { + stream.parse_buffer.InitAnyB(); + } + + auto& obj = stream.parse_buffer.GetAnyB(); + obj.properties.emplace(*stream.properties); + if (!(stream.parse_buffer >> obj)) { + return; + } + + const bool is_first_object = !stream.next_object_id.has_value(); + stream.next_object_id = is_first_object ? obj.object_delta : *stream.next_object_id + obj.object_delta; + + subscribe_track_metrics_.objects_received++; + subscribe_track_metrics_.bytes_received += obj.payload.size(); + + std::optional stream_mode; + if (is_first_object) { + stream_mode.emplace(*stream.properties); + } + + ObjectReceived({ group_id, + *stream.next_object_id, + subgroup_id, + obj.payload.size(), + obj.object_status, + stream.priority, + std::nullopt, + quicr::TrackMode::kStream, + obj.extensions, + obj.immutable_extensions }, + obj.payload, + stream_mode); + + *stream.next_object_id += 1; + stream.parse_buffer.ResetAnyB(); + + if (is_first_object) { + const auto remaining = stream.parse_buffer.Data(); + if (remaining.empty() || !GetTrackAlias().has_value()) { + continue; + } + + auto rest = std::make_shared>(remaining.begin(), remaining.end()); + for (const auto& [_, conn_subs] : sub_namespaces_) { + for (const auto& [_, handler] : conn_subs) { + handler->ForwardPublishedData(*GetTrackAlias(), false, group_id, subgroup_id, rest); + } + } + for (auto& [conn_handle, pub_handler] : subscribers_) { + if (conn_handle != 0 && pub_handler->SentFirstObject(group_id, subgroup_id)) { + pub_handler->ForwardPublishedData(false, group_id, subgroup_id, rest); + } + } + } + } + } + void SubscribeTrackHandler::ObjectReceived(const quicr::ObjectHeaders& object_headers, quicr::BytesSpan data, std::optional stream_mode) @@ -207,6 +389,11 @@ namespace laps { try { for (const auto& [_, conn_subs] : sub_namespaces_) { for (const auto& [_, handler] : conn_subs) { + if (!is_datagram_ && + handler->SentFirstObject(GetTrackAlias().value(), object_headers.group_id, + object_headers.subgroup_id)) { + continue; + } handler->PublishObject(GetTrackAlias().value(), object_headers, data, stream_mode); } } @@ -219,13 +406,18 @@ namespace laps { pub_handler->SetDefaultTrackMode(is_datagram_ ? quicr::TrackMode::kDatagram : quicr::TrackMode::kStream); + if (!is_datagram_ && + pub_handler->SentFirstObject(object_headers.group_id, object_headers.subgroup_id)) { + continue; + } + if (object_headers.group_id >= pub_handler->start_location_.group && object_headers.object_id >= pub_handler->start_location_.object) { pub_handler->PublishObject(object_headers, data, stream_mode); } } - if (!is_from_peer_ && subscribers_.contains(0)) { + if (is_datagram_ && !is_from_peer_ && subscribers_.contains(0)) { const auto ttl = GetDeliveryTimeout().value_or(std::chrono::milliseconds(server_.config_.object_ttl_)).count(); server_.SendObjectToPeers( @@ -295,6 +487,7 @@ namespace laps { void SubscribeTrackHandler::SubgroupEnded(std::uint64_t group_id, std::uint64_t subgroup_id, bool reset) { + forwarded_subgroups_.erase({ group_id, subgroup_id }); if (!is_from_peer_ && GetTrackAlias().has_value() && subscribers_.contains(0)) { server_.PeerSubgroupEnded(GetTrackAlias().value(), group_id, subgroup_id, reset); } diff --git a/src/subscribe_handler.h b/src/subscribe_handler.h index c7d428d..71f304e 100644 --- a/src/subscribe_handler.h +++ b/src/subscribe_handler.h @@ -3,7 +3,9 @@ #include "client_manager.h" #include "publish_namespace_handler.h" -#include +#include +#include +#include #include #include @@ -13,7 +15,7 @@ namespace laps { * @brief Subscribe track handler * @details Subscribe track handler used for the subscribe command line option. */ - class SubscribeTrackHandler : public quicr::SubscribeTrackHandler + class SubscribeTrackHandler : public quicr::ForwardingSubscribeTrackHandler { public: static constexpr uint64_t kRefreshRankingIntervalMs = 120; @@ -27,6 +29,14 @@ namespace laps { ~SubscribeTrackHandler(); + void SubgroupStarted(std::uint64_t group_id, + std::uint64_t subgroup_id, + std::optional priority, + quicr::messages::StreamHeaderProperties properties) override; + void StreamBytesForwarded(std::uint64_t group_id, std::uint64_t subgroup_id, quicr::Bytes&& data) override; + void StreamBytesForwarded(std::uint64_t group_id, + std::uint64_t subgroup_id, + std::shared_ptr> data); void ObjectReceived(const quicr::ObjectHeaders& object_headers, quicr::BytesSpan data, std::optional stream_mode = std::nullopt) override; @@ -102,6 +112,14 @@ namespace laps { std::size_t SubscriberCount() const { return subscribers_.size(); } private: + struct ForwardedSubgroup + { + std::optional properties; + std::optional priority; + quicr::StreamBuffer parse_buffer; + std::optional next_object_id; + }; + void UpdateTrackedProperties(std::optional extensions, std::optional immutable_extensions); @@ -131,6 +149,8 @@ namespace laps { */ std::map tracked_properties_value_; + std::map, ForwardedSubgroup> forwarded_subgroups_; + std::weak_ptr track_ranking_; }; } // namespace laps