MoQT Outgoing ANNOUNCE life cycle. PiperOrigin-RevId: 704718628
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h index c7bb493..0f8c26f 100644 --- a/quiche/quic/moqt/moqt_messages.h +++ b/quiche/quic/moqt/moqt_messages.h
@@ -493,8 +493,7 @@ struct QUICHE_EXPORT MoqtAnnounceCancel { FullTrackName track_namespace; - // TODO: What namespace is this error code in? - uint64_t error_code; + MoqtAnnounceErrorCode error_code; std::string reason_phrase; };
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index 7de1149..b8d77d1 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -623,13 +623,11 @@ return 0; } uint64_t error_code; - if (!reader.ReadVarInt62(&error_code)) { + if (!reader.ReadVarInt62(&error_code) || + !reader.ReadStringVarInt62(announce_error.reason_phrase)) { return 0; } announce_error.error_code = static_cast<MoqtAnnounceErrorCode>(error_code); - if (!reader.ReadStringVarInt62(announce_error.reason_phrase)) { - return 0; - } visitor_.OnAnnounceErrorMessage(announce_error); return reader.PreviouslyReadPayload().length(); } @@ -639,10 +637,12 @@ if (!ReadTrackNamespace(reader, announce_cancel.track_namespace)) { return 0; } - if (!reader.ReadVarInt62(&announce_cancel.error_code) || + uint64_t error_code; + if (!reader.ReadVarInt62(&error_code) || !reader.ReadStringVarInt62(announce_cancel.reason_phrase)) { return 0; } + announce_cancel.error_code = static_cast<MoqtAnnounceErrorCode>(error_code); visitor_.OnAnnounceCancelMessage(announce_cancel); return reader.PreviouslyReadPayload().length(); }
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index 81fb1b7..538f081 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -270,8 +270,6 @@ return true; } -// TODO: Create state that allows ANNOUNCE_OK/ERROR on spurious namespaces to -// trigger session errors. void MoqtSession::Announce(FullTrackName track_namespace, MoqtOutgoingAnnounceCallback announce_callback) { if (peer_role_ == MoqtRole::kPublisher) { @@ -281,7 +279,7 @@ "ANNOUNCE cannot be sent to Publisher"}); return; } - if (pending_outgoing_announces_.contains(track_namespace)) { + if (outgoing_announces_.contains(track_namespace)) { std::move(announce_callback)( track_namespace, MoqtAnnounceErrorReason{ @@ -294,7 +292,21 @@ SendControlMessage(framer_.SerializeAnnounce(message)); QUIC_DLOG(INFO) << ENDPOINT << "Sent ANNOUNCE message for " << message.track_namespace; - pending_outgoing_announces_[track_namespace] = std::move(announce_callback); + outgoing_announces_[track_namespace] = std::move(announce_callback); +} + +bool MoqtSession::Unannounce(FullTrackName track_namespace) { + auto it = outgoing_announces_.find(track_namespace); + if (it == outgoing_announces_.end()) { + return false; // Could have been destroyed by ANNOUNCE_CANCEL. + } + MoqtUnannounce message; + message.track_namespace = track_namespace; + SendControlMessage(framer_.SerializeUnannounce(message)); + QUIC_DLOG(INFO) << ENDPOINT << "Sent UNANNOUNCE message for " + << message.track_namespace; + outgoing_announces_.erase(it); + return true; } bool MoqtSession::SubscribeAbsolute(const FullTrackName& name, @@ -982,36 +994,45 @@ SendOrBufferMessage(session_->framer_.SerializeAnnounceOk(ok)); } +// Do not enforce that there is only one of OK or ERROR per ANNOUNCE. Upon +// ERROR, we immediately destroy the state. void MoqtSession::ControlStream::OnAnnounceOkMessage( const MoqtAnnounceOk& message) { - auto it = session_->pending_outgoing_announces_.find(message.track_namespace); - if (it == session_->pending_outgoing_announces_.end()) { - session_->Error(MoqtError::kProtocolViolation, - "Received ANNOUNCE_OK for nonexistent announce"); - return; + auto it = session_->outgoing_announces_.find(message.track_namespace); + if (it == session_->outgoing_announces_.end()) { + return; // State might have been destroyed due to UNANNOUNCE. } std::move(it->second)(message.track_namespace, std::nullopt); - session_->pending_outgoing_announces_.erase(it); } void MoqtSession::ControlStream::OnAnnounceErrorMessage( const MoqtAnnounceError& message) { - auto it = session_->pending_outgoing_announces_.find(message.track_namespace); - if (it == session_->pending_outgoing_announces_.end()) { - session_->Error(MoqtError::kProtocolViolation, - "Received ANNOUNCE_ERROR for nonexistent announce"); - return; + auto it = session_->outgoing_announces_.find(message.track_namespace); + if (it == session_->outgoing_announces_.end()) { + return; // State might have been destroyed due to UNANNOUNCE. } std::move(it->second)( message.track_namespace, MoqtAnnounceErrorReason{message.error_code, std::string(message.reason_phrase)}); - session_->pending_outgoing_announces_.erase(it); + session_->outgoing_announces_.erase(it); } void MoqtSession::ControlStream::OnAnnounceCancelMessage( const MoqtAnnounceCancel& message) { - // TODO: notify the application about this. + // The spec currently says that if a later SUBSCRIBE arrives for this + // namespace, that SHOULD be a session error. I'm hoping that via Issue #557, + // this will go away. Regardless, a SHOULD will not compel the session to keep + // state forever, so there is no support for this requirement. + auto it = session_->outgoing_announces_.find(message.track_namespace); + if (it == session_->outgoing_announces_.end()) { + return; // State might have been destroyed due to UNANNOUNCE. + } + std::move(it->second)( + message.track_namespace, + MoqtAnnounceErrorReason{message.error_code, + std::string(message.reason_phrase)}); + session_->outgoing_announces_.erase(it); } void MoqtSession::ControlStream::OnSubscribeAnnouncesOkMessage(
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index 97eb5e5..4bcba1c 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -42,8 +42,13 @@ using MoqtSessionTerminatedCallback = quiche::SingleUseCallback<void(absl::string_view error_message)>; using MoqtSessionDeletedCallback = quiche::SingleUseCallback<void()>; -// If |error_message| is nullopt, the ANNOUNCE was successful. -using MoqtOutgoingAnnounceCallback = quiche::SingleUseCallback<void( + +// If |error_message| is nullopt, this is triggered by an ANNOUNCE_OK. +// Otherwise, it is triggered by ANNOUNCE_ERROR or ANNOUNCE_CANCEL. For +// ERROR or CANCEL, MoqtSession is deleting all ANNOUNCE state immediately +// after calling this callback. Alternatively, the application can call +// Unannounce() to delete the state. +using MoqtOutgoingAnnounceCallback = quiche::MultiUseCallback<void( FullTrackName track_namespace, std::optional<MoqtAnnounceErrorReason> error)>; using MoqtIncomingAnnounceCallback = @@ -120,6 +125,8 @@ // there is already an unresolved ANNOUNCE for that namespace. void Announce(FullTrackName track_namespace, MoqtOutgoingAnnounceCallback announce_callback); + // Returns true if message was sent, false if there is no ANNOUNCE to cancel. + bool Unannounce(FullTrackName track_namespace); // Returns true if SUBSCRIBE was sent. If there is already a subscription to // the track, the message will still be sent. However, the visitor will be @@ -615,9 +622,11 @@ absl::flat_hash_map<FullTrackName, MoqtPublishingMonitorInterface*> monitoring_interfaces_for_published_tracks_; - // Indexed by track namespace. + // Indexed by track namespace. If the value is not nullptr, no OK or ERROR + // has been received. The entry is deleted after sending UNANNOUNCE or + // receiving ANNOUNCE_CANCEL. absl::flat_hash_map<FullTrackName, MoqtOutgoingAnnounceCallback> - pending_outgoing_announces_; + outgoing_announces_; // The value is nullptr after OK or ERROR is received. The entry is deleted // when sending UNSUBSCRIBE_ANNOUNCES, to make sure the application doesn't // unsubscribe from something that it isn't subscribed to. ANNOUNCEs that
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index c2fb556..2d91180 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -231,7 +231,7 @@ stream_input->OnSubscribeMessage(request); } -TEST_F(MoqtSessionTest, AnnounceWithOk) { +TEST_F(MoqtSessionTest, AnnounceWithOkAndCancel) { testing::MockFunction<void( FullTrackName track_namespace, std::optional<MoqtAnnounceErrorReason> error_message)> @@ -255,6 +255,56 @@ EXPECT_FALSE(error.has_value()); }); stream_input->OnAnnounceOkMessage(ok); + + MoqtAnnounceCancel cancel = { + /*track_namespace=*/FullTrackName{"foo"}, + /*error_code=*/MoqtAnnounceErrorCode::kInternalError, + /*reason_phrase=*/"Test error", + }; + EXPECT_CALL(announce_resolved_callback, Call(_, _)) + .WillOnce([&](FullTrackName track_namespace, + std::optional<MoqtAnnounceErrorReason> error) { + EXPECT_EQ(track_namespace, FullTrackName{"foo"}); + ASSERT_TRUE(error.has_value()); + EXPECT_EQ(error->error_code, MoqtAnnounceErrorCode::kInternalError); + EXPECT_EQ(error->reason_phrase, "Test error"); + }); + stream_input->OnAnnounceCancelMessage(cancel); + // State is gone. + EXPECT_FALSE(session_.Unannounce(FullTrackName{"foo"})); +} + +TEST_F(MoqtSessionTest, AnnounceWithOkAndUnannounce) { + testing::MockFunction<void( + FullTrackName track_namespace, + std::optional<MoqtAnnounceErrorReason> error_message)> + announce_resolved_callback; + 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_stream, + Writev(ControlMessageOfType(MoqtMessageType::kAnnounce), _)); + session_.Announce(FullTrackName{"foo"}, + announce_resolved_callback.AsStdFunction()); + + MoqtAnnounceOk ok = { + /*track_namespace=*/FullTrackName{"foo"}, + }; + EXPECT_CALL(announce_resolved_callback, Call(_, _)) + .WillOnce([&](FullTrackName track_namespace, + std::optional<MoqtAnnounceErrorReason> error) { + EXPECT_EQ(track_namespace, FullTrackName{"foo"}); + EXPECT_FALSE(error.has_value()); + }); + stream_input->OnAnnounceOkMessage(ok); + + EXPECT_CALL(mock_session_, GetStreamById(_)).WillOnce(Return(&mock_stream)); + EXPECT_CALL(mock_stream, + Writev(ControlMessageOfType(MoqtMessageType::kUnannounce), _)); + session_.Unannounce(FullTrackName{"foo"}); + // State is gone. + EXPECT_FALSE(session_.Unannounce(FullTrackName{"foo"})); } TEST_F(MoqtSessionTest, AnnounceWithError) { @@ -285,6 +335,8 @@ EXPECT_EQ(error->reason_phrase, "Test error"); }); stream_input->OnAnnounceErrorMessage(error); + // State is gone. + EXPECT_FALSE(session_.Unannounce(FullTrackName{"foo"})); } TEST_F(MoqtSessionTest, SubscribeForPast) {
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h index 5015bf7..41652e2 100644 --- a/quiche/quic/moqt/test_tools/moqt_test_message.h +++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -963,7 +963,7 @@ MoqtAnnounceCancel announce_cancel_ = { /*track_namespace=*/FullTrackName{"foo"}, - /*error_code=*/1, + /*error_code=*/MoqtAnnounceErrorCode::kAnnounceNotSupported, /*reason_phrase=*/"bar", }; };