UPdate to use the new streambuffer flow.

src/client_manager.cc

@@ -14,7 +14,10 @@
 #include "subscribe_handler.h"
 #include "transport_helpers.h"
 
+#include "peering/messages/data_header.h"
+
 #include <ranges>
+#include <span>
 
 namespace laps {
     ClientManager::ClientManager(State& state,
@@ -1705,32 +1708,157 @@ namespace laps {
                                          std::shared_ptr<const std::vector<uint8_t>> 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.started) {
+            it->second->StreamBytesForwarded(stream.group_id, stream.subgroup_id, std::move(data));
+            return;
+        }
+
+        if (!stream.header_initialized) {
+            stream.buffer.InitAny<quicr::messages::StreamHeaderSubGroup>();
+            stream.header_initialized = true;
         }
+
+        stream.buffer.Push(std::span<const uint8_t>(*data));
+        TryStartPeerStream(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::TryStartPeerStream(PeerRxStreamState& stream, SubscribeTrackHandler& handler)
+    {
+        auto& s_hdr = stream.buffer.GetAny<quicr::messages::StreamHeaderSubGroup>();
+        if (!(stream.buffer >> s_hdr)) {
+            return;
+        }
+
+        auto subgroup_id = s_hdr.subgroup_id;
+        if (!subgroup_id.has_value()) {
+            subgroup_id = stream.buffer.DecodeUintV(false);
+            if (!subgroup_id.has_value()) {
+                return;
+            }
+        }
+
+        stream.group_id = s_hdr.group_id;
+        stream.subgroup_id = *subgroup_id;
+        stream.started = true;
+
+        handler.SubgroupStarted(s_hdr.group_id, *subgroup_id, s_hdr.priority, *s_hdr.properties);
+
+        auto remaining = stream.buffer.TakeAll();
+        if (!remaining.empty()) {
+            handler.StreamBytesForwarded(stream.group_id, stream.subgroup_id, std::move(remaining));
+        }
+    }
+
+    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<quicr::messages::StreamHeaderProperties> stream_mode)
+    {
+        auto bytes = std::make_shared<std::vector<uint8_t>>();
+        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<quicr::Session>& session,

src/client_manager.h

@@ -7,6 +7,8 @@
 #include "track_ranking.h"
 #include <peering/peer_manager.h>
 #include <quicr/containers/cache.h>
+#include <quicr/containers/stream_buffer.h>
+#include <quicr/messages/messages.h>
 #include <quicr/messages/object.h>
 #include <quicr/session.h>
 #include <quicr/session_callbacks.h>
@@ -16,6 +18,7 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <set>
 
 namespace laps {
@@ -157,6 +160,32 @@ namespace laps {
         friend class peering::PeerManager;
 
       private:
+        struct PeerRxStreamState
+        {
+            quicr::StreamBuffer<uint8_t> buffer;
+            uint64_t group_id{ 0 };
+            uint64_t subgroup_id{ 0 };
+            bool header_initialized{ false };
+            bool started{ false };
+        };
+
+        struct PeerTxSubgroupState
+        {
+            std::optional<uint64_t> last_object_id;
+            std::optional<quicr::messages::StreamHeaderProperties> 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<quicr::messages::StreamHeaderProperties> stream_mode);
+
+        void PeerSubgroupEnded(uint64_t track_alias, uint64_t group_id, uint64_t subgroup_id, bool reset);
+
+        void TryStartPeerStream(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<std::uint64_t, std::shared_ptr<TrackRanking>> track_rankings_;
 
+        /// Inbound peer subgroup parse state, keyed by track full name hash and peer stream id
+        std::map<std::pair<std::uint64_t, std::uint64_t>, PeerRxStreamState> peer_rx_streams_;
+
+        /// Outbound peer subgroup serialization state, keyed by track alias, group id, and subgroup id
+        std::map<std::pair<std::uint64_t, std::pair<std::uint64_t, std::uint64_t>>, PeerTxSubgroupState>
+          peer_tx_subgroups_;
+
         friend class SubscribeTrackHandler;
         friend class PublishTrackHandler;
         friend class FetchTrackHandler;

src/fetch_handler.cc

@@ -4,7 +4,6 @@
 #include "fetch_handler.h"
 #include "config.h"
 #include <quicr/handlers/fetch_track_handler.h>
-#include <quicr/session.h>
 
 namespace laps {
     FetchTrackHandler::FetchTrackHandler(const std::shared_ptr<quicr::PublishFetchHandler> 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<quicr::messages::StreamHeaderProperties> 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<quicr::messages::FetchHeader>();
-        TryForwardInitialStreamData(stream_id, stream_it->second);
-    }
-
-    void FetchTrackHandler::StreamDataRecv(uint64_t stream_id, std::shared_ptr<const std::vector<uint8_t>> 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<quicr::messages::FetchHeader>();
-        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<quicr::Bytes>();
-        *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<std::vector<uint8_t>>(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)

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<const std::vector<uint8_t>> data) override;
+        void ObjectReceived(const quicr::ObjectHeaders& object_headers,
+                            quicr::BytesSpan data,
+                            std::optional<quicr::messages::StreamHeaderProperties> stream_mode = std::nullopt) override;
 
       private:
-        void TryForwardInitialStreamData(uint64_t stream_id, StreamContext& stream);
-
         bool initial_stream_data_forwarded_{ false };
         std::shared_ptr<quicr::PublishFetchHandler> publish_fetch_handler_;
     };