Update messages that will go in SUBSCRIBE_NAMESPACE streams up to draft-16.

REQUEST_OK, REQUEST_ERROR, SUBSCRIBE_NAMESPACE, NAMESPACE, NAMESPACE_DONE.

Session-layer processing of SUBSCRIBE_NAMESPACE does not change in this CL.

PiperOrigin-RevId: 864976974

Author: martinduke

quiche/quic/moqt/moqt_bidi_stream.h

@@ -17,6 +17,7 @@
 #include "absl/status/status.h"
 #include "absl/strings/string_view.h"
 #include "absl/types/span.h"
+#include "quiche/quic/core/quic_time.h"
 #include "quiche/quic/moqt/moqt_error.h"
 #include "quiche/quic/moqt/moqt_framer.h"
 #include "quiche/quic/moqt/moqt_key_value_pair.h"
@@ -98,6 +99,13 @@ class MoqtBidiStreamBase : public MoqtControlParserVisitor,
       const MoqtPublishNamespaceDone& message) override {
     OnParsingError(wrong_message_error_, wrong_message_reason_);
   }
+  virtual void OnNamespaceMessage(const MoqtNamespace& message) override {
+    OnParsingError(wrong_message_error_, wrong_message_reason_);
+  }
+  virtual void OnNamespaceDoneMessage(
+      const MoqtNamespaceDone& message) override {
+    OnParsingError(wrong_message_error_, wrong_message_reason_);
+  }
   virtual void OnPublishNamespaceCancelMessage(
       const MoqtPublishNamespaceCancel& message) override {
     OnParsingError(wrong_message_error_, wrong_message_reason_);
@@ -182,21 +190,27 @@ class MoqtBidiStreamBase : public MoqtControlParserVisitor,
     }
     SendMessage(std::move(message), fin);
   }
