Interop fixes and add subscribe namespace support and handling

CMakeLists.txt

@@ -18,7 +18,7 @@ set (CMAKE_POLICY_VERSION_MINIMUM 3.5)
 ### Define the project name and version
 ###       Version will be added to version output of apps and library
 project(laps
-        VERSION 0.18.23
+        VERSION 0.18.31
         DESCRIPTION "Latency Aware Publish/Subscribe"
         LANGUAGES CXX)
 
@@ -57,10 +57,11 @@ endif()
 ###
 add_subdirectory(dependencies)
 
-# quicr/log.h picks between std::format and fmt on this definition, which libquicr only defines for its own
-# sources. The relay logs through quicr::Logger, so it has to make the same choice libquicr did.
-if(HAS_STD_VFORMAT)
-    add_compile_definitions(QUICR_HAVE_STD_FORMAT)
+# quicr/utilities/format.h selects std::format or {fmt} with QUICR_HAS_STD_FORMAT.
+# libquicr only adds that definition under its own directory, so relay sources that
+# include quicr headers have to repeat the same choice.
+if(QUICR_HAS_STD_FORMAT)
+    add_compile_definitions(QUICR_HAS_STD_FORMAT)
 endif()
 
 # spdlog compiles a bundled copy of fmt by default. libquicr also links fmt::fmt

src/CMakeLists.txt

@@ -5,7 +5,6 @@ if(CLANG_TIDY_EXE)
   set_target_properties(quicr PROPERTIES CXX_CLANG_TIDY "${CLANG_TIDY_EXE}")
 endif()
 
