MoQT SUBSCRIBES_BLOCKED frame.
PiperOrigin-RevId: 722646372
diff --git a/quiche/quic/moqt/moqt_framer.cc b/quiche/quic/moqt/moqt_framer.cc
index e06183a..7e86e0d 100644
--- a/quiche/quic/moqt/moqt_framer.cc
+++ b/quiche/quic/moqt/moqt_framer.cc
@@ -680,6 +680,12 @@
WireStringWithVarInt62Length(message.reason_phrase));
}
+quiche::QuicheBuffer MoqtFramer::SerializeSubscribesBlocked(
+ const MoqtSubscribesBlocked& message) {
+ return SerializeControlMessage(MoqtMessageType::kSubscribesBlocked,
+ WireVarInt62(message.max_subscribe_id));
+}
+
quiche::QuicheBuffer MoqtFramer::SerializeObjectAck(
const MoqtObjectAck& message) {
return SerializeControlMessage(
diff --git a/quiche/quic/moqt/moqt_framer.h b/quiche/quic/moqt/moqt_framer.h
index 6969ab4..9601175 100644
--- a/quiche/quic/moqt/moqt_framer.h
+++ b/quiche/quic/moqt/moqt_framer.h
@@ -70,6 +70,8 @@
quiche::QuicheBuffer SerializeFetchCancel(const MoqtFetchCancel& message);
quiche::QuicheBuffer SerializeFetchOk(const MoqtFetchOk& message);
quiche::QuicheBuffer SerializeFetchError(const MoqtFetchError& message);
+ quiche::QuicheBuffer SerializeSubscribesBlocked(
+ const MoqtSubscribesBlocked& message);
quiche::QuicheBuffer SerializeObjectAck(const MoqtObjectAck& message);
private:
diff --git a/quiche/quic/moqt/moqt_framer_test.cc b/quiche/quic/moqt/moqt_framer_test.cc
index 1a17f77..d55fe4d 100644
--- a/quiche/quic/moqt/moqt_framer_test.cc
+++ b/quiche/quic/moqt/moqt_framer_test.cc
@@ -56,6 +56,7 @@
MoqtMessageType::kFetchCancel,
MoqtMessageType::kFetchOk,
MoqtMessageType::kFetchError,
+ MoqtMessageType::kSubscribesBlocked,
MoqtMessageType::kObjectAck,
MoqtMessageType::kClientSetup,
MoqtMessageType::kServerSetup,
@@ -205,6 +206,10 @@
auto data = std::get<MoqtFetchError>(structured_data);
return framer_.SerializeFetchError(data);
}
+ case moqt::MoqtMessageType::kSubscribesBlocked: {
+ auto data = std::get<MoqtSubscribesBlocked>(structured_data);
+ return framer_.SerializeSubscribesBlocked(data);
+ }
case moqt::MoqtMessageType::kObjectAck: {
auto data = std::get<MoqtObjectAck>(structured_data);
return framer_.SerializeObjectAck(data);
diff --git a/quiche/quic/moqt/moqt_messages.cc b/quiche/quic/moqt/moqt_messages.cc
index 7b535f2..44b37af 100644
--- a/quiche/quic/moqt/moqt_messages.cc
+++ b/quiche/quic/moqt/moqt_messages.cc
@@ -117,6 +117,8 @@
return "FETCH_OK";
case MoqtMessageType::kFetchError:
return "FETCH_ERROR";
+ case MoqtMessageType::kSubscribesBlocked:
+ return "SUBSCRIBES_BLOCKED";
case MoqtMessageType::kObjectAck:
return "OBJECT_ACK";
}
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h
index e022e28..d4722fc 100644
--- a/quiche/quic/moqt/moqt_messages.h
+++ b/quiche/quic/moqt/moqt_messages.h
@@ -102,6 +102,7 @@
kFetchCancel = 0x17,
kFetchOk = 0x18,
kFetchError = 0x19,
+ kSubscribesBlocked = 0x1a,
kClientSetup = 0x40,
kServerSetup = 0x41,
@@ -568,6 +569,10 @@
std::string reason_phrase;
};
+struct QUICHE_EXPORT MoqtSubscribesBlocked {
+ uint64_t max_subscribe_id;
+};
+
// All of the four values in this message are encoded as varints.
// `delta_from_deadline` is encoded as an absolute value, with the lowest bit
// indicating the sign (0 if positive).
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc
index d1ab1f2..66a8ad2 100644
--- a/quiche/quic/moqt/moqt_parser.cc
+++ b/quiche/quic/moqt/moqt_parser.cc
@@ -218,6 +218,9 @@
case MoqtMessageType::kFetchError:
bytes_read = ProcessFetchError(reader);
break;
+ case MoqtMessageType::kSubscribesBlocked:
+ bytes_read = ProcessSubscribesBlocked(reader);
+ break;
case moqt::MoqtMessageType::kObjectAck:
bytes_read = ProcessObjectAck(reader);
break;
@@ -796,6 +799,16 @@
return reader.PreviouslyReadPayload().length();
}
+size_t MoqtControlParser::ProcessSubscribesBlocked(
+ quic::QuicDataReader& reader) {
+ MoqtSubscribesBlocked subscribes_blocked;
+ if (!reader.ReadVarInt62(&subscribes_blocked.max_subscribe_id)) {
+ return 0;
+ }
+ visitor_.OnSubscribesBlockedMessage(subscribes_blocked);
+ return reader.PreviouslyReadPayload().length();
+}
+
size_t MoqtControlParser::ProcessObjectAck(quic::QuicDataReader& reader) {
MoqtObjectAck object_ack;
uint64_t raw_delta;
diff --git a/quiche/quic/moqt/moqt_parser.h b/quiche/quic/moqt/moqt_parser.h
index b05fdcc..591350d 100644
--- a/quiche/quic/moqt/moqt_parser.h
+++ b/quiche/quic/moqt/moqt_parser.h
@@ -62,6 +62,8 @@
virtual void OnFetchCancelMessage(const MoqtFetchCancel& message) = 0;
virtual void OnFetchOkMessage(const MoqtFetchOk& message) = 0;
virtual void OnFetchErrorMessage(const MoqtFetchError& message) = 0;
+ virtual void OnSubscribesBlockedMessage(
+ const MoqtSubscribesBlocked& message) = 0;
virtual void OnObjectAckMessage(const MoqtObjectAck& message) = 0;
virtual void OnParsingError(MoqtError code, absl::string_view reason) = 0;
@@ -135,6 +137,7 @@
size_t ProcessFetchCancel(quic::QuicDataReader& reader);
size_t ProcessFetchOk(quic::QuicDataReader& reader);
size_t ProcessFetchError(quic::QuicDataReader& reader);
+ size_t ProcessSubscribesBlocked(quic::QuicDataReader& reader);
size_t ProcessObjectAck(quic::QuicDataReader& reader);
// If |error| is not provided, assumes kProtocolViolation.
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc
index dc2b0c2..c7f4fdd 100644
--- a/quiche/quic/moqt/moqt_parser_test.cc
+++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -57,6 +57,7 @@
MoqtMessageType::kFetchCancel,
MoqtMessageType::kFetchOk,
MoqtMessageType::kFetchError,
+ MoqtMessageType::kSubscribesBlocked,
MoqtMessageType::kObjectAck,
};
constexpr std::array kDataStreamTypes{
@@ -212,6 +213,10 @@
void OnFetchErrorMessage(const MoqtFetchError& message) override {
OnControlMessage(message);
}
+ void OnSubscribesBlockedMessage(
+ const MoqtSubscribesBlocked& message) override {
+ OnControlMessage(message);
+ }
void OnObjectAckMessage(const MoqtObjectAck& message) override {
OnControlMessage(message);
}
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index 6d5e4d7..91bd0dc 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -524,6 +524,14 @@
std::optional<uint64_t> provided_track_alias) {
// TODO(martinduke): support authorization info
if (next_subscribe_id_ >= peer_max_subscribe_id_) {
+ if (!last_subscribes_blocked_sent_.has_value() ||
+ peer_max_subscribe_id_ > *last_subscribes_blocked_sent_) {
+ MoqtSubscribesBlocked subscribes_blocked;
+ subscribes_blocked.max_subscribe_id = peer_max_subscribe_id_;
+ SendControlMessage(
+ framer_.SerializeSubscribesBlocked(subscribes_blocked));
+ last_subscribes_blocked_sent_ = peer_max_subscribe_id_;
+ }
QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE with ID "
<< next_subscribe_id_
<< " which is greater than the maximum ID "
@@ -1276,6 +1284,11 @@
session_->upstream_by_id_.erase(message.subscribe_id);
}
+void MoqtSession::ControlStream::OnSubscribesBlockedMessage(
+ const MoqtSubscribesBlocked& message) {
+ // TODO(martinduke): Derive logic for granting more subscribes.
+}
+
void MoqtSession::ControlStream::OnParsingError(MoqtError error_code,
absl::string_view reason) {
session_->Error(error_code, absl::StrCat("Parse error: ", reason));
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index be9800c..d185138 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -275,6 +275,8 @@
void OnFetchCancelMessage(const MoqtFetchCancel& message) override {}
void OnFetchOkMessage(const MoqtFetchOk& message) override;
void OnFetchErrorMessage(const MoqtFetchError& message) override;
+ void OnSubscribesBlockedMessage(
+ const MoqtSubscribesBlocked& message) override;
void OnObjectAckMessage(const MoqtObjectAck& message) override {
auto subscription_it =
session_->published_subscriptions_.find(message.subscribe_id);
@@ -632,6 +634,7 @@
uint64_t next_subscribe_id_ = 0;
// The local endpoint can send subscribe IDs less than this value.
uint64_t peer_max_subscribe_id_ = 0;
+ std::optional<uint64_t> last_subscribes_blocked_sent_;
// All open incoming subscriptions, indexed by track name, used to check for
// duplicates.
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 5230311..21cb88f 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -513,11 +513,19 @@
webtransport::test::MockStream mock_stream;
std::unique_ptr<MoqtControlParserVisitor> stream_input =
MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
- EXPECT_CALL(mock_session_, GetStreamById(_)).WillOnce(Return(&mock_stream));
+ EXPECT_CALL(mock_session_, GetStreamById(_))
+ .WillRepeatedly(Return(&mock_stream));
EXPECT_CALL(mock_stream,
Writev(ControlMessageOfType(MoqtMessageType::kSubscribe), _));
EXPECT_TRUE(session_.SubscribeCurrentGroup(FullTrackName("foo", "bar"),
&remote_track_visitor));
+ EXPECT_CALL(
+ mock_stream,
+ Writev(ControlMessageOfType(MoqtMessageType::kSubscribesBlocked), _))
+ .Times(1);
+ EXPECT_FALSE(session_.SubscribeCurrentGroup(FullTrackName("foo2", "bar2"),
+ &remote_track_visitor));
+ // Second time does not send SUBSCRIBES_BLOCKED.
EXPECT_FALSE(session_.SubscribeCurrentGroup(FullTrackName("foo2", "bar2"),
&remote_track_visitor));
}
@@ -566,16 +574,21 @@
MoqtSessionPeer::set_next_subscribe_id(&session_,
kDefaultInitialMaxSubscribeId);
MockSubscribeRemoteTrackVisitor remote_track_visitor;
+ webtransport::test::MockStream mock_stream;
+ std::unique_ptr<MoqtControlParserVisitor> stream_input =
+ MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
+ EXPECT_CALL(mock_session_, GetStreamById(_))
+ .WillRepeatedly(Return(&mock_stream));
+ EXPECT_CALL(
+ mock_stream,
+ Writev(ControlMessageOfType(MoqtMessageType::kSubscribesBlocked), _));
EXPECT_FALSE(session_.SubscribeCurrentGroup(FullTrackName("foo", "bar"),
&remote_track_visitor));
MoqtMaxSubscribeId max_subscribe_id = {
/*max_subscribe_id=*/kDefaultInitialMaxSubscribeId + 1,
};
- webtransport::test::MockStream mock_stream;
- std::unique_ptr<MoqtControlParserVisitor> stream_input =
- MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
stream_input->OnMaxSubscribeIdMessage(max_subscribe_id);
- EXPECT_CALL(mock_session_, GetStreamById(_)).WillOnce(Return(&mock_stream));
+
EXPECT_CALL(mock_stream,
Writev(ControlMessageOfType(MoqtMessageType::kSubscribe), _));
EXPECT_TRUE(session_.SubscribeCurrentGroup(FullTrackName("foo", "bar"),
diff --git a/quiche/quic/moqt/test_tools/moqt_framer_utils.cc b/quiche/quic/moqt/test_tools/moqt_framer_utils.cc
index 8005d4e..e991830 100644
--- a/quiche/quic/moqt/test_tools/moqt_framer_utils.cc
+++ b/quiche/quic/moqt/test_tools/moqt_framer_utils.cc
@@ -100,6 +100,9 @@
MoqtMessageType operator()(const MoqtFetchError&) {
return MoqtMessageType::kFetchError;
}
+ MoqtMessageType operator()(const MoqtSubscribesBlocked&) {
+ return MoqtMessageType::kSubscribesBlocked;
+ }
MoqtMessageType operator()(const MoqtObjectAck&) {
return MoqtMessageType::kObjectAck;
}
@@ -181,6 +184,9 @@
quiche::QuicheBuffer operator()(const MoqtFetchError& message) {
return framer.SerializeFetchError(message);
}
+ quiche::QuicheBuffer operator()(const MoqtSubscribesBlocked& message) {
+ return framer.SerializeSubscribesBlocked(message);
+ }
quiche::QuicheBuffer operator()(const MoqtObjectAck& message) {
return framer.SerializeObjectAck(message);
}
@@ -267,6 +273,9 @@
void OnFetchErrorMessage(const MoqtFetchError& message) {
frames_.push_back(message);
}
+ void OnSubscribesBlockedMessage(const MoqtSubscribesBlocked& message) {
+ frames_.push_back(message);
+ }
void OnObjectAckMessage(const MoqtObjectAck& message) {
frames_.push_back(message);
}
diff --git a/quiche/quic/moqt/test_tools/moqt_framer_utils.h b/quiche/quic/moqt/test_tools/moqt_framer_utils.h
index efbe5ca..944e50b 100644
--- a/quiche/quic/moqt/test_tools/moqt_framer_utils.h
+++ b/quiche/quic/moqt/test_tools/moqt_framer_utils.h
@@ -24,14 +24,16 @@
// TODO: remove MoqtObject from TestMessageBase::MessageStructuredData and merge
// those two types.
-using MoqtGenericFrame = absl::variant<
- MoqtClientSetup, MoqtServerSetup, MoqtSubscribe, MoqtSubscribeOk,
- MoqtSubscribeError, MoqtUnsubscribe, MoqtSubscribeDone, MoqtSubscribeUpdate,
- MoqtAnnounce, MoqtAnnounceOk, MoqtAnnounceError, MoqtAnnounceCancel,
- MoqtTrackStatusRequest, MoqtUnannounce, MoqtTrackStatus, MoqtGoAway,
- MoqtSubscribeAnnounces, MoqtSubscribeAnnouncesOk,
- MoqtSubscribeAnnouncesError, MoqtUnsubscribeAnnounces, MoqtMaxSubscribeId,
- MoqtFetch, MoqtFetchCancel, MoqtFetchOk, MoqtFetchError, MoqtObjectAck>;
+using MoqtGenericFrame =
+ absl::variant<MoqtClientSetup, MoqtServerSetup, MoqtSubscribe,
+ MoqtSubscribeOk, MoqtSubscribeError, MoqtUnsubscribe,
+ MoqtSubscribeDone, MoqtSubscribeUpdate, MoqtAnnounce,
+ MoqtAnnounceOk, MoqtAnnounceError, MoqtAnnounceCancel,
+ MoqtTrackStatusRequest, MoqtUnannounce, MoqtTrackStatus,
+ MoqtGoAway, MoqtSubscribeAnnounces, MoqtSubscribeAnnouncesOk,
+ MoqtSubscribeAnnouncesError, MoqtUnsubscribeAnnounces,
+ MoqtMaxSubscribeId, MoqtFetch, MoqtFetchCancel, MoqtFetchOk,
+ MoqtFetchError, MoqtSubscribesBlocked, MoqtObjectAck>;
MoqtMessageType MessageTypeForGenericMessage(const MoqtGenericFrame& frame);
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h
index 64732c4..1b8ad42 100644
--- a/quiche/quic/moqt/test_tools/moqt_test_message.h
+++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -40,7 +40,7 @@
MoqtTrackStatus, MoqtGoAway, MoqtSubscribeAnnounces,
MoqtSubscribeAnnouncesOk, MoqtSubscribeAnnouncesError,
MoqtUnsubscribeAnnounces, MoqtMaxSubscribeId, MoqtFetch, MoqtFetchCancel,
- MoqtFetchOk, MoqtFetchError, MoqtObjectAck>;
+ MoqtFetchOk, MoqtFetchError, MoqtSubscribesBlocked, MoqtObjectAck>;
// The total actual size of the message.
size_t total_message_size() const { return wire_image_size_; }
@@ -1491,6 +1491,37 @@
};
};
+class QUICHE_NO_EXPORT SubscribesBlockedMessage : public TestMessageBase {
+ public:
+ SubscribesBlockedMessage() : TestMessageBase() {
+ SetWireImage(raw_packet_, sizeof(raw_packet_));
+ }
+ bool EqualFieldValues(MessageStructuredData& values) const override {
+ auto cast = std::get<MoqtSubscribesBlocked>(values);
+ if (cast.max_subscribe_id != subscribes_blocked_.max_subscribe_id) {
+ QUIC_LOG(INFO) << "SUBSCRIBES_BLOCKED max_subscribe_id mismatch";
+ return false;
+ }
+ return true;
+ }
+
+ void ExpandVarints() override { ExpandVarintsImpl("vvv"); }
+
+ MessageStructuredData structured_data() const override {
+ return TestMessageBase::MessageStructuredData(subscribes_blocked_);
+ }
+
+ private:
+ uint8_t raw_packet_[3] = {
+ 0x1a, 0x01,
+ 0x0b, // max_subscribe_id = 11
+ };
+
+ MoqtSubscribesBlocked subscribes_blocked_ = {
+ /*max_subscribe_id=*/11,
+ };
+};
+
class QUICHE_NO_EXPORT ObjectAckMessage : public TestMessageBase {
public:
ObjectAckMessage() : TestMessageBase() {
@@ -1589,6 +1620,8 @@
return std::make_unique<FetchOkMessage>();
case MoqtMessageType::kFetchError:
return std::make_unique<FetchErrorMessage>();
+ case MoqtMessageType::kSubscribesBlocked:
+ return std::make_unique<SubscribesBlockedMessage>();
case MoqtMessageType::kObjectAck:
return std::make_unique<ObjectAckMessage>();
case MoqtMessageType::kClientSetup:
