Allow NEW_GROUP_REQUEST on subscription update (including PUBLISH_OK). Forbid NEW_GROUP_REQUEST if DYNAMIC_GROUPS is known to be false. Part 3 of 3 for NEW_GROUP_REQUEST. PiperOrigin-RevId: 984015957
diff --git a/quiche/quic/moqt/moqt_integration_test.cc b/quiche/quic/moqt/moqt_integration_test.cc index ec75f16..f1b7918 100644 --- a/quiche/quic/moqt/moqt_integration_test.cc +++ b/quiche/quic/moqt/moqt_integration_test.cc
@@ -1415,6 +1415,109 @@ EXPECT_EQ(received_error.error_code, RequestErrorCode::kDoesNotExist); } +TEST_F(MoqtIntegrationTest, SubscribeWithNewGroupRequest) { + EstablishSession(); + MoqtKnownTrackPublisher publisher; + server_->session()->set_publisher(&publisher); + + FullTrackName track_name("test", "data"); + std::shared_ptr<MoqtOutgoingQueue> queue; + queue = std::make_shared<MoqtOutgoingQueue>( + track_name, test_harness_.simulator().GetClock(), [&]() { + queue->AddObject(quiche::QuicheMemSlice::Copy("object 3"), true); + }); + publisher.Add(queue); + + // Publish some objects before having any subscribers. + queue->AddObject(quiche::QuicheMemSlice::Copy("object 1"), /*key=*/true); + queue->AddObject(quiche::QuicheMemSlice::Copy("object 2"), /*key=*/false); + + MessageParameters parameters(MoqtFilterType::kNextGroupStart); + parameters.new_group_request = 0; + client_->session()->Subscribe(track_name, &subscribe_visitor_, parameters); + + std::optional<Location> largest_id; + bool dynamic_groups = false; + EXPECT_CALL(subscribe_visitor_, OnReply) + .WillOnce( + [&](const FullTrackName&, + std::variant<SubscribeOkData, MoqtRequestErrorInfo> response) { + ASSERT_TRUE(std::holds_alternative<SubscribeOkData>(response)); + const auto& ok_data = std::get<SubscribeOkData>(response); + largest_id = ok_data.parameters.largest_object; + dynamic_groups = ok_data.properties.dynamic_groups(); + }); + int received = 0; + // EndOfGroup object not sent. + EXPECT_CALL(subscribe_visitor_, + OnObjectFragment(track_name, + MetadataLocationAndStatus( + Location{1, 0}, MoqtObjectStatus::kNormal), + "object 3", /*offset=*/0)) + .WillOnce([&] { ++received; }); + bool success = test_harness_.RunUntilWithDefaultTimeout( + [&]() { return largest_id.has_value() && received == 1; }); + EXPECT_TRUE(success); + EXPECT_EQ(largest_id, Location(0, 1)); + EXPECT_TRUE(dynamic_groups); +} + +TEST_F(MoqtIntegrationTest, PublishWithNewGroupRequest) { + EstablishSession(); + FullTrackName track_name("test", "data"); + std::shared_ptr<MoqtOutgoingQueue> queue; + queue = std::make_shared<MoqtOutgoingQueue>( + track_name, test_harness_.simulator().GetClock(), [&]() { + queue->AddObject(quiche::QuicheMemSlice::Copy("object 3"), true); + }); + + // Publish some objects before having any subscribers. + queue->AddObject(quiche::QuicheMemSlice::Copy("object 1"), true); + queue->AddObject(quiche::QuicheMemSlice::Copy("object 2"), false); + + std::optional<Location> largest_id; + bool dynamic_groups = false; + client_->session()->callbacks().incoming_publish_callback = + [&](const FullTrackName& name, const MessageParameters& parameters, + const TrackProperties& properties, MoqtResponseCallback callback) { + EXPECT_EQ(name, track_name); + largest_id = parameters.largest_object; + dynamic_groups = properties.dynamic_groups(); + MessageParameters ok_parameters; + ok_parameters.new_group_request = 0; + ok_parameters.subscription_filter = + SubscriptionFilter(MoqtFilterType::kNextGroupStart); + std::move(callback)(ok_parameters); + return &subscribe_visitor_; + }; + EXPECT_CALL(subscribe_visitor_, OnReply).Times(1); + + bool publish_ok_received = false; + ASSERT_TRUE(server_->session()->Publish( + queue, MessageParameters(), queue->properties(), + [&](std::variant<MessageParameters, MoqtRequestErrorInfo> response) { + ASSERT_TRUE(std::holds_alternative<MessageParameters>(response)); + EXPECT_EQ(std::get<MessageParameters>(response).new_group_request, 0u); + publish_ok_received = true; + })); + + int received = 0; + // EndOfGroup object not sent. + EXPECT_CALL(subscribe_visitor_, + OnObjectFragment(track_name, + MetadataLocationAndStatus( + Location{1, 0}, MoqtObjectStatus::kNormal), + "object 3", /*offset=*/0)) + .WillOnce([&] { ++received; }); + + bool success = test_harness_.RunUntilWithDefaultTimeout([&]() { + return publish_ok_received && largest_id.has_value() && received == 1; + }); + EXPECT_TRUE(success); + EXPECT_EQ(largest_id, Location(0, 1)); + EXPECT_TRUE(dynamic_groups); +} + } // namespace } // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_live_publisher.cc b/quiche/quic/moqt/moqt_live_publisher.cc index d48dad3..49630dc 100644 --- a/quiche/quic/moqt/moqt_live_publisher.cc +++ b/quiche/quic/moqt/moqt_live_publisher.cc
@@ -73,8 +73,9 @@ } void LivePublisher::Update(const MessageParameters& parameters) { - // TODO(martinduke): If there are auth tokens, this probably has to go to the - // application. + if (!track_publisher_->UpdateObjectListener(this, parameters).ok()) { + return; + } // TODO(martinduke): If the subscribe window has shrunk, close any streams // that are now outside the window. Also send PUBLISH_DONE if now done. MoqtPriority old_priority =
diff --git a/quiche/quic/moqt/moqt_object_subscriber.h b/quiche/quic/moqt/moqt_object_subscriber.h index 6325d3f..7bc8b87 100644 --- a/quiche/quic/moqt/moqt_object_subscriber.h +++ b/quiche/quic/moqt/moqt_object_subscriber.h
@@ -161,6 +161,8 @@ void SendObjectAck(uint64_t group_id, uint64_t object_id, quic::QuicTimeDelta delta_from_deadline); + bool dynamic_groups() const { return dynamic_groups_; } + private: friend class test::MoqtSessionPeer; friend class test::LiveSubscriberPeer;
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.h b/quiche/quic/moqt/moqt_outgoing_queue.h index c2696eb..991a07d 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue.h +++ b/quiche/quic/moqt/moqt_outgoing_queue.h
@@ -74,18 +74,20 @@ const MessageParameters& parameters) override { listeners_.insert(listener); listener->OnSubscribeAccepted(); - if (properties_.dynamic_groups() && !expect_new_group_ && - parameters.new_group_request.has_value() && - (*parameters.new_group_request == 0 || queue_.empty() || - *parameters.new_group_request > current_group_id_) && - new_group_callback_ != nullptr) { - expect_new_group_ = true; - new_group_callback_(); - } + MaybeRequestNewGroup(parameters); } void RemoveObjectListener(MoqtObjectListener* listener) override { listeners_.erase(listener); } + absl::Status UpdateObjectListener( + MoqtObjectListener* listener, + const MessageParameters& parameters) override { + if (!listeners_.contains(listener)) { + return absl::NotFoundError("Listener not found."); + } + MaybeRequestNewGroup(parameters); + return absl::OkStatus(); + } std::optional<Location> largest_location() const override; std::optional<quic::QuicTimeDelta> expiration() const override { @@ -131,6 +133,18 @@ // The number of recent groups to keep around for newly joined subscribers. static constexpr size_t kMaxQueuedGroups = 3; + void MaybeRequestNewGroup(const MessageParameters& parameters) { + if (!properties_.dynamic_groups() || expect_new_group_ || + !parameters.new_group_request.has_value() || + (*parameters.new_group_request > 0 && !queue_.empty() && + *parameters.new_group_request <= current_group_id_) || + new_group_callback_ == nullptr) { + return; + } + expect_new_group_ = true; + new_group_callback_(); + } + // Fetch task for a fetch from the cache. class FetchTask : public MoqtFetchTask { public:
diff --git a/quiche/quic/moqt/moqt_outgoing_queue_test.cc b/quiche/quic/moqt/moqt_outgoing_queue_test.cc index 0d60cb3..a94bcf5 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue_test.cc +++ b/quiche/quic/moqt/moqt_outgoing_queue_test.cc
@@ -565,5 +565,38 @@ static_cast<uint64_t>(PropertyType::kDynamicGroups))); } +TEST(MoqtOutgoingQueue, UpdateObjectListenerNotFound) { + TestMoqtOutgoingQueue queue; + MockMoqtObjectListener listener; + EXPECT_TRUE( + IsNotFound(queue.UpdateObjectListener(&listener, MessageParameters()))); +} + +TEST(MoqtOutgoingQueue, UpdateObjectListenerNewGroupRequest) { + testing::MockFunction<void()> callback; + TestMoqtOutgoingQueue queue(callback.AsStdFunction()); + { + testing::InSequence seq; + EXPECT_CALL(queue, PublishObject(0, 0, "a")); + EXPECT_CALL(callback, Call()); + EXPECT_CALL(queue, CloseStreamForGroup(0)); + EXPECT_CALL(queue, PublishObject(1, 0, "b")); + } + queue.AddObject(quiche::QuicheMemSlice::Copy("a"), true); + + MockMoqtObjectListener listener; + EXPECT_CALL(listener, OnSubscribeAccepted); + queue.AddObjectListener(&listener, MessageParameters()); + + MessageParameters parameters; + parameters.new_group_request = 0; + QUICHE_EXPECT_OK(queue.UpdateObjectListener(&listener, parameters)); + EXPECT_CALL(listener, + OnNewObjectAvailable(Location(0, 1), testing::Optional(0), _)); + EXPECT_CALL(listener, + OnNewObjectAvailable(Location(1, 0), testing::Optional(0), _)); + queue.AddObject(quiche::QuicheMemSlice::Copy("b"), true); +} + } // namespace } // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_publish_stream.cc b/quiche/quic/moqt/moqt_publish_stream.cc index 51c7c51..38bca29 100644 --- a/quiche/quic/moqt/moqt_publish_stream.cc +++ b/quiche/quic/moqt/moqt_publish_stream.cc
@@ -137,7 +137,8 @@ // There was no existing SUBSCRIBE, so invoke the callback. subscriber_->set_visitor((*incoming_publish_callback_)( message.full_track_name, message.parameters, message.properties, - [weakptr = weak_ptr_factory_.Create()]( + [weakptr = weak_ptr_factory_.Create(), + dynamic_groups = message.properties.dynamic_groups()]( const std::variant<MessageParameters, MoqtRequestErrorInfo> response) { MoqtPublishResponseStream* stream = weakptr.GetIfAvailable(); @@ -146,12 +147,20 @@ } std::visit( absl::Overload{ - [&](const MessageParameters& parameters) { - stream->subscriber_->Update(parameters); - stream->CheckStatus(stream->SendRequestOk(parameters)); + [response_stream = stream, + dg = dynamic_groups](const MessageParameters& parameters) { + MessageParameters update_parameters = parameters; + if (!dg) { + update_parameters.new_group_request.reset(); + } + response_stream->subscriber_->Update(update_parameters); + response_stream->CheckStatus( + response_stream->SendRequestOk(update_parameters)); }, - [&](const MoqtRequestErrorInfo& error_info) { - stream->CheckStatus(stream->SendRequestError(error_info)); + [response_stream = + stream](const MoqtRequestErrorInfo& error_info) { + response_stream->CheckStatus( + response_stream->SendRequestError(error_info)); }}, response); }));
diff --git a/quiche/quic/moqt/moqt_publish_stream_test.cc b/quiche/quic/moqt/moqt_publish_stream_test.cc index b175a5e..aa0c0e6 100644 --- a/quiche/quic/moqt/moqt_publish_stream_test.cc +++ b/quiche/quic/moqt/moqt_publish_stream_test.cc
@@ -303,18 +303,65 @@ EXPECT_EQ(captured_subscriber->track_alias(), kTrackAlias); EXPECT_EQ(captured_subscriber->visitor(), &mock_subscribe_visitor_); - // Verify REQUEST_OK response was sent. - EXPECT_CALL(mock_stream_, - Writev(ControlMessageOfType(MoqtMessageType::kRequestOk), _)) + // Verify REQUEST_OK response was sent with new_group_request filtered out + // because DYNAMIC_GROUPS is absent/false. + MoqtRequestOk expected_ok; + expected_ok.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(2); + EXPECT_CALL(mock_stream_, Writev(SerializedControlMessage(expected_ok), _)) .WillOnce(Return(absl::OkStatus())); - MessageParameters response_parameters; - response_parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(2); + MessageParameters response_parameters = expected_ok.parameters; + response_parameters.new_group_request = 0; std::move(captured_response_callback_)(response_parameters); - // Verify subscriber parameters were updated. + // Verify subscriber parameters were updated and new_group_request was + // filtered out. const MessageParameters& sub_params = LiveSubscriberPeer::parameters(*captured_subscriber); EXPECT_EQ(sub_params.delivery_timeout, response_parameters.delivery_timeout); + EXPECT_EQ(sub_params.new_group_request, std::nullopt); + EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone); +} + +TEST_F(MoqtPublishResponseStreamTest, ReceivePublishWithDynamicGroups) { + EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _)) + .WillOnce( + [](const FullTrackName&, + const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) { + EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply)); + }); + EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) + .WillOnce([this](const FullTrackName&, const MessageParameters&, + const TrackProperties&, MoqtResponseCallback callback) { + captured_response_callback_ = std::move(callback); + return &mock_subscribe_visitor_; + }); + LiveSubscriber* captured_subscriber = nullptr; + EXPECT_CALL(mock_add_callback_, Call(NotNull())) + .WillOnce([&](LiveSubscriber* subscriber) { + captured_subscriber = subscriber; + return true; + }); + MoqtPublish publish = DefaultPublish(); + publish.properties = TrackProperties( + /*delivery_timeout=*/std::nullopt, + /*max_cache_duration=*/std::nullopt, + /*publisher_priority=*/std::nullopt, + /*group_order=*/std::nullopt, + /*dynamic_groups=*/true, + /*immutable_extensions=*/std::nullopt); + QUICHE_EXPECT_OK(stream_->OnControlMessage(publish)); + ASSERT_NE(captured_subscriber, nullptr); + + // Verify REQUEST_OK response preserves new_group_request when DYNAMIC_GROUPS + // is true. + MoqtRequestOk expected_ok; + expected_ok.parameters.new_group_request = 0; + EXPECT_CALL(mock_stream_, Writev(SerializedControlMessage(expected_ok), _)) + .WillOnce(Return(absl::OkStatus())); + std::move(captured_response_callback_)(expected_ok.parameters); + const MessageParameters& sub_params = + LiveSubscriberPeer::parameters(*captured_subscriber); + EXPECT_EQ(sub_params.new_group_request, 0); EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone); }
diff --git a/quiche/quic/moqt/moqt_publisher.h b/quiche/quic/moqt/moqt_publisher.h index 6c3b9a1..bcbd0d2 100644 --- a/quiche/quic/moqt/moqt_publisher.h +++ b/quiche/quic/moqt/moqt_publisher.h
@@ -10,6 +10,7 @@ #include <optional> #include "absl/base/nullability.h" +#include "absl/status/status.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_fetch_task.h" @@ -99,6 +100,8 @@ virtual void AddObjectListener(MoqtObjectListener* listener, const MessageParameters& parameters) = 0; virtual void RemoveObjectListener(MoqtObjectListener* listener) = 0; + virtual absl::Status UpdateObjectListener( + MoqtObjectListener* listener, const MessageParameters& parameters) = 0; // Methods to return various track properties. Returns nullopt if the value is // not yet available.
diff --git a/quiche/quic/moqt/moqt_relay_track_publisher.cc b/quiche/quic/moqt/moqt_relay_track_publisher.cc index abdee6f..0bc1e11 100644 --- a/quiche/quic/moqt/moqt_relay_track_publisher.cc +++ b/quiche/quic/moqt/moqt_relay_track_publisher.cc
@@ -10,9 +10,11 @@ #include <variant> #include "absl/base/attributes.h" +#include "absl/status/status.h" #include "absl/strings/string_view.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/moqt/moqt_error.h" +#include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" #include "quiche/quic/moqt/moqt_names.h" #include "quiche/quic/moqt/moqt_object.h" @@ -392,37 +394,13 @@ DeleteTrack(); return; } + listeners_.insert(listener); pending_new_group_request_ = upstream_parameters.new_group_request; } else { - if (parameters.new_group_request.has_value() && - (!got_response_ || properties_.dynamic_groups()) && - (*parameters.new_group_request == 0 || - *parameters.new_group_request > next_location_.group) && - (!pending_new_group_request_.has_value() || - *pending_new_group_request_ < *parameters.new_group_request)) { - MoqtSessionInterface* session = upstream_.GetIfAvailable(); - if (session != nullptr) { - MessageParameters update_parameters; - if (*parameters.new_group_request == 0 && - next_location_ > Location(0, 0)) { - // The relay has more information than the client, so update - // NEW_GROUP_REQUEST to be more specific and avoid duplicate - // SUBSCRIBE_UPDATEs. - update_parameters.new_group_request = next_location_.group + 1; - } else { - // The client might have other access to a higher group ID, so - // preserve requests that are greater than next_location_. - update_parameters.new_group_request = *parameters.new_group_request; - } - if (session->SubscribeUpdate( - track_, update_parameters, - [](std::variant<MessageParameters, MoqtRequestErrorInfo>) {})) { - pending_new_group_request_ = update_parameters.new_group_request; - }; - } - } + listeners_.insert(listener); + // SUBSCRIBE will be accepted regardless of outcome. + UpdateObjectListener(listener, parameters).IgnoreError(); } - listeners_.insert(listener); if (got_response_) { listener->OnSubscribeAccepted(); } @@ -440,6 +418,46 @@ // No class access below this line! } +absl::Status MoqtRelayTrackPublisher::UpdateObjectListener( + MoqtObjectListener* listener, const MessageParameters& parameters) { + if (is_closing_) { + return absl::InternalError("The track publisher is closing."); + } + if (!listeners_.contains(listener)) { + return absl::NotFoundError("Listener not found."); + } + MoqtSessionInterface* session = upstream_.GetIfAvailable(); + if (session == nullptr) { + return absl::InternalError("The upstream session was closed."); + } + if (parameters.new_group_request.has_value() && + (!got_response_ || properties_.dynamic_groups()) && + (*parameters.new_group_request == 0 || + *parameters.new_group_request > next_location_.group) && + (!pending_new_group_request_.has_value() || + *pending_new_group_request_ < *parameters.new_group_request)) { + MessageParameters update_parameters; + if (*parameters.new_group_request == 0 && next_location_ > Location(0, 0)) { + // The relay has more information than the client, so update + // NEW_GROUP_REQUEST to be more specific and avoid duplicate + // SUBSCRIBE_UPDATEs. + update_parameters.new_group_request = next_location_.group + 1; + } else { + // The client might have other access to a higher group ID, so + // preserve requests that are greater than next_location_. + update_parameters.new_group_request = *parameters.new_group_request; + } + if (session->SubscribeUpdate( + track_, update_parameters, + [](std::variant<MessageParameters, MoqtRequestErrorInfo>) {})) { + pending_new_group_request_ = update_parameters.new_group_request; + return absl::OkStatus(); + } + return absl::InternalError("Could not send SUBSCRIBE_UPDATE upstream."); + } + return absl::OkStatus(); +} + void MoqtRelayTrackPublisher::ForAllObjects( quiche::UnretainedCallback<void(const CachedObject&)> callback) { for (auto& group_it : queue_) {
diff --git a/quiche/quic/moqt/moqt_relay_track_publisher.h b/quiche/quic/moqt/moqt_relay_track_publisher.h index 3390e5a..6a76372 100644 --- a/quiche/quic/moqt/moqt_relay_track_publisher.h +++ b/quiche/quic/moqt/moqt_relay_track_publisher.h
@@ -16,6 +16,7 @@ #include "absl/base/nullability.h" #include "absl/container/btree_map.h" #include "absl/container/flat_hash_set.h" +#include "absl/status/status.h" #include "absl/strings/string_view.h" #include "quiche/quic/core/quic_clock.h" #include "quiche/quic/core/quic_default_clock.h" @@ -98,6 +99,10 @@ void AddObjectListener(MoqtObjectListener* listener, const MessageParameters& parameters) override; void RemoveObjectListener(MoqtObjectListener* listener) override; + // Check if any parameters require an update to pass upstream. + absl::Status UpdateObjectListener( + MoqtObjectListener* listener, + const MessageParameters& parameters) override; std::optional<Location> largest_location() const override; const TrackProperties& properties() const override { return properties_; } std::optional<quic::QuicTimeDelta> expiration() const override;
diff --git a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc index 40bd853..30bf951 100644 --- a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc +++ b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc
@@ -8,10 +8,14 @@ #include <memory> #include <optional> #include <string> +#include <utility> +#include <variant> +#include "absl/status/status.h" #include "absl/strings/string_view.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/moqt/moqt_error.h" +#include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" #include "quiche/quic/moqt/moqt_names.h" #include "quiche/quic/moqt/moqt_object.h" @@ -22,6 +26,7 @@ #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h" #include "quiche/common/platform/api/quiche_expect_bug.h" #include "quiche/common/platform/api/quiche_test.h" +#include "quiche/common/test_tools/quiche_test_utils.h" namespace moqt::test { @@ -939,6 +944,84 @@ publisher_.AddObjectListener(&listener_after_new_group, params_zero); } +TEST_F(MoqtRelayTrackPublisherTest, UpdateObjectListenerNotFound) { + SubscribeAndOk(); + MockMoqtObjectListener listener; + EXPECT_TRUE(IsNotFound( + publisher_.UpdateObjectListener(&listener, MessageParameters()))); +} + +TEST_F(MoqtRelayTrackPublisherTest, UpdateObjectListenerUpstreamClosed) { + SubscribeAndOk(); + session_.reset(); + EXPECT_TRUE(IsInternal( + publisher_.UpdateObjectListener(&listener_, MessageParameters()))); +} + +TEST_F(MoqtRelayTrackPublisherTest, UpdateObjectListenerClosing) { + SubscribeAndOk(); + publisher_.Close(); + EXPECT_TRUE(IsInternal( + publisher_.UpdateObjectListener(&listener_, MessageParameters()))); +} + +TEST_F(MoqtRelayTrackPublisherTest, + UpdateObjectListenerForwardsNewGroupRequest) { + EXPECT_CALL(*session_, Subscribe).WillOnce(testing::Return(true)); + publisher_.AddObjectListener(&listener_, MessageParameters()); + EXPECT_CALL(listener_, OnSubscribeAccepted); + MessageParameters ok_parameters; + ok_parameters.largest_object = kLargestLocation; // Location(3, 2) + ok_parameters.expires = quic::QuicTimeDelta::FromSeconds(30); + TrackProperties properties( + /*delivery_timeout=*/std::nullopt, + /*max_cache_duration=*/std::nullopt, + /*publisher_priority=*/std::nullopt, + /*group_order=*/std::nullopt, + /*dynamic_groups=*/true, + /*immutable_properties=*/std::nullopt); + publisher_.OnReply(kTrackName, SubscribeOkData{ok_parameters, properties}); + + // 1. Update with new_group_request = 4 forwards to session->SubscribeUpdate + // and passes the callback through. + MessageParameters update_params; + update_params.new_group_request = 4; + EXPECT_CALL( + *session_, + SubscribeUpdate( + kTrackName, + testing::Field(&MessageParameters::new_group_request, Optional(4)), + testing::_)) + .WillOnce(testing::Return(true)); + QUICHE_EXPECT_OK(publisher_.UpdateObjectListener(&listener_, update_params)); + + // 2. Subsequent update with same new_group_request while pending does not + // call SubscribeUpdate and immediately invokes the callback with + // MessageParameters(). + EXPECT_CALL(*session_, SubscribeUpdate).Times(0); + QUICHE_EXPECT_OK(publisher_.UpdateObjectListener(&listener_, update_params)); + + // 3. After group 4 arrives, new_group_request = 0 translates to + // next_location_.group + 1 (5), triggers SubscribeUpdate(5), and passes the + // callback through. + ObjectArrives(Location(4, 0), /*subgroup=*/0, MoqtObjectStatus::kNormal, "a"); + MessageParameters params_zero; + params_zero.new_group_request = 0; + MoqtResponseCallback saved_callback3; + EXPECT_CALL( + *session_, + SubscribeUpdate( + kTrackName, + testing::Field(&MessageParameters::new_group_request, Optional(5)), + testing::_)) + .WillOnce([&](const FullTrackName&, const MessageParameters&, + MoqtResponseCallback cb) { + saved_callback3 = std::move(cb); + return true; + }); + QUICHE_EXPECT_OK(publisher_.UpdateObjectListener(&listener_, params_zero)); +} + } // namespace } // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index e2a9a7a..e361f42 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -525,9 +525,16 @@ if (it == subscribe_by_name_.end()) { return false; } + MessageParameters update_parameters = parameters; + if (parameters.new_group_request.has_value() && + !it->second->dynamic_groups() && !it->second->ErrorIsAllowed()) { + // Sending NEW_GROUP_REQUEST if the Track Properties have arrived and do not + // include DYNAMIC_GROUPS. + update_parameters.new_group_request.reset(); + } // sending zero because related request ID is ignored for SUBSCRIBE. return it->second->request_stream() - ->SendRequestUpdate(NextRequestId(), 0, parameters, + ->SendRequestUpdate(NextRequestId(), 0, update_parameters, std::move(response_callback)) .ok(); }
diff --git a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h index 09d2e86..7d5154f 100644 --- a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h +++ b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h
@@ -89,6 +89,10 @@ (override)); MOCK_METHOD(void, RemoveObjectListener, (MoqtObjectListener * listener), (override)); + MOCK_METHOD(absl::Status, UpdateObjectListener, + (MoqtObjectListener * listener, + const MessageParameters& parameters), + (override)); MOCK_METHOD(std::optional<Location>, largest_location, (), (const, override)); MOCK_METHOD(const TrackProperties&, properties, (), (const, override)); MOCK_METHOD(std::optional<quic::QuicTimeDelta>, expiration, (), @@ -130,6 +134,10 @@ void RemoveObjectListener(MoqtObjectListener* listener) override { listeners_.erase(listener); } + absl::Status UpdateObjectListener(MoqtObjectListener*, + const MessageParameters&) override { + return absl::OkStatus(); + } std::optional<Location> largest_location() const override { return largest_location_; }