-
 add_executable(lapsRelay
         main.cc
         config.cc
@@ -30,7 +29,11 @@ add_executable(lapsRelay
 )
 
 target_link_libraries(lapsRelay PRIVATE quicr spdlog::spdlog)
-target_include_directories(lapsRelay PRIVATE ${PROJECT_SOURCE_DIR}/dependencies/oss )
+target_include_directories(lapsRelay PRIVATE
+        ${PROJECT_SOURCE_DIR}/dependencies/oss
+        # Peer sessions read quicr::Stream directly. That type is defined for the library, not installed
+        # with the public headers.
+        ${PROJECT_SOURCE_DIR}/dependencies/libquicr/src)
 
 target_compile_options(lapsRelay
         PRIVATE

src/client_manager.h

@@ -164,52 +164,52 @@ namespace laps {
         void MetricsSampled(const std::shared_ptr<quicr::Session>& session,
                             const quicr::ConnectionMetrics& metrics) override;
 
-        quicr::Reply<void, quicr::ErrorCode> ClientSetupReceived(
+        quicr::Expected<void, quicr::Error<quicr::ErrorCode>> SetupReceived(
           const std::shared_ptr<quicr::Session>& session,
-          const quicr::ClientSetupAttributes& client_setup_attributes) override;
+          const quicr::SetupAttributes& setup_attributes) override;
 
         quicr::Reply<std::vector<quicr::TrackNamespace>, quicr::RequestErrorCode> SubscribeTracksReceived(
           const std::shared_ptr<quicr::Session>& session,
           const quicr::TrackNamespace& prefix_namespace,
           const quicr::SubscribeNamespaceAttributes& attributes) override;
 
-        quicr::Reply<void, quicr::ErrorCode> UnsubscribeNamespaceReceived(
+        quicr::Reply<std::vector<quicr::TrackNamespace>, quicr::RequestErrorCode> SubscribeNamespaceReceived(
           const std::shared_ptr<quicr::Session>& session,
-          const quicr::TrackNamespace& prefix_namespace) override;
+          const quicr::TrackNamespace& prefix_namespace,
+          const quicr::SubscribeNamespaceAttributes& attributes) override;
+
+        void UnsubscribeNamespaceReceived(const std::shared_ptr<quicr::Session>& session,
+                                          const quicr::TrackNamespace& prefix_namespace) override;
 
         quicr::Reply<void, quicr::PublishNamespaceErrorCode> PublishNamespaceReceived(
           const std::shared_ptr<quicr::Session>& session,
           const quicr::TrackNamespace& track_namespace,
           const quicr::PublishNamespaceAttributes& attributes) override;
 
-        quicr::Reply<void, quicr::PublishNamespaceErrorCode> PublishNamespaceDoneReceived(
-          const std::shared_ptr<quicr::Session>& session,
-          std::uint64_t request_id) override;
+        void PublishNamespaceDoneReceived(const std::shared_ptr<quicr::Session>& session,
+                                          std::uint64_t request_id) override;
 
         quicr::Reply<const quicr::PublishResponse, quicr::PublishErrorCode> PublishReceived(
           const std::shared_ptr<quicr::Session>& session,
           std::uint64_t request_id,
           const quicr::PublishAttributes& publish_attributes,
           std::weak_ptr<quicr::SubscribeNamespaceHandler> sub_ns_handler) override;
 
-        quicr::Reply<void, quicr::ErrorCode> PublishDoneReceived(const std::shared_ptr<quicr::Session>& session,
-                                                                 std::uint64_t request_id) override;
+        void PublishDoneReceived(const std::shared_ptr<quicr::Session>& session, std::uint64_t request_id) override;
 
-        quicr::Reply<quicr::RequestResponse, quicr::RequestErrorCode> SubscribeReceived(
+        quicr::Reply<quicr::SubscribeResponse, quicr::RequestErrorCode> SubscribeReceived(
           const std::shared_ptr<quicr::Session>& session,
           std::uint64_t request_id,
           const quicr::FullTrackName& track_full_name,
           const quicr::SubscribeAttributes& subscribe_attributes) override;
 
-        quicr::Reply<void, quicr::ErrorCode> UnsubscribeReceived(const std::shared_ptr<quicr::Session>& session,
-                                                                 std::uint64_t request_id) override;
+        void UnsubscribeReceived(const std::shared_ptr<quicr::Session>& session, std::uint64_t request_id) override;
 
         quicr::Reply<void, quicr::ErrorCode> NewGroupRequested(const quicr::FullTrackName& track_full_name,
                                                                std::uint64_t group_id) override;
 
-        quicr::Reply<quicr::RequestResponse, quicr::RequestErrorCode> TrackStatusReceived(
+        quicr::Reply<quicr::TrackStatusResponse, quicr::RequestErrorCode> TrackStatusReceived(
           const std::shared_ptr<quicr::Session>& session,
-          std::uint64_t request_id,
           const quicr::FullTrackName& track_full_name) override;
 
         quicr::Reply<const quicr::FetchResponse, quicr::FetchErrorCode> StandaloneFetchReceived(
@@ -259,6 +259,27 @@ namespace laps {
          */
         void FanoutPublishNamespaceDone(const quicr::TrackNamespace& track_namespace);
 
+        /**
+         * @brief Send NAMESPACE or NAMESPACE_DONE to subscribe-namespace subscribers whose prefix matches
+         *
+         * @param track_namespace      Fully qualified namespace that became active or ended
+         * @param exclude_connection   Publishing connection. Skipped unless self-forwarding is allowed
+         * @param done                 True sends NAMESPACE_DONE, false sends NAMESPACE
+         */
+        void NotifySubscribeNamespaces(const quicr::TrackNamespace& track_namespace,
+                                       std::uint64_t exclude_connection,
+                                       bool done);
+
+        /**
+         * @brief True when some publisher still has this namespace announced or a track in it
+         */
+        bool NamespaceStillPublished(const quicr::TrackNamespace& track_namespace) const;
+
+        /**
+         * @brief Drop one subscribe-namespace subscription and any NAMESPACE already sent for it
+         */
+        void DropSubscribeNamespace(std::uint64_t connection_handle, const quicr::TrackNamespace& prefix);
+
         quicr::Reply<const quicr::PublishResponse, quicr::PublishErrorCode> PublishReceivedInternal(
           std::uint64_t connection_handle,
           uint64_t request_id,
@@ -298,6 +319,9 @@ namespace laps {
         std::map<std::pair<quicr::TrackNamespace, std::uint64_t>, std::shared_ptr<quicr::PublishNamespaceHandler>>
           fanout_namespaces_;
 
+        /// Namespaces already reported with NAMESPACE, keyed by namespace and subscriber connection
+        std::set<std::pair<quicr::TrackNamespace, std::uint64_t>> notified_namespaces_;
+
         /**
          * @brief Map of atomic bools to mark if a fetch thread should be interrupted.
          */

src/fetch_handler.cc

@@ -18,64 +18,14 @@ namespace laps {
     {
     }
 
-    void FetchTrackHandler::StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer)
+    void FetchTrackHandler::ObjectReceived(
+      const quicr::ObjectHeaders& object_headers,
+      quicr::BytesSpan data,
+      [[maybe_unused]] 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)) {
-            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());
-        }
-
-        publish_fetch_handler_->ForwardPublishedData(true, 0, 0, std::move(bytes));
-        initial_stream_data_forwarded_ = true;
-        streams_.erase(stream_id);
+        // The session has already parsed the fetch header, so the outbound fetch writes its own, using this
+        // handler's request id.
+        publish_fetch_handler_->PublishObject(object_headers, data);
     }
 
     void FetchTrackHandler::StatusChanged(Status status)

src/fetch_handler.h

@@ -32,13 +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_;
     };
 } // namespace laps

src/peering/info_base.h

@@ -7,9 +7,11 @@
 #include "peer_session.h"
 #include "peering/messages/subscribe_info.h"
 
+#include <functional>
 #include <map>
 #include <quicr/messages/message_serialisation.h>
 #include <set>
+#include <unordered_map>
 
 namespace laps::peering {
 
@@ -148,17 +150,45 @@ namespace laps::peering {
 
         std::map<std::uint64_t, std::map<NodeIdValueType, SubscribeInfo>> subscribes_;
 
+        /**
+         * @brief Identity of one egress stream held by a forwarding entry
+         *
+         * @details Client entries use `a` as the group id and `b` as the subgroup id. Peer entries use `a` as
+         *      the ingress QUIC stream id and leave `b` at 0. Those two maps do not share entries.
+         */
+        struct StreamKey
+        {
+            uint64_t a{ 0 };
+            uint64_t b{ 0 };
+
+            friend bool operator==(const StreamKey& lhs, const StreamKey& rhs) noexcept
+            {
+                return lhs.a == rhs.a && lhs.b == rhs.b;
+            }
+        };
+
+        struct StreamKeyHash
+        {
+            std::size_t operator()(const StreamKey& key) const noexcept
+            {
+                const auto h1 = std::hash<uint64_t>{}(key.a);
+                const auto h2 = std::hash<uint64_t>{}(key.b);
+                return h1 ^ (h2 + 0x9e3779b97f4a7c15ULL + (h1 << 6) + (h1 >> 2));
+            }
+        };
+
         struct FibEntry
         {
             uint64_t update_ref{ 0 }; ///< Random reference number to detect if entry was updated or not
 
             /**
-             * @brief Egress stream for each ingress stream, keyed by ingress stream ID
+             * @brief Egress stream for each forwarded flow
              *
              * @details The transport hands out stream handles rather than IDs, and holding one keeps the
-             *      stream usable for as long as objects are being forwarded onto it.
+             *      stream usable for as long as objects are being forwarded onto it. See StreamKey for how
+             *      each map identifies a flow.
              */
-            std::unordered_map<uint64_t, std::shared_ptr<quicr::Stream>> streams;
+            std::unordered_map<StreamKey, std::shared_ptr<quicr::Stream>, StreamKeyHash> streams;
 
             SubscribeNodeSetId out_sns_id; ///< Egress SNS ID
             decltype(nodes_best_)::mapped_type peer_session;