Draft 11 Control Messages (#553)

* Run parser for draft11

* Rerun with location

* remove duplicate auto-gen files

* add back the location stream ops

* Change control message payload size to 16bit

Resolves #552

* Regen messages

* ReasonPhrase<->ErrorReason

* Use request ID

* Use Location structure

* Add draft 11 subgroup header + logic (#554)

* Add draft 11 subgroup header + logic

* Make test case parameterized

* Naming

* Rename

* Add more location changes

* More location updates

* Standardize error reason/reason phrase

* Update type

* Implement new KVP MoQ structure (#555)

* Implement new KVP MoQ structure

- Add for parameters
- Consolidate uint64_t enum coding
- Move parameters to new structure

* Remove parameter parsing from parser

* Update for odd values

* Fix decode and tests

* Decode to encoded size

* more request id updates

* Use sizeof

* Fix tests

* more updates to support request id changes

* format

* Update remaining changes for draft-11

* Update tests for draft-11 changes

* Fix example resolve announce

* Fix merge & Request ID

* Fix test cases for new output

* Add stream operator for uint16_t

* Add ControlMessage encoder

---------

Co-authored-by: suhasHere <[email protected]>
Co-authored-by: Tim Evens <[email protected]>
Co-authored-by: Tim Evens <[email protected]>

Author: 3 people

cmd/examples/server.cpp

@@ -72,7 +72,7 @@ namespace qserver_vars {
      * @example
      *      track_alias = subscribe_alias_sub_id[conn_id][subscribe_id]
      */
-    std::map<quicr::ConnectionHandle, std::map<quicr::messages::SubscribeID, quicr::messages::TrackAlias>>
+    std::map<quicr::ConnectionHandle, std::map<quicr::messages::RequestID, quicr::messages::TrackAlias>>
       subscribe_alias_sub_id;
 
     /**
@@ -128,7 +128,7 @@ namespace qserver_vars {
     /**
      * @brief Map of atomic bools to mark if a fetch thread should be interrupted.
      */
-    std::map<std::pair<quicr::ConnectionHandle, quicr::messages::SubscribeID>, std::atomic_bool> stop_fetch;
+    std::map<std::pair<quicr::ConnectionHandle, quicr::messages::RequestID>, std::atomic_bool> stop_fetch;
 }
 
 /**
@@ -428,7 +428,7 @@ class MyServer : public quicr::Server
 
     void AnnounceReceived(quicr::ConnectionHandle connection_handle,
                           const quicr::TrackNamespace& track_namespace,
-                          const quicr::PublishAnnounceAttributes&) override
+                          const quicr::PublishAnnounceAttributes& attrs) override
     {
         auto th = quicr::TrackHash({ track_namespace, {}, std::nullopt });
 
@@ -469,7 +469,7 @@ class MyServer : public quicr::Server
             }
         }
 
-        ResolveAnnounce(connection_handle, track_namespace, sub_annos_connections, announce_response);
+        ResolveAnnounce(connection_handle, attrs.request_id, track_namespace, sub_annos_connections, announce_response);
 
         // Check if there are any subscribes. If so, send subscribe to announce for all tracks matching namespace
         for (const auto& [ns, sub_tracks] : qserver_vars::subscribe_active) {
@@ -528,7 +528,7 @@ class MyServer : public quicr::Server
         }
 
         // Remove active subscribes
-        std::vector<quicr::messages::SubscribeID> subscribe_ids;
+        std::vector<quicr::messages::RequestID> subscribe_ids;
         auto ta_conn_it = qserver_vars::subscribe_alias_sub_id.find(connection_handle);
         if (ta_conn_it != qserver_vars::subscribe_alias_sub_id.end()) {
             for (const auto& [sub_id, _] : ta_conn_it->second) {
@@ -662,16 +662,14 @@ class MyServer : public quicr::Server
             return;
         }
 
-        std::optional<uint64_t> latest_group_id = std::nullopt;
-        std::optional<uint64_t> latest_object_id = std::nullopt;
+        std::optional<quicr::messages::Location> largest_location = std::nullopt;
 
         auto cache_entry_it = qserver_vars::cache.find(th.track_fullname_hash);
         if (cache_entry_it != qserver_vars::cache.end()) {
             auto& [_, cache] = *cache_entry_it;
             if (const auto& latest_group = cache.Last(); latest_group && !latest_group->empty()) {
                 const auto& latest_object = std::prev(latest_group->end());
-                latest_group_id = latest_object->headers.group_id;
-                latest_object_id = latest_object->headers.object_id;
+                largest_location = { latest_object->headers.group_id, latest_object->headers.object_id };
             }
         }
 
@@ -681,8 +679,7 @@ class MyServer : public quicr::Server
                            quicr::SubscribeResponse::ReasonCode::kOk,
                            std::nullopt,
                            std::nullopt,
-                           latest_group_id,
-                           latest_object_id,
+                           largest_location,
                          });
 
         auto pub_track_h =
@@ -760,25 +757,22 @@ class MyServer : public quicr::Server
         }
     }
 
-    LargestAvailable GetLargestAvailable(const quicr::FullTrackName& track_name) override
+    std::optional<quicr::messages::Location> GetLargestAvailable(const quicr::FullTrackName& track_name) override
     {
         // Get the largest object from the cache.
-        std::optional<uint64_t> largest_group_id = std::nullopt;
-        std::optional<uint64_t> largest_object_id = std::nullopt;
+        std::optional<quicr::messages::Location> largest_location = std::nullopt;
+
         const auto& th = quicr::TrackHash(track_name);
         const auto cache_entry_it = qserver_vars::cache.find(th.track_fullname_hash);
         if (cache_entry_it != qserver_vars::cache.end()) {
             auto& [_, cache] = *cache_entry_it;
             if (const auto& latest_group = cache.Last(); latest_group && !latest_group->empty()) {
                 const auto& latest_object = std::prev(latest_group->end());
-                largest_group_id = latest_object->headers.group_id;
-                largest_object_id = latest_object->headers.object_id;
+                largest_location = { latest_object->headers.group_id, latest_object->headers.object_id };
             }
         }
-        if (!largest_group_id.has_value() || !largest_object_id.has_value()) {
-            return std::nullopt;
-        }
-        return std::make_pair(*largest_group_id, *largest_object_id);
+
+        return largest_location;
     }
 
     bool OnFetchOk(quicr::ConnectionHandle connection_handle,

include/quicr/client.h

@@ -137,11 +137,11 @@ namespace quicr {
          *      will send the protocol message based on the SubscribeResponse
          *
          * @param connection_handle        source connection ID
-         * @param subscribe_id             subscribe ID
+         * @param request_id               request ID
          * @param subscribe_response       response to for the subscribe
          */
         virtual void ResolveSubscribe(ConnectionHandle connection_handle,
-                                      uint64_t subscribe_id,
+                                      uint64_t request_id,
                                       const SubscribeResponse& subscribe_response);
 
         /**

include/quicr/common.h

@@ -26,7 +26,9 @@ namespace quicr {
      * @details Various attributes relative to the publish announce
      */
     struct PublishAnnounceAttributes
-    {};
+    {
+        uint64_t request_id{ 0 };
+    };
 
     /**
      * @brief Client Setup Attributes

include/quicr/detail/base_track_handler.h

@@ -4,6 +4,7 @@
 #pragma once
 
 #include "quicr/common.h"
+#include "quicr/detail/ctrl_message_types.h"
 #include "quicr/track_name.h"
 
 #include <optional>
@@ -42,11 +43,10 @@ namespace quicr {
         };
         ReasonCode reason_code;
 
-        std::optional<std::string> reason_phrase = std::nullopt;
+        std::optional<std::string> error_reason = std::nullopt;
         std::optional<uint64_t> track_alias = std::nullopt; ///< Set only when ResponseCode is kRetryTrackAlias
 
-        std::optional<uint64_t> largest_group_id = std::nullopt;
-        std::optional<uint64_t> largest_object_id = std::nullopt;
+        std::optional<messages::Location> largest_location = std::nullopt;
     };
 
     /**
@@ -101,20 +101,20 @@ namespace quicr {
         std::optional<uint64_t> GetTrackAlias() const noexcept { return full_track_name_.track_alias; }
 
         /**
-         * @brief Sets the subscribe ID
-         * @details MoQ instance sets the subscribe id based on subscribe track method call. Subscribe
+         * @brief Sets the reqeust ID
+         * @details MoQ instance sets the request id based on subscribe track method call. Request
          *      id is specific to the connection, so it must be set by the moq instance/connection.
          *
-         * @param subscribe_id          62bit subscribe ID
+         * @param request_id          62bit request ID
          */
-        void SetSubscribeId(std::optional<uint64_t> subscribe_id) { subscribe_id_ = subscribe_id; }
+        void SetRequestId(std::optional<uint64_t> request_id) { request_id_ = request_id; }
 
         /**
-         * @brief Get the subscribe ID
+         * @brief Get the request ID
          *
-         * @return nullopt if not subscribed, otherwise the subscribe ID
+         * @return nullopt if not subscribed, otherwise the request ID
          */
-        std::optional<uint64_t> GetSubscribeId() const noexcept { return subscribe_id_; }
+        std::optional<uint64_t> GetRequestId() const noexcept { return request_id_; }
 
         /**
          * @brief Get the full track name
@@ -150,12 +150,12 @@ namespace quicr {
         ConnectionHandle connection_handle_; // QUIC transport connection ID
 
         /**
-         * subscribe_id_ is the primary index/key for subscribe subscribe context/delegate storage.
-         *   It is use as the subscribe_id in MoQ related subscribes.  Subscribe ID will adapt
-         *   to received subscribe IDs, so the value will reflect either the received subscribe ID
+         * request_id_ is the primary index/key for subscribe context/delegate storage.
+         *   It is use as the request_id in MoQ related subscribes.  Request ID will adapt
+         *   to received reqeust IDs, so the value will reflect either the received reqeust ID
          *   or the next one that increments from last received ID.
          */
-        std::optional<uint64_t> subscribe_id_;
+        std::optional<uint64_t> request_id_;
     };
 
 } // namespace moq

include/quicr/detail/ctrl_message_types.h

@@ -14,27 +14,108 @@ namespace quicr::messages {
     quicr::Bytes& operator<<(quicr::Bytes& buffer, std::uint8_t value);
     quicr::BytesSpan operator>>(quicr::BytesSpan buffer, uint8_t& value);
 
+    Bytes& operator<<(Bytes& buffer, std::uint16_t value);
+    BytesSpan operator>>(BytesSpan buffer, std::uint16_t& value);
+
     quicr::Bytes& operator<<(quicr::Bytes& buffer, const quicr::UintVar& value);
 
     using GroupId = uint64_t;
     using ObjectId = uint64_t;
+    // TODO(RichLogan): Remove when ErrorReason -> ReasonPhrase.
+    using ReasonPhrase = Bytes;
 
-    enum struct ParameterType : uint64_t
+    struct ControlMessage
     {
-        kPath = 0x1,
-        kMaxSubscribeId = 0x2, // version specific, unused
-        kEndpointId = 0xF0,    // Endpoint ID, using temp value for now
-        kInvalid = 0xFF,       // used internally.
+        std::uint64_t type{ 0 };
+        Bytes payload{};
     };
+    Bytes& operator<<(Bytes& buffer, const ControlMessage& message);
+    BytesSpan operator>>(BytesSpan buffer, ControlMessage& message);
 
-    struct Parameter
+    struct Location
     {
-        ParameterType type{ 0 };
+        GroupId group{ 0 };
+        ObjectId object{ 0 };
+    };
+    Bytes& operator<<(Bytes& buffer, const Location& location);
+    BytesSpan operator>>(BytesSpan buffer, Location& location);
+
+    /// MoQ Key Value Pair.
+    template<typename T>
+    concept KeyType =
+      std::same_as<T, std::uint64_t> || (std::is_enum_v<T> && std::same_as<std::underlying_type_t<T>, std::uint64_t>);
+    template<KeyType T>
+    struct KeyValuePair
+    {
+        T type;
         Bytes value;
     };
+    template<KeyType T>
+    Bytes& operator<<(Bytes& buffer, const KeyValuePair<T>& param)
+    {
+        const auto type = static_cast<std::uint64_t>(param.type);
+        buffer << UintVar(type);
+        if (type % 2 == 0) {
+            // Even, single varint of value.
+            assert(param.value.size() <= 8);
+            std::uint64_t val = 0;
+            std::memcpy(&val, param.value.data(), std::min(param.value.size(), sizeof(std::uint64_t)));
+            buffer << UintVar(val);
+        } else {
+            // Odd, encode bytes.
+            buffer << UintVar(param.value.size());
+            buffer.insert(buffer.end(), param.value.begin(), param.value.end());
+        }
+        return buffer;
+    }
+    template<KeyType T>
+    BytesSpan operator>>(BytesSpan buffer, KeyValuePair<T>& param)
+    {
+        std::uint64_t type;
+        buffer = buffer >> type;
+        param.type = static_cast<T>(type);
+        if (type % 2 == 0) {
+            // Even, single varint of value.
+            UintVar uvar(buffer);
+            buffer = buffer.subspan(uvar.size());
+            std::uint64_t val(uvar);
+            param.value.resize(uvar.size());
+            std::memcpy(param.value.data(), &val, uvar.size());
+        } else {
+            // Odd, decode bytes.
+            uint64_t size = 0;
+            buffer = buffer >> size;
+            param.value.assign(buffer.begin(), std::next(buffer.begin(), size));
+            buffer = buffer.subspan(size);
+        }
+        return buffer;
+    }
+
+    // Serialization for all uint64_t/enum(uint64_t to varint).
+    template<KeyType T>
+    Bytes& operator<<(Bytes& buffer, const T value)
+    {
+        buffer << UintVar(static_cast<std::uint64_t>(value));
+        return buffer;
+    }
+    template<KeyType T>
+    BytesSpan operator>>(BytesSpan buffer, T& value)
+    {
+        std::uint64_t uvalue;
+        buffer = buffer >> uvalue;
+        value = static_cast<T>(uvalue);
+        return buffer;
+    }
 
-    Bytes& operator<<(Bytes& buffer, ParameterType value);
-    BytesSpan operator>>(BytesSpan buffer, ParameterType& value);
+    enum struct ParameterType : uint64_t
+    {
+        kPath = 0x1,
+        kMaxRequestId = 0x2, // version specific, unused
+        kEndpointId = 0xF1,  // Endpoint ID, using temp value for now
+        kInvalid = 0xFF,     // used internally.
+    };
+
+    using Parameter = KeyValuePair<ParameterType>;
 
     enum struct GroupOrder : uint8_t
     {
@@ -55,9 +136,6 @@ namespace quicr::messages {
         kAbsoluteRange
     };
 
-    Bytes& operator<<(Bytes& buffer, FilterType value);
-    BytesSpan operator>>(BytesSpan buffer, FilterType& value);
-
     enum class TrackStatusCode : uint64_t
     {
         kInProgress = 0x00,
@@ -67,9 +145,6 @@ namespace quicr::messages {
         kUnknown
     };
 
-    Bytes& operator<<(Bytes& buffer, TrackStatusCode value);
-    BytesSpan operator>>(BytesSpan buffer, TrackStatusCode& value);
-
     enum class SubscribeDoneStatusCode : uint64_t
     {
         kInternalError = 0x00,
@@ -81,9 +156,6 @@ namespace quicr::messages {
         kTooFarBehind,
     };
 
-    Bytes& operator<<(Bytes& buffer, SubscribeDoneStatusCode value);
-    BytesSpan operator>>(BytesSpan buffer, SubscribeDoneStatusCode& value);
-
     enum class FetchType : uint8_t
     {
         kStandalone = 0x1,
@@ -104,9 +176,6 @@ namespace quicr::messages {
         kGoAwayTimeout = 0x10,
     };
 
-    Bytes& operator<<(Bytes& buffer, TerminationReason value);
-    BytesSpan operator>>(BytesSpan buffer, TerminationReason& value);
-
     enum class FetchErrorCode : uint8_t
     {
         kInternalError = 0x0,
@@ -129,9 +198,6 @@ namespace quicr::messages {
         kUninterested
     };
 
-    Bytes& operator<<(Bytes& buffer, AnnounceErrorCode value);
-    BytesSpan operator>>(BytesSpan buffer, AnnounceErrorCode& value);
-
     // TODO (Suhas): rename it to StreamMapping
     enum ForwardingPreference : uint8_t
     {
@@ -158,9 +224,6 @@ namespace quicr::messages {
         kTrackNotExist = 0xF0 // Missing in draft
     };
 
-    Bytes& operator<<(Bytes& buffer, SubscribeErrorCode value);
-    BytesSpan operator>>(BytesSpan buffer, SubscribeErrorCode& value);
-
     enum class SubscribeAnnouncesErrorCode : uint64_t
     {
         kInternalError = 0x0,
@@ -169,7 +232,4 @@ namespace quicr::messages {
         kNotSupported,
         kNamespacePrefixUnknown,
     };
-
-    Bytes& operator<<(Bytes& buffer, SubscribeAnnouncesErrorCode value);
-    BytesSpan operator>>(BytesSpan buffer, SubscribeAnnouncesErrorCode& value);
 } // namespace