-  void SendRequestOk(uint64_t request_id,
-                     const VersionSpecificParameters& parameters,
+  void SendRequestOk(uint64_t request_id, const MessageParameters& parameters,
                      bool fin = false) {
     SendOrBufferMessage(
         framer_->SerializeRequestOk(MoqtRequestOk{request_id, parameters}),
         fin);
   }
   void SendRequestError(uint64_t request_id, RequestErrorCode error_code,
+                        std::optional<quic::QuicTimeDelta> retry_interval,
                         absl::string_view reason_phrase, bool fin = false) {
     MoqtRequestError request_error;
     request_error.request_id = request_id;
     request_error.error_code = error_code;
+    request_error.retry_interval = retry_interval;
     request_error.reason_phrase = reason_phrase;
     SendOrBufferMessage(framer_->SerializeRequestError(request_error), fin);
   }
+  void SendRequestError(uint64_t request_id, MoqtRequestErrorInfo info,
+                        bool fin = false) {
+    SendRequestError(request_id, info.error_code, info.retry_interval,
+                     info.reason_phrase, fin);
+  }
   void Fin() {
     fin_queued_ = true;
     if (pending_messages_.empty()) {

quiche/quic/moqt/moqt_bidi_stream_test.cc

@@ -214,8 +214,9 @@ TEST_F(MoqtBidiStreamTest, MessageBufferedThenSent) {
   stream_->set_stream(&mock_stream_);
   EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(false));
   EXPECT_CALL(mock_stream_, Writev).Times(0);
-  stream_->SendRequestOk(0, VersionSpecificParameters());
-  stream_->SendRequestError(2, RequestErrorCode::kUnauthorized, "bad request");
+  stream_->SendRequestOk(0, MessageParameters());
+  stream_->SendRequestError(2, RequestErrorCode::kUnauthorized, std::nullopt,
+                            "bad request");
   stream_->Fin();
   {
     testing::InSequence seq;

quiche/quic/moqt/moqt_error.h

@@ -6,10 +6,12 @@
 #define QUICHE_QUIC_MOQT_MOQT_ERROR_H_
 
 #include <cstdint>
+#include <optional>
 #include <string>
 
 #include "absl/status/status.h"
 #include "absl/strings/string_view.h"
+#include "quiche/quic/core/quic_time.h"
 #include "quiche/common/platform/api/quiche_export.h"
 #include "quiche/web_transport/web_transport.h"
 
@@ -70,8 +72,9 @@ enum class QUICHE_EXPORT RequestErrorCode : uint64_t {
   kExpiredAuthToken = 0x12,
 };
 
-struct MoqtErrorPair {
+struct MoqtRequestErrorInfo {
   RequestErrorCode error_code;
+  std::optional<quic::QuicTimeDelta> retry_interval;
   std::string reason_phrase;
 };
 

quiche/quic/moqt/moqt_fetch_task.h

@@ -5,11 +5,13 @@
 #ifndef QUICHE_QUIC_MOQT_MOQT_FETCH_TASK_H_
 #define QUICHE_QUIC_MOQT_MOQT_FETCH_TASK_H_
 
+#include <optional>
 #include <string>
 #include <utility>
 #include <variant>
 
 #include "absl/status/status.h"
+#include "quiche/quic/moqt/moqt_error.h"
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_object.h"
 #include "quiche/common/quiche_callbacks.h"
@@ -76,7 +78,7 @@ class MoqtFailedFetch : public MoqtFetchTask {
       ObjectsAvailableCallback /*callback*/) override {}
   void SetFetchResponseCallback(FetchResponseCallback callback) override {
     MoqtRequestError error{/*request_id=*/0, StatusToRequestErrorCode(status_),
-                           std::string(status_.message())};
+                           std::nullopt, std::string(status_.message())};
     std::move(callback)(error);
   }
 

quiche/quic/moqt/moqt_framer.cc

@@ -511,14 +511,9 @@ quiche::QuicheBuffer MoqtFramer::SerializeServerSetup(
 
 quiche::QuicheBuffer MoqtFramer::SerializeRequestOk(
     const MoqtRequestOk& message) {
-  KeyValuePairList parameters;
-  if (!FillAndValidateVersionSpecificParameters(
-          MoqtMessageType::kRequestOk, message.parameters, parameters)) {
-    return quiche::QuicheBuffer();
-  };
-  return SerializeControlMessage(MoqtMessageType::kRequestOk,
-                                 WireVarInt62(message.request_id),
-                                 WireKeyValuePairList(parameters));
+  return SerializeControlMessage(
+      MoqtMessageType::kRequestOk, WireVarInt62(message.request_id),
+      WireKeyValuePairList(message.parameters.ToKeyValuePairList()));
 }
 
 quiche::QuicheBuffer MoqtFramer::SerializeSubscribe(
@@ -548,6 +543,9 @@ quiche::QuicheBuffer MoqtFramer::SerializeRequestError(
   return SerializeControlMessage(
       MoqtMessageType::kRequestError, WireVarInt62(message.request_id),
       WireVarInt62(message.error_code),
+      WireVarInt62(message.retry_interval.has_value()
+                       ? message.retry_interval->ToMilliseconds() + 1
+                       : 0),
       WireStringWithVarInt62Length(message.reason_phrase));
 }
 
@@ -600,6 +598,20 @@ quiche::QuicheBuffer MoqtFramer::SerializePublishNamespaceDone(
                                  WireTrackNamespace(message.track_namespace));
 }
 
+quiche::QuicheBuffer MoqtFramer::SerializeNamespace(
+    const MoqtNamespace& message) {
+  return SerializeControlMessage(
+      MoqtMessageType::kNamespace,
+      WireTrackNamespace(message.track_namespace_suffix));
+}
+
+quiche::QuicheBuffer MoqtFramer::SerializeNamespaceDone(
+    const MoqtNamespaceDone& message) {
+  return SerializeControlMessage(
+      MoqtMessageType::kNamespaceDone,
+      WireTrackNamespace(message.track_namespace_suffix));
+}
+
 quiche::QuicheBuffer MoqtFramer::SerializePublishNamespaceCancel(
     const MoqtPublishNamespaceCancel& message) {
   return SerializeControlMessage(
@@ -623,16 +635,11 @@ quiche::QuicheBuffer MoqtFramer::SerializeGoAway(const MoqtGoAway& message) {
 
 quiche::QuicheBuffer MoqtFramer::SerializeSubscribeNamespace(
     const MoqtSubscribeNamespace& message) {
-  KeyValuePairList parameters;
-  if (!FillAndValidateVersionSpecificParameters(
-          MoqtMessageType::kSubscribeNamespace, message.parameters,
-          parameters)) {
-    return quiche::QuicheBuffer();
-  };
-  return SerializeControlMessage(MoqtMessageType::kSubscribeNamespace,
-                                 WireVarInt62(message.request_id),
-                                 WireTrackNamespace(message.track_namespace),
-                                 WireKeyValuePairList(parameters));
+  return SerializeControlMessage(
+      MoqtMessageType::kSubscribeNamespace, WireVarInt62(message.request_id),
+      WireTrackNamespace(message.track_namespace_prefix),
+      WireVarInt62(message.subscribe_options),
+      WireKeyValuePairList(message.parameters.ToKeyValuePairList()));
 }
 
 quiche::QuicheBuffer MoqtFramer::SerializeUnsubscribeNamespace(

quiche/quic/moqt/moqt_framer.h

@@ -61,6 +61,8 @@ class QUICHE_EXPORT MoqtFramer {
       const MoqtPublishNamespace& message);
   quiche::QuicheBuffer SerializePublishNamespaceDone(
       const MoqtPublishNamespaceDone& message);
+  quiche::QuicheBuffer SerializeNamespace(const MoqtNamespace& message);
+  quiche::QuicheBuffer SerializeNamespaceDone(const MoqtNamespaceDone& message);
   quiche::QuicheBuffer SerializePublishNamespaceCancel(
       const MoqtPublishNamespaceCancel& message);
   quiche::QuicheBuffer SerializeTrackStatus(const MoqtTrackStatus& message);

quiche/quic/moqt/moqt_framer_test.cc

@@ -44,6 +44,8 @@ std::vector<MoqtFramerTestParams> GetMoqtFramerTestParams() {
       MoqtMessageType::kPublishDone,
       MoqtMessageType::kPublishNamespace,
       MoqtMessageType::kPublishNamespaceDone,
+      MoqtMessageType::kNamespace,
+      MoqtMessageType::kNamespaceDone,
       MoqtMessageType::kPublishNamespaceCancel,
       MoqtMessageType::kTrackStatus,
       MoqtMessageType::kGoAway,
@@ -134,7 +136,6 @@ class MoqtFramerTest
         auto data = std::get<MoqtSubscribeOk>(structured_data);
         return framer_.SerializeSubscribeOk(data);
       }
-
       case MoqtMessageType::kUnsubscribe: {
         auto data = std::get<MoqtUnsubscribe>(structured_data);
         return framer_.SerializeUnsubscribe(data);
@@ -151,6 +152,14 @@ class MoqtFramerTest
         auto data = std::get<MoqtPublishNamespaceDone>(structured_data);
         return framer_.SerializePublishNamespaceDone(data);
       }
+      case MoqtMessageType::kNamespace: {
+        auto data = std::get<MoqtNamespace>(structured_data);
+        return framer_.SerializeNamespace(data);
+      }
+      case MoqtMessageType::kNamespaceDone: {
+        auto data = std::get<MoqtNamespaceDone>(structured_data);
+        return framer_.SerializeNamespaceDone(data);
+      }
       case moqt::MoqtMessageType::kPublishNamespaceCancel: {
         auto data = std::get<MoqtPublishNamespaceCancel>(structured_data);
         return framer_.SerializePublishNamespaceCancel(data);