Mass renaming of Moqt classes SubscriptionPublisher -> LivePublisher SubscribeRemoteTrack -> LiveSubscriber RemoteTrack -> ObjectSubscriber MoqtNamespaceSubscriberStream -> MoqtSubscribeNamespaceRequestStream MoqtNamespacePublisherStream -> MoqtSubscribeNamespaceResponseStream MoqtPublishPublisherStream -> MoqtPublishRequestStream MoqtPublishSubscriberStream -> MoqtPublishResponseStream PiperOrigin-RevId: 949620450
diff --git a/build/source_list.bzl b/build/source_list.bzl index b1fa635..537de81 100644 --- a/build/source_list.bzl +++ b/build/source_list.bzl
@@ -1589,10 +1589,12 @@ "quic/moqt/moqt_framer.h", "quic/moqt/moqt_key_value_pair.h", "quic/moqt/moqt_known_track_publisher.h", + "quic/moqt/moqt_live_publisher.h", "quic/moqt/moqt_messages.h", "quic/moqt/moqt_names.h", "quic/moqt/moqt_namespace_stream.h", "quic/moqt/moqt_object.h", + "quic/moqt/moqt_object_subscriber.h", "quic/moqt/moqt_outgoing_queue.h", "quic/moqt/moqt_outstanding_objects.h", "quic/moqt/moqt_parser.h", @@ -1608,9 +1610,7 @@ "quic/moqt/moqt_session_interface.h", "quic/moqt/moqt_stream_map.h", "quic/moqt/moqt_subscribe_stream.h", - "quic/moqt/moqt_subscription.h", "quic/moqt/moqt_trace_recorder.h", - "quic/moqt/moqt_track.h", "quic/moqt/moqt_types.h", "quic/moqt/moqt_uni_stream.h", "quic/moqt/relay_namespace_tree.h", @@ -1629,10 +1629,12 @@ "quic/moqt/moqt_framer.cc", "quic/moqt/moqt_key_value_pair.cc", "quic/moqt/moqt_known_track_publisher.cc", + "quic/moqt/moqt_live_publisher.cc", "quic/moqt/moqt_messages.cc", "quic/moqt/moqt_names.cc", "quic/moqt/moqt_namespace_stream.cc", "quic/moqt/moqt_object.cc", + "quic/moqt/moqt_object_subscriber.cc", "quic/moqt/moqt_outgoing_queue.cc", "quic/moqt/moqt_outstanding_objects.cc", "quic/moqt/moqt_parser.cc", @@ -1645,9 +1647,7 @@ "quic/moqt/moqt_session.cc", "quic/moqt/moqt_stream_map.cc", "quic/moqt/moqt_subscribe_stream.cc", - "quic/moqt/moqt_subscription.cc", "quic/moqt/moqt_trace_recorder.cc", - "quic/moqt/moqt_track.cc", "quic/moqt/moqt_uni_stream.cc", "quic/moqt/relay_namespace_tree.cc", "quic/moqt/tools/chat_client.cc", @@ -1665,9 +1665,11 @@ "quic/moqt/moqt_framer_test.cc", "quic/moqt/moqt_integration_test.cc", "quic/moqt/moqt_key_value_pair_test.cc", + "quic/moqt/moqt_live_publisher_test.cc", "quic/moqt/moqt_messages_test.cc", "quic/moqt/moqt_names_test.cc", "quic/moqt/moqt_namespace_stream_test.cc", + "quic/moqt/moqt_object_subscriber_test.cc", "quic/moqt/moqt_object_test.cc", "quic/moqt/moqt_outgoing_queue_test.cc", "quic/moqt/moqt_outstanding_objects_test.cc", @@ -1681,8 +1683,6 @@ "quic/moqt/moqt_session_test.cc", "quic/moqt/moqt_stream_map_test.cc", "quic/moqt/moqt_subscribe_stream_test.cc", - "quic/moqt/moqt_subscription_test.cc", - "quic/moqt/moqt_track_test.cc", "quic/moqt/moqt_uni_stream_test.cc", "quic/moqt/relay_namespace_tree_test.cc", "quic/moqt/session_namespace_tree_test.cc",
diff --git a/build/source_list.gni b/build/source_list.gni index 43addc5..cfa9727 100644 --- a/build/source_list.gni +++ b/build/source_list.gni
@@ -1593,10 +1593,12 @@ "src/quiche/quic/moqt/moqt_framer.h", "src/quiche/quic/moqt/moqt_key_value_pair.h", "src/quiche/quic/moqt/moqt_known_track_publisher.h", + "src/quiche/quic/moqt/moqt_live_publisher.h", "src/quiche/quic/moqt/moqt_messages.h", "src/quiche/quic/moqt/moqt_names.h", "src/quiche/quic/moqt/moqt_namespace_stream.h", "src/quiche/quic/moqt/moqt_object.h", + "src/quiche/quic/moqt/moqt_object_subscriber.h", "src/quiche/quic/moqt/moqt_outgoing_queue.h", "src/quiche/quic/moqt/moqt_outstanding_objects.h", "src/quiche/quic/moqt/moqt_parser.h", @@ -1612,9 +1614,7 @@ "src/quiche/quic/moqt/moqt_session_interface.h", "src/quiche/quic/moqt/moqt_stream_map.h", "src/quiche/quic/moqt/moqt_subscribe_stream.h", - "src/quiche/quic/moqt/moqt_subscription.h", "src/quiche/quic/moqt/moqt_trace_recorder.h", - "src/quiche/quic/moqt/moqt_track.h", "src/quiche/quic/moqt/moqt_types.h", "src/quiche/quic/moqt/moqt_uni_stream.h", "src/quiche/quic/moqt/relay_namespace_tree.h", @@ -1633,10 +1633,12 @@ "src/quiche/quic/moqt/moqt_framer.cc", "src/quiche/quic/moqt/moqt_key_value_pair.cc", "src/quiche/quic/moqt/moqt_known_track_publisher.cc", + "src/quiche/quic/moqt/moqt_live_publisher.cc", "src/quiche/quic/moqt/moqt_messages.cc", "src/quiche/quic/moqt/moqt_names.cc", "src/quiche/quic/moqt/moqt_namespace_stream.cc", "src/quiche/quic/moqt/moqt_object.cc", + "src/quiche/quic/moqt/moqt_object_subscriber.cc", "src/quiche/quic/moqt/moqt_outgoing_queue.cc", "src/quiche/quic/moqt/moqt_outstanding_objects.cc", "src/quiche/quic/moqt/moqt_parser.cc", @@ -1649,9 +1651,7 @@ "src/quiche/quic/moqt/moqt_session.cc", "src/quiche/quic/moqt/moqt_stream_map.cc", "src/quiche/quic/moqt/moqt_subscribe_stream.cc", - "src/quiche/quic/moqt/moqt_subscription.cc", "src/quiche/quic/moqt/moqt_trace_recorder.cc", - "src/quiche/quic/moqt/moqt_track.cc", "src/quiche/quic/moqt/moqt_uni_stream.cc", "src/quiche/quic/moqt/relay_namespace_tree.cc", "src/quiche/quic/moqt/tools/chat_client.cc", @@ -1670,9 +1670,11 @@ "src/quiche/quic/moqt/moqt_framer_test.cc", "src/quiche/quic/moqt/moqt_integration_test.cc", "src/quiche/quic/moqt/moqt_key_value_pair_test.cc", + "src/quiche/quic/moqt/moqt_live_publisher_test.cc", "src/quiche/quic/moqt/moqt_messages_test.cc", "src/quiche/quic/moqt/moqt_names_test.cc", "src/quiche/quic/moqt/moqt_namespace_stream_test.cc", + "src/quiche/quic/moqt/moqt_object_subscriber_test.cc", "src/quiche/quic/moqt/moqt_object_test.cc", "src/quiche/quic/moqt/moqt_outgoing_queue_test.cc", "src/quiche/quic/moqt/moqt_outstanding_objects_test.cc", @@ -1686,8 +1688,6 @@ "src/quiche/quic/moqt/moqt_session_test.cc", "src/quiche/quic/moqt/moqt_stream_map_test.cc", "src/quiche/quic/moqt/moqt_subscribe_stream_test.cc", - "src/quiche/quic/moqt/moqt_subscription_test.cc", - "src/quiche/quic/moqt/moqt_track_test.cc", "src/quiche/quic/moqt/moqt_uni_stream_test.cc", "src/quiche/quic/moqt/relay_namespace_tree_test.cc", "src/quiche/quic/moqt/session_namespace_tree_test.cc",
diff --git a/build/source_list.json b/build/source_list.json index ea8c649..515e539 100644 --- a/build/source_list.json +++ b/build/source_list.json
@@ -1592,10 +1592,12 @@ "quiche/quic/moqt/moqt_framer.h", "quiche/quic/moqt/moqt_key_value_pair.h", "quiche/quic/moqt/moqt_known_track_publisher.h", + "quiche/quic/moqt/moqt_live_publisher.h", "quiche/quic/moqt/moqt_messages.h", "quiche/quic/moqt/moqt_names.h", "quiche/quic/moqt/moqt_namespace_stream.h", "quiche/quic/moqt/moqt_object.h", + "quiche/quic/moqt/moqt_object_subscriber.h", "quiche/quic/moqt/moqt_outgoing_queue.h", "quiche/quic/moqt/moqt_outstanding_objects.h", "quiche/quic/moqt/moqt_parser.h", @@ -1611,9 +1613,7 @@ "quiche/quic/moqt/moqt_session_interface.h", "quiche/quic/moqt/moqt_stream_map.h", "quiche/quic/moqt/moqt_subscribe_stream.h", - "quiche/quic/moqt/moqt_subscription.h", "quiche/quic/moqt/moqt_trace_recorder.h", - "quiche/quic/moqt/moqt_track.h", "quiche/quic/moqt/moqt_types.h", "quiche/quic/moqt/moqt_uni_stream.h", "quiche/quic/moqt/relay_namespace_tree.h", @@ -1632,10 +1632,12 @@ "quiche/quic/moqt/moqt_framer.cc", "quiche/quic/moqt/moqt_key_value_pair.cc", "quiche/quic/moqt/moqt_known_track_publisher.cc", + "quiche/quic/moqt/moqt_live_publisher.cc", "quiche/quic/moqt/moqt_messages.cc", "quiche/quic/moqt/moqt_names.cc", "quiche/quic/moqt/moqt_namespace_stream.cc", "quiche/quic/moqt/moqt_object.cc", + "quiche/quic/moqt/moqt_object_subscriber.cc", "quiche/quic/moqt/moqt_outgoing_queue.cc", "quiche/quic/moqt/moqt_outstanding_objects.cc", "quiche/quic/moqt/moqt_parser.cc", @@ -1648,9 +1650,7 @@ "quiche/quic/moqt/moqt_session.cc", "quiche/quic/moqt/moqt_stream_map.cc", "quiche/quic/moqt/moqt_subscribe_stream.cc", - "quiche/quic/moqt/moqt_subscription.cc", "quiche/quic/moqt/moqt_trace_recorder.cc", - "quiche/quic/moqt/moqt_track.cc", "quiche/quic/moqt/moqt_uni_stream.cc", "quiche/quic/moqt/relay_namespace_tree.cc", "quiche/quic/moqt/tools/chat_client.cc", @@ -1669,9 +1669,11 @@ "quiche/quic/moqt/moqt_framer_test.cc", "quiche/quic/moqt/moqt_integration_test.cc", "quiche/quic/moqt/moqt_key_value_pair_test.cc", + "quiche/quic/moqt/moqt_live_publisher_test.cc", "quiche/quic/moqt/moqt_messages_test.cc", "quiche/quic/moqt/moqt_names_test.cc", "quiche/quic/moqt/moqt_namespace_stream_test.cc", + "quiche/quic/moqt/moqt_object_subscriber_test.cc", "quiche/quic/moqt/moqt_object_test.cc", "quiche/quic/moqt/moqt_outgoing_queue_test.cc", "quiche/quic/moqt/moqt_outstanding_objects_test.cc", @@ -1685,8 +1687,6 @@ "quiche/quic/moqt/moqt_session_test.cc", "quiche/quic/moqt/moqt_stream_map_test.cc", "quiche/quic/moqt/moqt_subscribe_stream_test.cc", - "quiche/quic/moqt/moqt_subscription_test.cc", - "quiche/quic/moqt/moqt_track_test.cc", "quiche/quic/moqt/moqt_uni_stream_test.cc", "quiche/quic/moqt/relay_namespace_tree_test.cc", "quiche/quic/moqt/session_namespace_tree_test.cc",
diff --git a/quiche/quic/moqt/moqt_integration_test.cc b/quiche/quic/moqt/moqt_integration_test.cc index 89a2d22..7671934 100644 --- a/quiche/quic/moqt/moqt_integration_test.cc +++ b/quiche/quic/moqt/moqt_integration_test.cc
@@ -96,7 +96,7 @@ // Client subscribes to the latest object in |track_name|. void SubscribeLatestObject(FullTrackName track_name, - MockSubscribeRemoteTrackVisitor* visitor) { + MockLiveSubscriberVisitor* visitor) { bool received_ok = false; EXPECT_CALL(*visitor, OnReply) .WillOnce( @@ -117,7 +117,7 @@ MockSessionCallbacks client_callbacks_; MockSessionCallbacks server_callbacks_; - MockSubscribeRemoteTrackVisitor subscribe_visitor_; + MockLiveSubscriberVisitor subscribe_visitor_; testing::MockFunction<void(TrackNamespace track_namespace, std::optional<MoqtRequestErrorInfo> error_message)> outgoing_publish_namespace_callback_;
diff --git a/quiche/quic/moqt/moqt_subscription.cc b/quiche/quic/moqt/moqt_live_publisher.cc similarity index 91% rename from quiche/quic/moqt/moqt_subscription.cc rename to quiche/quic/moqt/moqt_live_publisher.cc index fa83093..49f89e1 100644 --- a/quiche/quic/moqt/moqt_subscription.cc +++ b/quiche/quic/moqt/moqt_live_publisher.cc
@@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. -#include "quiche/quic/moqt/moqt_subscription.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include <algorithm> #include <cstdint> @@ -35,7 +35,7 @@ namespace moqt { -SubscriptionPublisher::SubscriptionPublisher( +LivePublisher::LivePublisher( MoqtFramer framer, std::shared_ptr<MoqtTrackPublisher> track_publisher, MoqtBidiStreamBase* absl_nonnull bidi_stream, uint64_t request_id, uint64_t track_alias, const MessageParameters& parameters, @@ -63,13 +63,13 @@ // TODO(martinduke): Handle NEW_GROUP_REQUEST } -SubscriptionPublisher::~SubscriptionPublisher() { +LivePublisher::~LivePublisher() { if (track_publisher_ != nullptr) { track_publisher_->RemoveObjectListener(this); } } -void SubscriptionPublisher::Update(const MessageParameters& parameters) { +void LivePublisher::Update(const MessageParameters& parameters) { // TODO(martinduke): If there are auth tokens, this probably has to go to the // application. // TODO(martinduke): If the subscribe window has shrunk, close any streams @@ -107,7 +107,7 @@ } } -void SubscriptionPublisher::ResetAllStreams() { +void LivePublisher::ResetAllStreams() { if (ignore_reset_all_streams_) { return; } @@ -119,7 +119,7 @@ } } -void SubscriptionPublisher::OnSubscribeAccepted() { +void LivePublisher::OnSubscribeAccepted() { if (established_) { return; // It's a PUBLISH. } @@ -150,15 +150,15 @@ // them. } -void SubscriptionPublisher::OnSubscribeRejected(MoqtRequestErrorInfo info) { +void LivePublisher::OnSubscribeRejected(MoqtRequestErrorInfo info) { bidi_stream_->CheckStatus(bidi_stream_->SendRequestError(request_id_, info, /*fin=*/true)); // Sending FIN will delete the class. } -void SubscriptionPublisher::OnNewObjectAvailable( - Location location, std::optional<uint64_t> subgroup, - MoqtPriority publisher_priority) { +void LivePublisher::OnNewObjectAvailable(Location location, + std::optional<uint64_t> subgroup, + MoqtPriority publisher_priority) { if (!InWindow(location)) { return; } @@ -255,13 +255,12 @@ } } -void SubscriptionPublisher::OnTrackPublisherGone() { +void LivePublisher::OnTrackPublisherGone() { PublishIsDone(PublishDoneCode::kGoingAway, "Publisher is gone"); } // TODO(martinduke): Revise to check if the last object has been delivered. -void SubscriptionPublisher::OnNewFinAvailable(Location location, - uint64_t subgroup) { +void LivePublisher::OnNewFinAvailable(Location location, uint64_t subgroup) { if (!InWindow(location.group)) { return; } @@ -281,7 +280,7 @@ // Sending FIN will delete the class. } -void SubscriptionPublisher::OnSubgroupAbandoned( +void LivePublisher::OnSubgroupAbandoned( uint64_t group, uint64_t subgroup, webtransport::StreamErrorCode error_code) { if (!InWindow(group)) { @@ -306,7 +305,7 @@ raw_stream->ResetWithUserCode(error_code); } -void SubscriptionPublisher::OnGroupAbandoned(uint64_t group_id) { +void LivePublisher::OnGroupAbandoned(uint64_t group_id) { if (!InWindow(group_id)) { // The group is not in the window, ignore. return; @@ -334,7 +333,7 @@ }); } -void SubscriptionPublisher::SendDatagram(Location sequence) { +void LivePublisher::SendDatagram(Location sequence) { std::optional<PublishedObject> object = track_publisher_->GetCachedObject( sequence.group, std::nullopt, sequence.object); if (!object.has_value()) { @@ -351,7 +350,7 @@ header.object_status = object->metadata.status; header.subgroup_id = std::nullopt; header.payload_length = object->metadata.payload_length; - QUICHE_BUG_IF(SubscriptionPublisher_SendDatagram_partial_payload, + QUICHE_BUG_IF(LivePublisher_SendDatagram_partial_payload, object->payload.size() > 1) << "Datagram is split into multiple slices"; quiche::QuicheBuffer datagram = framer_.SerializeObjectDatagram( @@ -364,7 +363,7 @@ OnObjectSent(object->metadata.location); } -void SubscriptionPublisher::ProcessObjectAck(const MoqtObjectAck& message) { +void LivePublisher::ProcessObjectAck(const MoqtObjectAck& message) { SessionToPublisherInterface* session_info = visitor(); if (session_info == nullptr) { return; @@ -379,7 +378,7 @@ } } -webtransport::Stream* absl_nullable SubscriptionPublisher::OpenDataStream( +webtransport::Stream* absl_nullable LivePublisher::OpenDataStream( const NewDataStreamParameters& parameters) { SessionToPublisherInterface* session_info = visitor(); if (session_info == nullptr) { @@ -404,8 +403,8 @@ return new_stream; } -void SubscriptionPublisher::PublishIsDone(PublishDoneCode code, - absl::string_view error_reason) { +void LivePublisher::PublishIsDone(PublishDoneCode code, + absl::string_view error_reason) { MoqtPublishDone publish_done; publish_done.request_id = request_id_; publish_done.status_code = code; @@ -421,12 +420,11 @@ // sending FIN will delete the class. } -void SubscriptionPublisher::OnDataStreamDestroyed( - DataStreamIndex end_sequence) { +void LivePublisher::OnDataStreamDestroyed(DataStreamIndex end_sequence) { stream_map_.RemoveStream(end_sequence); } -void SubscriptionPublisher::OnCanCreateNewUniStream() { +void LivePublisher::OnCanCreateNewUniStream() { SessionToPublisherInterface* session_info = visitor(); if (session_info == nullptr) { return; @@ -458,7 +456,7 @@ } } -void SubscriptionPublisher::OnObjectSent(Location sequence) { +void LivePublisher::OnObjectSent(Location sequence) { if (largest_sent_.has_value()) { largest_sent_ = std::max(*largest_sent_, sequence); } else {
diff --git a/quiche/quic/moqt/moqt_subscription.h b/quiche/quic/moqt/moqt_live_publisher.h similarity index 88% rename from quiche/quic/moqt/moqt_subscription.h rename to quiche/quic/moqt/moqt_live_publisher.h index d4af902..3d26abc 100644 --- a/quiche/quic/moqt/moqt_subscription.h +++ b/quiche/quic/moqt/moqt_live_publisher.h
@@ -36,7 +36,7 @@ namespace moqt { namespace test { -class SubscriptionPublisherPeer; +class LivePublisherPeer; } // namespace test // This is the part of the send order useful for ranking streams within the @@ -71,7 +71,7 @@ quic::QuicTimeDelta delta_from_deadline) = 0; }; -// Allows SubscriptionPublisher to get data from the session. +// Allows LivePublisher to get data from the session. class QUICHE_EXPORT SessionToPublisherInterface { public: virtual ~SessionToPublisherInterface() = default; @@ -94,28 +94,27 @@ // State for delivery of objects via a subscription, whether initiated by a // SUBSCRIBE or PUBLISH. -class SubscriptionPublisher : public MoqtObjectListener, - public SubscriptionPublisherInterface { +class LivePublisher : public MoqtObjectListener, public LivePublisherInterface { public: // The provider of this callback will add/delete whatever state it is tracking // for the subscription. This will be used by both PUBLISH and SUBSCRIBE // streams. AddCallback returns |false| if the add fails because the key // already exists. - using AddCallback = quiche::SingleUseCallback<bool(SubscriptionPublisher*)>; - using RemoveCallback = - quiche::SingleUseCallback<void(SubscriptionPublisher*)>; - SubscriptionPublisher( - MoqtFramer framer, std::shared_ptr<MoqtTrackPublisher> track_publisher, - MoqtBidiStreamBase* absl_nonnull bidi_stream, uint64_t request_id, - uint64_t track_alias, const MessageParameters& parameters, - quiche::QuicheWeakPtr<SessionToPublisherInterface> visitor, - bool is_publish); - ~SubscriptionPublisher(); + using AddCallback = quiche::SingleUseCallback<bool(LivePublisher*)>; + using RemoveCallback = quiche::SingleUseCallback<void(LivePublisher*)>; + LivePublisher(MoqtFramer framer, + std::shared_ptr<MoqtTrackPublisher> track_publisher, + MoqtBidiStreamBase* absl_nonnull bidi_stream, + uint64_t request_id, uint64_t track_alias, + const MessageParameters& parameters, + quiche::QuicheWeakPtr<SessionToPublisherInterface> visitor, + bool is_publish); + ~LivePublisher(); - SubscriptionPublisher(const SubscriptionPublisher&) = delete; - SubscriptionPublisher(SubscriptionPublisher&&) = delete; - SubscriptionPublisher& operator=(const SubscriptionPublisher&) = delete; - SubscriptionPublisher& operator=(SubscriptionPublisher&&) = delete; + LivePublisher(const LivePublisher&) = delete; + LivePublisher(LivePublisher&&) = delete; + LivePublisher& operator=(const LivePublisher&) = delete; + LivePublisher& operator=(LivePublisher&&) = delete; uint64_t request_id() const { return request_id_; } MoqtTrackPublisher& publisher() { return *track_publisher_; } @@ -130,13 +129,13 @@ MoqtPriority publisher_priority) override; void OnTrackPublisherGone() override; void OnNewFinAvailable(Location location, uint64_t subgroup) override; - // also a part of SubscriptionPublisherInterface. + // also a part of LivePublisherInterface. void OnSubgroupAbandoned(uint64_t group, uint64_t subgroup, webtransport::StreamErrorCode error_code) override; void OnGroupAbandoned(uint64_t group_id) override; void ProcessObjectAck(const MoqtObjectAck& message); - // SubscriptionPublisherInterface implementation. + // LivePublisherInterface implementation. bool InWindow(Location location) override { return parameters_.forward() && (!parameters_.subscription_filter.has_value() || @@ -186,7 +185,7 @@ bool established() const { return established_; } - quiche::QuicheWeakPtr<SubscriptionPublisherInterface> GetWeakPtr() { + quiche::QuicheWeakPtr<LivePublisherInterface> GetWeakPtr() { return weak_ptr_factory_.Create(); } @@ -198,7 +197,7 @@ void IgnoreResetAllStreams() { ignore_reset_all_streams_ = true; } private: - friend class test::SubscriptionPublisherPeer; + friend class test::LivePublisherPeer; MoqtPriority default_publisher_priority() const { return default_publisher_priority_.value_or(kDefaultPublisherPriority); @@ -291,8 +290,7 @@ // stream. absl::btree_multimap<StreamRank, NewDataStreamParameters> pending_streams_; // Must be last. - quiche::QuicheWeakPtrFactory<SubscriptionPublisherInterface> - weak_ptr_factory_; + quiche::QuicheWeakPtrFactory<LivePublisherInterface> weak_ptr_factory_; }; } // namespace moqt
diff --git a/quiche/quic/moqt/moqt_subscription_test.cc b/quiche/quic/moqt/moqt_live_publisher_test.cc similarity index 91% rename from quiche/quic/moqt/moqt_subscription_test.cc rename to quiche/quic/moqt/moqt_live_publisher_test.cc index b35818d..dd5b0b7 100644 --- a/quiche/quic/moqt/moqt_subscription_test.cc +++ b/quiche/quic/moqt/moqt_live_publisher_test.cc
@@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. -#include "quiche/quic/moqt/moqt_subscription.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include <cstddef> #include <cstdint> @@ -45,17 +45,16 @@ namespace moqt::test { -class SubscriptionPublisherPeer { +class LivePublisherPeer { public: - static size_t num_open_streams(SubscriptionPublisher* publisher) { + static size_t num_open_streams(LivePublisher* publisher) { return publisher->stream_map_.GetAllStreams().size(); } - static std::optional<Location> largest_sent( - const SubscriptionPublisher* publisher) { + static std::optional<Location> largest_sent(const LivePublisher* publisher) { return publisher->largest_sent_; } static const absl::flat_hash_set<DataStreamIndex>& reset_subgroups( - const SubscriptionPublisher* publisher) { + const LivePublisher* publisher) { return publisher->reset_subgroups_; } }; @@ -104,9 +103,9 @@ return object; } -class SubscriptionPublisherTest : public quic::test::QuicTest { +class LivePublisherTest : public quic::test::QuicTest { public: - SubscriptionPublisherTest() + LivePublisherTest() : track_publisher_( std::make_shared<MockTrackPublisher>(FullTrackName("foo", "bar"))), bidi_stream_(&framer_, message_parser_, @@ -121,7 +120,7 @@ EXPECT_CALL(visitor_, session).WillRepeatedly(Return(&webtrans_)); ON_CALL(visitor_, ReleaseMonitoringInterface) .WillByDefault(Return(&monitoring_interface_)); - publisher_ = std::make_unique<SubscriptionPublisher>( + publisher_ = std::make_unique<LivePublisher>( framer_, track_publisher_, &bidi_stream_, kRequestId, kTrackAlias, parameters_, visitor_.weak_ptr_factory_.Create(), /*is_publish=*/false); @@ -133,7 +132,7 @@ ON_CALL(visitor_, trace_recorder).WillByDefault(ReturnRef(trace_recorder_)); } - ~SubscriptionPublisherTest() override { + ~LivePublisherTest() override { if (track_publisher_ == nullptr) { return; } @@ -227,7 +226,7 @@ MockSessionToPublisherInterface visitor_; StrictMock<MockPublishingMonitorInterface> monitoring_interface_; MoqtTraceRecorder trace_recorder_; - std::unique_ptr<SubscriptionPublisher> publisher_; + std::unique_ptr<LivePublisher> publisher_; const TrackExtensions extensions_; quic::MockClock mock_clock_; MoqtSessionCallbacks callbacks_; @@ -235,7 +234,7 @@ int open_streams_ = 0; }; -TEST_F(SubscriptionPublisherTest, OnSubscribeAcceptedNoFilter) { +TEST_F(LivePublisherTest, OnSubscribeAcceptedNoFilter) { EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true)); EXPECT_CALL(*track_publisher_, largest_location()) .WillOnce(Return(Location(1, 2))); @@ -252,7 +251,7 @@ EXPECT_FALSE(publisher_->parameters().subscription_filter.has_value()); } -TEST_F(SubscriptionPublisherTest, OnSubscribeAcceptedWithFilter) { +TEST_F(LivePublisherTest, OnSubscribeAcceptedWithFilter) { publisher_->parameters().subscription_filter = SubscriptionFilter(MoqtFilterType::kLargestObject); const TrackExtensions extensions(std::nullopt, std::nullopt, @@ -292,7 +291,7 @@ publisher_->OnNewObjectAvailable(Location(1, 3), std::nullopt, 64); } -TEST_F(SubscriptionPublisherTest, OnSubscribeRejected) { +TEST_F(LivePublisherTest, OnSubscribeRejected) { EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true)); EXPECT_CALL(mock_bidi_stream_, Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _)) @@ -301,7 +300,7 @@ RequestErrorCode::kDoesNotExist, std::nullopt, "reason")); } -TEST_F(SubscriptionPublisherTest, Update) { +TEST_F(LivePublisherTest, Update) { MessageParameters new_params; new_params.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); publisher_->Update(new_params); @@ -313,14 +312,14 @@ EXPECT_FALSE(publisher_->can_have_joining_fetch()); } -TEST_F(SubscriptionPublisherTest, UpdatePriorityNoStreams) { +TEST_F(LivePublisherTest, UpdatePriorityNoStreams) { MessageParameters new_params; new_params.subscriber_priority = 20; publisher_->Update(new_params); EXPECT_EQ(publisher_->parameters().subscriber_priority, 20); } -TEST_F(SubscriptionPublisherTest, UpdatePriorityWithPendingStreams) { +TEST_F(LivePublisherTest, UpdatePriorityWithPendingStreams) { CreatePendingStream(Location(1, 0), 0, 64); MessageParameters new_params; new_params.subscriber_priority = 20; @@ -333,7 +332,7 @@ publisher_->Update(new_params); } -TEST_F(SubscriptionPublisherTest, UpdatePriorityWithActiveStreams) { +TEST_F(LivePublisherTest, UpdatePriorityWithActiveStreams) { CreateStream( Location(1, 0), 0, 127, {0x51, static_cast<uint8_t>(kTrackAlias), 0x01, 0x7f, 0x00, 0x0a}); @@ -343,7 +342,7 @@ publisher_->Update(new_params); } -TEST_F(SubscriptionPublisherTest, OnNewObjectAvailableNotInWindow) { +TEST_F(LivePublisherTest, OnNewObjectAvailableNotInWindow) { MessageParameters params; params.subscription_filter = SubscriptionFilter(Location(10, 0), 10); publisher_->Update(params); @@ -351,7 +350,7 @@ publisher_->OnNewObjectAvailable(Location(5, 0), 0, 128); } -TEST_F(SubscriptionPublisherTest, OnNewObjectAvailableDatagram) { +TEST_F(LivePublisherTest, OnNewObjectAvailableDatagram) { EXPECT_CALL(*track_publisher_, GetCachedObject(1, std::optional<uint64_t>(), 0, 0)) .WillOnce( @@ -366,11 +365,11 @@ publisher_->OnNewObjectAvailable(Location(1, 0), std::nullopt, 128); } -TEST_F(SubscriptionPublisherTest, OnNewObjectAvailableStreamCreationBlocked) { +TEST_F(LivePublisherTest, OnNewObjectAvailableStreamCreationBlocked) { CreatePendingStream(Location(1, 0), 0, 128); } -TEST_F(SubscriptionPublisherTest, OnNewFinAvailableNoops) { +TEST_F(LivePublisherTest, OnNewFinAvailableNoops) { // Not in window MessageParameters params; params.subscription_filter = SubscriptionFilter(Location(10, 0), 10); @@ -393,7 +392,7 @@ publisher_->OnNewFinAvailable(Location(10, 1), 0); } -TEST_F(SubscriptionPublisherTest, OnNewFinAvailableWithStream) { +TEST_F(LivePublisherTest, OnNewFinAvailableWithStream) { CreateStream(Location(1, 0), 0, 128); EXPECT_CALL(mock_uni_stream_, Writev) .WillOnce([](absl::Span<quiche::QuicheMemSlice> data, @@ -407,7 +406,7 @@ publisher_->OnNewFinAvailable(Location(1, 0), 0); } -TEST_F(SubscriptionPublisherTest, OnSubgroupAbandonedNoEffect) { +TEST_F(LivePublisherTest, OnSubgroupAbandonedNoEffect) { // Not in window MessageParameters params; params.subscription_filter = SubscriptionFilter(Location(10, 0), 10); @@ -421,7 +420,7 @@ publisher_->OnSubgroupAbandoned(1, 0, 17); } -TEST_F(SubscriptionPublisherTest, OnGroupAbandoned) { +TEST_F(LivePublisherTest, OnGroupAbandoned) { // Not in window MessageParameters params; params.subscription_filter = SubscriptionFilter(Location(10, 0), 10); @@ -437,7 +436,7 @@ publisher_->OnNewObjectAvailable(Location(1, 0), 0, 128); } -TEST_F(SubscriptionPublisherTest, OnGroupAbandonedWithStreams) { +TEST_F(LivePublisherTest, OnGroupAbandonedWithStreams) { // The delivery timeout is not infinite, so it will not send a PUBLISH_DONE // with kTooFarBehind. CreateStream(Location(1, 0), 0, 128); @@ -446,7 +445,7 @@ publisher_->OnGroupAbandoned(1); } -TEST_F(SubscriptionPublisherTest, OnGroupAbandonedTooFarBehind) { +TEST_F(LivePublisherTest, OnGroupAbandonedTooFarBehind) { // Set the delivery timeout to infinite so that TooFarBehind is possible. parameters_.delivery_timeout = quic::QuicTimeDelta::Infinite(); publisher_->Update(parameters_); @@ -470,7 +469,7 @@ track_publisher_ = nullptr; } -TEST_F(SubscriptionPublisherTest, OnCanCreateNewUniStreamPendingCleanup) { +TEST_F(LivePublisherTest, OnCanCreateNewUniStreamPendingCleanup) { CreatePendingStream(Location(1, 0), 0, 128); // Abandon the group. publisher_->OnGroupAbandoned(1); @@ -481,7 +480,7 @@ publisher_->OnCanCreateNewUniStream(); } -TEST_F(SubscriptionPublisherTest, AlternateDeliveryTimeoutSetAlarm) { +TEST_F(LivePublisherTest, AlternateDeliveryTimeoutSetAlarm) { ON_CALL(visitor_, alternate_delivery_timeout).WillByDefault(Return(true)); // Create a stream for group 1. CreateStream(Location(1, 0), 0, 128); @@ -491,7 +490,7 @@ CreatePendingStream(Location(2, 0), 0, 128); } -TEST_F(SubscriptionPublisherTest, OnTrackPublisherGone) { +TEST_F(LivePublisherTest, OnTrackPublisherGone) { EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true)); EXPECT_CALL(mock_bidi_stream_, Writev(ControlMessageOfType(MoqtMessageType::kPublishDone), _)) @@ -501,7 +500,7 @@ track_publisher_ = nullptr; } -TEST_F(SubscriptionPublisherTest, ProcessObjectAck) { +TEST_F(LivePublisherTest, ProcessObjectAck) { MoqtObjectAck ack; ack.group_id = 1; ack.object_id = 2; @@ -511,13 +510,13 @@ publisher_->ProcessObjectAck(ack); } -TEST_F(SubscriptionPublisherTest, OnSubgroupAbandonedWithStream) { +TEST_F(LivePublisherTest, OnSubgroupAbandonedWithStream) { CreateStream(Location(1, 0), 0, 128); EXPECT_CALL(mock_uni_stream_, ResetWithUserCode(17)); publisher_->OnSubgroupAbandoned(1, 0, 17); } -TEST_F(SubscriptionPublisherTest, OnCanCreateNewUniStreamSuccess) { +TEST_F(LivePublisherTest, OnCanCreateNewUniStreamSuccess) { CreatePendingStream(Location(1, 0), 0, 128); // Call OnCanCreateNewUniStream and succeed. EXPECT_CALL(mock_uni_stream_, GetStreamId()) @@ -545,7 +544,7 @@ publisher_->OnCanCreateNewUniStream(); } -TEST_F(SubscriptionPublisherTest, PendingStreamsInOrder) { +TEST_F(LivePublisherTest, PendingStreamsInOrder) { CreatePendingStream(Location(1, 0), 0, 128); CreatePendingStream(Location(0, 0), 0, 128); CreatePendingStream(Location(2, 0), 0, 127); @@ -633,7 +632,7 @@ publisher_->OnCanCreateNewUniStream(); } -TEST_F(SubscriptionPublisherTest, OnDataStreamDestroyed) { +TEST_F(LivePublisherTest, OnDataStreamDestroyed) { CreateStream(Location(1, 0), 0, 128); DataStreamIndex index(1, 0); publisher_->OnDataStreamDestroyed(index); @@ -643,15 +642,14 @@ publisher_->Update(parameters_); } -TEST_F(SubscriptionPublisherTest, OnObjectSentTwice) { +TEST_F(LivePublisherTest, OnObjectSentTwice) { publisher_->OnObjectSent(Location(1, 0)); - EXPECT_TRUE( - SubscriptionPublisherPeer::largest_sent(publisher_.get()).has_value() && - *SubscriptionPublisherPeer::largest_sent(publisher_.get()) == - Location(1, 0)); + EXPECT_TRUE(LivePublisherPeer::largest_sent(publisher_.get()).has_value() && + *LivePublisherPeer::largest_sent(publisher_.get()) == + Location(1, 0)); } -TEST_F(SubscriptionPublisherTest, AlternateDeliveryTimeout) { +TEST_F(LivePublisherTest, AlternateDeliveryTimeout) { EXPECT_CALL(visitor_, alternate_delivery_timeout) .WillRepeatedly(Return(true)); CreateStream(Location(0, 0), 0, 128); @@ -687,7 +685,7 @@ nullptr); } -TEST_F(SubscriptionPublisherTest, IncomingUpdateTruncatesSubscription) { +TEST_F(LivePublisherTest, IncomingUpdateTruncatesSubscription) { // Track gets to Group 5. CreateStream(Location(5, 0), 0, 128); parameters_.subscription_filter = SubscriptionFilter(Location(0, 0), 4); @@ -696,7 +694,7 @@ publisher_->OnNewObjectAvailable(Location(5, 1), 0, 128); } -TEST_F(SubscriptionPublisherTest, OnNewFinAvailable) { +TEST_F(LivePublisherTest, OnNewFinAvailable) { CreateStream( Location(1, 0), 0, 127, {0x51, static_cast<uint8_t>(kTrackAlias), 0x01, 0x7f, 0x00, 0x0a}); @@ -709,7 +707,7 @@ publisher_->OnNewFinAvailable(Location(1, 0), 0); } -TEST_F(SubscriptionPublisherTest, OnSubgroupAbandoned) { +TEST_F(LivePublisherTest, OnSubgroupAbandoned) { CreateStream( Location(1, 0), 0, 127, {0x51, static_cast<uint8_t>(kTrackAlias), 0x01, 0x7f, 0x00, 0x0a}); @@ -717,7 +715,7 @@ publisher_->OnSubgroupAbandoned(1, 0, 1234); } -TEST_F(SubscriptionPublisherTest, OnSubgroupAbandonedOutsideWindow) { +TEST_F(LivePublisherTest, OnSubgroupAbandonedOutsideWindow) { parameters_.subscription_filter = SubscriptionFilter(Location(20, 0)); publisher_->Update(parameters_); EXPECT_CALL(mock_uni_stream_, ResetWithUserCode).Times(0);
diff --git a/quiche/quic/moqt/moqt_namespace_stream.cc b/quiche/quic/moqt/moqt_namespace_stream.cc index b7a9672..4d32218 100644 --- a/quiche/quic/moqt/moqt_namespace_stream.cc +++ b/quiche/quic/moqt/moqt_namespace_stream.cc
@@ -28,24 +28,24 @@ namespace moqt { -MoqtNamespaceSubscriberStream::~MoqtNamespaceSubscriberStream() { +MoqtSubscribeNamespaceRequestStream::~MoqtSubscribeNamespaceRequestStream() { NamespaceTask* task = task_.GetIfAvailable(); if (task != nullptr) { task->DeclareEof(); } Detach(); } -absl::Status MoqtNamespaceSubscriberStream::OnRawControlMessage( +absl::Status MoqtSubscribeNamespaceRequestStream::OnRawControlMessage( const MoqtRawControlMessage& message) { return ControlMessageDispatcher::DispatchControlMessage( *this, message_parser(), message, "namespace subscriber"); } -void MoqtNamespaceSubscriberStream::OnStreamBound() { +void MoqtSubscribeNamespaceRequestStream::OnStreamBound() { // TODO(martinduke): Set the priority for this stream. } -absl::Status MoqtNamespaceSubscriberStream::OnControlMessage( +absl::Status MoqtSubscribeNamespaceRequestStream::OnControlMessage( const MoqtRequestOk& message) { if (message.request_id == request_id_) { // Response to the initial SUBSCRIBE_NAMESPACE. @@ -70,7 +70,7 @@ return absl::OkStatus(); } -absl::Status MoqtNamespaceSubscriberStream::OnControlMessage( +absl::Status MoqtSubscribeNamespaceRequestStream::OnControlMessage( const MoqtRequestError& message) { if (message.request_id == request_id_) { if (response_callback_ == nullptr) { @@ -96,7 +96,7 @@ return absl::OkStatus(); } -absl::Status MoqtNamespaceSubscriberStream::OnControlMessage( +absl::Status MoqtSubscribeNamespaceRequestStream::OnControlMessage( const MoqtNamespace& message) { if (response_callback_ != nullptr) { return absl::InvalidArgumentError( @@ -130,7 +130,7 @@ return absl::OkStatus(); } -absl::Status MoqtNamespaceSubscriberStream::OnControlMessage( +absl::Status MoqtSubscribeNamespaceRequestStream::OnControlMessage( const MoqtNamespaceDone& message) { if (response_callback_ != nullptr) { return absl::InvalidArgumentError( @@ -151,8 +151,8 @@ return absl::OkStatus(); } -std::unique_ptr<MoqtNamespaceTask> MoqtNamespaceSubscriberStream::CreateTask( - const TrackNamespace& prefix) { +std::unique_ptr<MoqtNamespaceTask> +MoqtSubscribeNamespaceRequestStream::CreateTask(const TrackNamespace& prefix) { auto task = std::make_unique<NamespaceTask>(this, prefix); QUICHE_DCHECK(task != nullptr); task_ = task->GetWeakPtr(); @@ -160,21 +160,22 @@ return std::move(task); } -MoqtNamespaceSubscriberStream::NamespaceTask::~NamespaceTask() { +MoqtSubscribeNamespaceRequestStream::NamespaceTask::~NamespaceTask() { if (state_ != nullptr) { state_->Reset(kResetCodeCancelled); } } -void MoqtNamespaceSubscriberStream::NamespaceTask::SetObjectsAvailableCallback( - ObjectsAvailableCallback absl_nullable callback) { +void MoqtSubscribeNamespaceRequestStream::NamespaceTask:: + SetObjectsAvailableCallback(ObjectsAvailableCallback + absl_nullable callback) { callback_ = std::move(callback); if (!pending_suffixes_.empty() && callback_ != nullptr) { callback_(); } } -void MoqtNamespaceSubscriberStream::NamespaceTask::Update( +void MoqtSubscribeNamespaceRequestStream::NamespaceTask::Update( const MessageParameters& parameters, MoqtResponseCallback response_callback) { if (state_ == nullptr) { @@ -190,7 +191,7 @@ next_request_id_ += 2; } -GetNextResult MoqtNamespaceSubscriberStream::NamespaceTask::GetNextSuffix( +GetNextResult MoqtSubscribeNamespaceRequestStream::NamespaceTask::GetNextSuffix( TrackNamespace& suffix, TransactionType& type) { if (pending_suffixes_.empty()) { if (error_.has_value()) { @@ -207,7 +208,7 @@ return kSuccess; } -void MoqtNamespaceSubscriberStream::NamespaceTask::AddPendingSuffix( +void MoqtSubscribeNamespaceRequestStream::NamespaceTask::AddPendingSuffix( TrackNamespace suffix, TransactionType type) { if (pending_suffixes_.size() == kMaxPendingSuffixes) { error_ = kResetCodeTooFarBehind; @@ -222,7 +223,7 @@ } } -void MoqtNamespaceSubscriberStream::NamespaceTask::DeclareEof() { +void MoqtSubscribeNamespaceRequestStream::NamespaceTask::DeclareEof() { if (eof_) { return; } @@ -234,7 +235,7 @@ } MoqtResponseCallback -MoqtNamespaceSubscriberStream::NamespaceTask::GetResponseCallback( +MoqtSubscribeNamespaceRequestStream::NamespaceTask::GetResponseCallback( uint64_t request_id) { auto it = pending_updates_.find(request_id); if (it == pending_updates_.end()) { @@ -245,7 +246,7 @@ return callback; } -MoqtNamespacePublisherStream::MoqtNamespacePublisherStream( +MoqtSubscribeNamespaceResponseStream::MoqtSubscribeNamespaceResponseStream( MoqtFramer* framer, const MoqtControlMessageParser& message_parser, AddPrefixCallback add_callback, RemovePrefixCallback remove_callback, SessionErrorCallback session_error_callback, @@ -257,13 +258,13 @@ remove_callback_(std::move(remove_callback)), application_(application) {} -absl::Status MoqtNamespacePublisherStream::OnRawControlMessage( +absl::Status MoqtSubscribeNamespaceResponseStream::OnRawControlMessage( const MoqtRawControlMessage& message) { return ControlMessageDispatcher::DispatchControlMessage( *this, message_parser(), message, "namespace publisher"); } -absl::Status MoqtNamespacePublisherStream::OnControlMessage( +absl::Status MoqtSubscribeNamespaceResponseStream::OnControlMessage( const MoqtSubscribeNamespace& message) { request_id_ = message.request_id; if (add_callback_ == nullptr) { @@ -285,7 +286,7 @@ return absl::OkStatus(); } -absl::Status MoqtNamespacePublisherStream::OnControlMessage( +absl::Status MoqtSubscribeNamespaceResponseStream::OnControlMessage( const MoqtRequestUpdate& message) { if (task_ == nullptr) { // This stream is dying. @@ -295,14 +296,14 @@ return absl::OkStatus(); } -void MoqtNamespacePublisherStream::Detach() { +void MoqtSubscribeNamespaceResponseStream::Detach() { if (remove_callback_ != nullptr) { std::move(remove_callback_)(prefix_); remove_callback_ = nullptr; } } -void MoqtNamespacePublisherStream::ProcessNamespaces() { +void MoqtSubscribeNamespaceResponseStream::ProcessNamespaces() { if (task_ == nullptr) { return; } @@ -364,7 +365,7 @@ } } -MoqtResponseCallback MoqtNamespacePublisherStream::ResponseCallback( +MoqtResponseCallback MoqtSubscribeNamespaceResponseStream::ResponseCallback( uint64_t request_id) { return [this, request_id]( std::variant<MessageParameters, MoqtRequestErrorInfo> response) {
diff --git a/quiche/quic/moqt/moqt_namespace_stream.h b/quiche/quic/moqt/moqt_namespace_stream.h index 34e64d1..4d1bdbc 100644 --- a/quiche/quic/moqt/moqt_namespace_stream.h +++ b/quiche/quic/moqt/moqt_namespace_stream.h
@@ -36,21 +36,20 @@ quiche::SingleUseCallback<void(const TrackNamespace&)>; // This class will be owned by the webtransport stream. -class MoqtNamespaceSubscriberStream : public MoqtBidiStreamBase { +class MoqtSubscribeNamespaceRequestStream : public MoqtBidiStreamBase { public: // Assumes the caller will send or queue the SUBSCRIBE_NAMESPACE. - MoqtNamespaceSubscriberStream(MoqtFramer* framer, - const MoqtControlMessageParser& message_parser, - uint64_t request_id, - RemovePrefixCallback remove_callback, - SessionErrorCallback session_error_callback, - MoqtResponseCallback response_callback) + MoqtSubscribeNamespaceRequestStream( + MoqtFramer* framer, const MoqtControlMessageParser& message_parser, + uint64_t request_id, RemovePrefixCallback remove_callback, + SessionErrorCallback session_error_callback, + MoqtResponseCallback response_callback) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), request_id_(request_id), remove_callback_(std::move(remove_callback)), response_callback_(std::move(response_callback)) {} - ~MoqtNamespaceSubscriberStream(); + ~MoqtSubscribeNamespaceRequestStream(); // MoqtBidiStreamBase overrides. void OnStreamBound() override; @@ -85,7 +84,7 @@ // information. Owned by the application. class NamespaceTask : public MoqtNamespaceTask { public: - NamespaceTask(MoqtNamespaceSubscriberStream* absl_nonnull state, + NamespaceTask(MoqtSubscribeNamespaceRequestStream* absl_nonnull state, const TrackNamespace& prefix) : MoqtNamespaceTask(), prefix_(prefix), @@ -128,7 +127,7 @@ static constexpr size_t kMaxPendingSuffixes = 100; const TrackNamespace prefix_; // Must be nonnull initially, will be nullptr if the stream is closed. - MoqtNamespaceSubscriberStream* state_; + MoqtSubscribeNamespaceRequestStream* state_; quiche::QuicheCircularDeque<PendingSuffix> pending_suffixes_; ObjectsAvailableCallback absl_nullable callback_ = nullptr; std::optional<webtransport::StreamErrorCode> error_; @@ -146,15 +145,15 @@ quiche::QuicheWeakPtr<NamespaceTask> task_; }; -class MoqtNamespacePublisherStream : public MoqtBidiStreamBase { +class MoqtSubscribeNamespaceResponseStream : public MoqtBidiStreamBase { public: // Constructor for the publisher side. - MoqtNamespacePublisherStream( + MoqtSubscribeNamespaceResponseStream( MoqtFramer* framer, const MoqtControlMessageParser& message_parser, AddPrefixCallback add_callback, RemovePrefixCallback remove_callback, SessionErrorCallback session_error_callback, MoqtIncomingSubscribeNamespaceCallback& application); - ~MoqtNamespacePublisherStream() { Detach(); } + ~MoqtSubscribeNamespaceResponseStream() { Detach(); } void OnStreamBound() override { // TODO(martinduke): Set the priority for this stream.
diff --git a/quiche/quic/moqt/moqt_namespace_stream_test.cc b/quiche/quic/moqt/moqt_namespace_stream_test.cc index e94a45c..c38258f 100644 --- a/quiche/quic/moqt/moqt_namespace_stream_test.cc +++ b/quiche/quic/moqt/moqt_namespace_stream_test.cc
@@ -48,9 +48,10 @@ quic::Perspective::IS_CLIENT); } -class MoqtNamespaceSubscriberStreamTest : public quiche::test::QuicheTest { +class MoqtSubscribeNamespaceRequestStreamTest + : public quiche::test::QuicheTest { public: - MoqtNamespaceSubscriberStreamTest() + MoqtSubscribeNamespaceRequestStreamTest() : framer_(true, quic::Perspective::IS_CLIENT), stream_(&framer_, ControlMessageParser(), kRequestId, deleted_callback_.AsStdFunction(), @@ -78,32 +79,32 @@ std::variant<MessageParameters, MoqtRequestErrorInfo>)> response_callback_; webtransport::test::MockStream mock_stream_; - MoqtNamespaceSubscriberStream stream_; + MoqtSubscribeNamespaceRequestStream stream_; int objects_available_ = 0; std::unique_ptr<MoqtNamespaceTask> task_; }; -TEST_F(MoqtNamespaceSubscriberStreamTest, RequestOk) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, RequestOk) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); ReceiveControlMessage(MoqtRequestOk{kRequestId}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, RequestOkWrongId) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, RequestOkWrongId) { EXPECT_CALL(error_callback_, Call(MoqtError::kProtocolViolation, "Unexpected request ID in response")); ReceiveControlMessage(MoqtRequestOk{kRequestId + 1}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, RequestError) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, RequestError) { EXPECT_CALL(response_callback_, Call); ReceiveControlMessage( MoqtRequestError{kRequestId, RequestErrorCode::kInternalError, quic::QuicTimeDelta::FromMilliseconds(100), "bar"}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, RequestErrorWrongId) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, RequestErrorWrongId) { EXPECT_CALL(error_callback_, Call(MoqtError::kProtocolViolation, "Unexpected request ID in response")); ReceiveControlMessage( @@ -111,21 +112,21 @@ quic::QuicTimeDelta::FromMilliseconds(100), "bar"}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, NamespaceBeforeResponse) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, NamespaceBeforeResponse) { EXPECT_CALL(error_callback_, Call(MoqtError::kProtocolViolation, "First message must be REQUEST_OK or REQUEST_ERROR")); ReceiveControlMessage(MoqtNamespace{TrackNamespace({"bar"})}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, NamespaceDoneBeforeResponse) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, NamespaceDoneBeforeResponse) { EXPECT_CALL(error_callback_, Call(MoqtError::kProtocolViolation, "First message must be REQUEST_OK or REQUEST_ERROR")); ReceiveControlMessage(MoqtNamespaceDone{TrackNamespace({"bar"})}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, NamespaceAfterResponse) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, NamespaceAfterResponse) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); @@ -140,7 +141,7 @@ EXPECT_EQ(task_->GetNextSuffix(received_namespace, type), kPending); } -TEST_F(MoqtNamespaceSubscriberStreamTest, NamespaceDoneAfterResponse) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, NamespaceDoneAfterResponse) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); @@ -160,7 +161,7 @@ EXPECT_EQ(task_->GetNextSuffix(received_namespace, type), kPending); } -TEST_F(MoqtNamespaceSubscriberStreamTest, DuplicateNamespace) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, DuplicateNamespace) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); @@ -173,7 +174,7 @@ ReceiveControlMessage(MoqtNamespace{TrackNamespace({"bar"})}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, NamespaceDoneWithoutNamespace) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, NamespaceDoneWithoutNamespace) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); @@ -183,7 +184,7 @@ ReceiveControlMessage(MoqtNamespaceDone{TrackNamespace({"bar"})}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, NamespaceDoneThenNamespace) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, NamespaceDoneThenNamespace) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); @@ -197,7 +198,7 @@ CheckNumberOfObjectsAvailable(3); } -TEST_F(MoqtNamespaceSubscriberStreamTest, TaskGetNextSuffix) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, TaskGetNextSuffix) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); @@ -228,8 +229,8 @@ EXPECT_EQ(task_->GetNextSuffix(received_namespace, type), kPending); } -TEST_F(MoqtNamespaceSubscriberStreamTest, DeclareEof) { - auto stream = std::make_unique<MoqtNamespaceSubscriberStream>( +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, DeclareEof) { + auto stream = std::make_unique<MoqtSubscribeNamespaceRequestStream>( &framer_, ControlMessageParser(), kRequestId, deleted_callback_.AsStdFunction(), error_callback_.AsStdFunction(), response_callback_.AsStdFunction()); @@ -253,7 +254,7 @@ EXPECT_EQ(task->GetNextSuffix(received_namespace, type), kEof); } -TEST_F(MoqtNamespaceSubscriberStreamTest, UpdateAndRequestOk) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, UpdateAndRequestOk) { EXPECT_CALL( response_callback_, Call(testing::VariantWith<MessageParameters>(Eq(MessageParameters())))); @@ -273,7 +274,7 @@ ReceiveControlMessage(MoqtRequestOk{kRequestId + 2, ok_params}); } -TEST_F(MoqtNamespaceSubscriberStreamTest, UpdateAndRequestError) { +TEST_F(MoqtSubscribeNamespaceRequestStreamTest, UpdateAndRequestError) { MessageParameters ok_params; ok_params.expires = quic::QuicTimeDelta::FromSeconds(60); EXPECT_CALL(response_callback_, @@ -293,9 +294,10 @@ quic::QuicTimeDelta::FromMilliseconds(100), "bar"}); } -class MoqtNamespacePublisherStreamTest : public quiche::test::QuicheTest { +class MoqtSubscribeNamespaceResponseStreamTest + : public quiche::test::QuicheTest { public: - MoqtNamespacePublisherStreamTest() + MoqtSubscribeNamespaceResponseStreamTest() : framer_(false, quic::Perspective::IS_CLIENT), application_callback_(mock_application_.AsStdFunction()), stream_(&framer_, ControlMessageParser(), add_callback_.AsStdFunction(), @@ -320,10 +322,10 @@ MoqtResponseCallback)> mock_application_; MoqtIncomingSubscribeNamespaceCallback application_callback_; - MoqtNamespacePublisherStream stream_; + MoqtSubscribeNamespaceResponseStream stream_; }; -TEST_F(MoqtNamespacePublisherStreamTest, Subscribe) { +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, Subscribe) { MoqtSubscribeNamespace message = { kRequestId, TrackNamespace({"foo"}), @@ -394,7 +396,7 @@ task_ptr->InvokeCallback(); } -TEST_F(MoqtNamespacePublisherStreamTest, SubscribeUnsubscribe) { +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, SubscribeUnsubscribe) { MoqtSubscribeNamespace message = { kRequestId, TrackNamespace({"foo"}), @@ -425,7 +427,7 @@ stream_.OnResetStreamReceived(kResetCodeCancelled); } -TEST_F(MoqtNamespacePublisherStreamTest, RequestError) { +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, RequestError) { MoqtSubscribeNamespace message = { kRequestId, TrackNamespace({"foo"}), @@ -447,7 +449,7 @@ ReceiveControlMessage(message); } -TEST_F(MoqtNamespacePublisherStreamTest, RequestUpdateOk) { +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, RequestUpdateOk) { MoqtSubscribeNamespace message = { kRequestId, TrackNamespace({"foo"}), @@ -489,7 +491,7 @@ ReceiveControlMessage(update_message); } -TEST_F(MoqtNamespacePublisherStreamTest, RequestUpdateError) { +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, RequestUpdateError) { MoqtSubscribeNamespace message = { kRequestId, TrackNamespace({"foo"}), @@ -532,7 +534,7 @@ ReceiveControlMessage(update_message); } -TEST_F(MoqtNamespacePublisherStreamTest, SubscribePrefixOverlap) { +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, SubscribePrefixOverlap) { MoqtSubscribeNamespace message = { kRequestId, TrackNamespace({"foo", "bar", "baz"}), @@ -546,7 +548,7 @@ ReceiveControlMessage(message); } -TEST_F(MoqtNamespacePublisherStreamTest, +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, DuplicateSubscribeNamespaceOnSameStream) { MoqtSubscribeNamespace message = { kRequestId, @@ -578,7 +580,7 @@ ReceiveControlMessage(message2); } -TEST_F(MoqtNamespacePublisherStreamTest, +TEST_F(MoqtSubscribeNamespaceResponseStreamTest, DuplicateSubscribeNamespaceOnDifferentStreams) { MoqtSubscribeNamespace message1 = { kRequestId, @@ -601,7 +603,7 @@ testing::MockFunction<void(MoqtError, absl::string_view)> error_callback2; webtransport::test::MockStream mock_stream2; - MoqtNamespacePublisherStream stream2( + MoqtSubscribeNamespaceResponseStream stream2( &framer_, ControlMessageParser(), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), error_callback2.AsStdFunction(), application_callback_);
diff --git a/quiche/quic/moqt/moqt_track.cc b/quiche/quic/moqt/moqt_object_subscriber.cc similarity index 91% rename from quiche/quic/moqt/moqt_track.cc rename to quiche/quic/moqt/moqt_object_subscriber.cc index 23060f0..478044e 100644 --- a/quiche/quic/moqt/moqt_track.cc +++ b/quiche/quic/moqt/moqt_object_subscriber.cc
@@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file -#include "quiche/quic/moqt/moqt_track.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include <algorithm> #include <cstdint> @@ -39,7 +39,7 @@ } // namespace -SubscribeRemoteTrack::~SubscribeRemoteTrack() { +LiveSubscriber::~LiveSubscriber() { if (publish_done_alarm_ != nullptr) { publish_done_alarm_->PermanentCancel(); } @@ -49,7 +49,7 @@ } } -void SubscribeRemoteTrack::OnObjectOrOk(const SubscribeOkData& data) { +void LiveSubscriber::OnObjectOrOk(const SubscribeOkData& data) { if (parameters().subscription_filter.has_value()) { parameters().subscription_filter->OnLargestObject( data.parameters.largest_object); @@ -62,15 +62,15 @@ OnObjectOrOk(); } -void SubscribeRemoteTrack::OnStreamOpened() { +void LiveSubscriber::OnStreamOpened() { ++currently_open_streams_; if (publish_done_alarm_ != nullptr && publish_done_alarm_->IsSet()) { publish_done_alarm_->Cancel(); } } -void SubscribeRemoteTrack::OnStreamClosed( - bool fin_received, std::optional<DataStreamIndex> index) { +void LiveSubscriber::OnStreamClosed(bool fin_received, + std::optional<DataStreamIndex> index) { ++streams_closed_; --currently_open_streams_; QUICHE_DCHECK_GE(currently_open_streams_, -1); @@ -93,9 +93,9 @@ MaybeSetPublishDoneAlarm(); } -void SubscribeRemoteTrack::OnPublishDone( - uint64_t stream_count, const quic::QuicClock* clock, - quic::QuicAlarmFactory* alarm_factory) { +void LiveSubscriber::OnPublishDone(uint64_t stream_count, + const quic::QuicClock* clock, + quic::QuicAlarmFactory* alarm_factory) { total_streams_ = stream_count; clock_ = clock; if (all_streams_closed()) { @@ -108,7 +108,7 @@ MaybeSetPublishDoneAlarm(); } -void SubscribeRemoteTrack::MaybeSetPublishDoneAlarm() { +void LiveSubscriber::MaybeSetPublishDoneAlarm() { if (currently_open_streams_ == 0 && total_streams_.has_value() && clock_ != nullptr) { quic::QuicTimeDelta timeout = std::min( @@ -120,13 +120,13 @@ } } -void SubscribeRemoteTrack::OnJoiningFetchReady( +void LiveSubscriber::OnJoiningFetchReady( std::unique_ptr<MoqtFetchTask> fetch_task) { fetch_task_ = std::move(fetch_task); fetch_task_->SetObjectAvailableCallback([this]() { FetchObjects(); }); } -void SubscribeRemoteTrack::FetchObjects() { +void LiveSubscriber::FetchObjects() { if (fetch_task_ == nullptr) { return; } @@ -145,7 +145,7 @@ } for (size_t i = 0; i < object.payload.size(); ++i) { if (fetch_object_offset_ > 0 && object.payload[i].empty()) { - QUICHE_BUG(SubscribeRemoteTrack_empty_payload) + QUICHE_BUG(LiveSubscriber_empty_payload) << "Empty payload for partial object " << object.metadata.location; continue; @@ -170,9 +170,8 @@ } } -void SubscribeRemoteTrack::SendObjectAck( - uint64_t group_id, uint64_t object_id, - quic::QuicTimeDelta delta_from_deadline) { +void LiveSubscriber::SendObjectAck(uint64_t group_id, uint64_t object_id, + quic::QuicTimeDelta delta_from_deadline) { request_stream()->SendOrBufferMessageOrFatal( request_stream()->framer()->SerializeObjectAck( {group_id, object_id, delta_from_deadline}));
diff --git a/quiche/quic/moqt/moqt_track.h b/quiche/quic/moqt/moqt_object_subscriber.h similarity index 87% rename from quiche/quic/moqt/moqt_track.h rename to quiche/quic/moqt/moqt_object_subscriber.h index 269ef2e..a5da585 100644 --- a/quiche/quic/moqt/moqt_track.h +++ b/quiche/quic/moqt/moqt_object_subscriber.h
@@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. -// TODO(martinduke): Rename this file to moqt_subscriber.h +// TODO(martinduke): Rename this file to moqt_object_subscriber.h #ifndef QUICHE_QUIC_MOQT_MOQT_TRACK_H_ #define QUICHE_QUIC_MOQT_MOQT_TRACK_H_ @@ -39,21 +39,21 @@ namespace test { class MoqtSessionPeer; -class SubscribeRemoteTrackPeer; +class LiveSubscriberPeer; } // namespace test // State common to both SUBSCRIBE and FETCH upstream. -class RemoteTrack { +class ObjectSubscriber { public: - RemoteTrack(const FullTrackName& full_track_name, uint64_t id, - const MessageParameters& parameters, - MoqtBidiStreamBase* request_stream) + ObjectSubscriber(const FullTrackName& full_track_name, uint64_t id, + const MessageParameters& parameters, + MoqtBidiStreamBase* request_stream) : full_track_name_(full_track_name), request_id_(id), request_stream_(request_stream), parameters_(parameters), weak_ptr_factory_(this) {} - virtual ~RemoteTrack() {} + virtual ~ObjectSubscriber() {} const FullTrackName& full_track_name() const { return full_track_name_; } // If REQUEST_ERROR arrives after OK or an object, it is a protocol violation. @@ -65,7 +65,7 @@ // Is the object one that was requested? virtual bool InWindow(Location sequence) const = 0; - quiche::QuicheWeakPtr<RemoteTrack> weak_ptr() { + quiche::QuicheWeakPtr<ObjectSubscriber> weak_ptr() { return weak_ptr_factory_.Create(); } @@ -93,34 +93,31 @@ bool error_is_allowed_ = true; // Must be last. - quiche::QuicheWeakPtrFactory<RemoteTrack> weak_ptr_factory_; + quiche::QuicheWeakPtrFactory<ObjectSubscriber> weak_ptr_factory_; }; // A track on the peer to which the session has subscribed. -class SubscribeRemoteTrack : public RemoteTrack { +class LiveSubscriber : public ObjectSubscriber { public: // Returns the existing subscription, if present. - using AddCallback = quiche::SingleUseCallback<bool(SubscribeRemoteTrack*)>; - using RemoveCallback = quiche::SingleUseCallback<void(SubscribeRemoteTrack*)>; - SubscribeRemoteTrack(const MoqtSubscribe& subscribe, - SubscribeVisitor* visitor, - MoqtBidiStreamBase* request_stream) - : RemoteTrack(subscribe.full_track_name, subscribe.request_id, - subscribe.parameters, request_stream), + using AddCallback = quiche::SingleUseCallback<bool(LiveSubscriber*)>; + using RemoveCallback = quiche::SingleUseCallback<void(LiveSubscriber*)>; + LiveSubscriber(const MoqtSubscribe& subscribe, SubscribeVisitor* visitor, + MoqtBidiStreamBase* request_stream) + : ObjectSubscriber(subscribe.full_track_name, subscribe.request_id, + subscribe.parameters, request_stream), visitor_(visitor) {} - SubscribeRemoteTrack(const MoqtPublish& publish, SubscribeVisitor* visitor, - MoqtBidiStreamBase* request_stream) - : RemoteTrack(publish.full_track_name, publish.request_id, - publish.parameters, request_stream), + LiveSubscriber(const MoqtPublish& publish, SubscribeVisitor* visitor, + MoqtBidiStreamBase* request_stream) + : ObjectSubscriber(publish.full_track_name, publish.request_id, + publish.parameters, request_stream), visitor_(visitor) { track_alias_.emplace(publish.track_alias); } - ~SubscribeRemoteTrack() override; + ~LiveSubscriber() override; void OnObjectOrOk(const SubscribeOkData& data); - void OnObjectOrOk() override { - RemoteTrack::OnObjectOrOk(); - } + void OnObjectOrOk() override { ObjectSubscriber::OnObjectOrOk(); } std::optional<uint64_t> track_alias() const { return track_alias_; } // Returns false if the callback returns false, meaning the session has been // destroyed. @@ -133,7 +130,7 @@ quic::QuicAlarmFactory* alarm_factory); // The application can request a Joining FETCH but also for FETCH objects to - // be delivered via SubscribeRemoteTrack::Visitor::OnObjectFragment(). When + // be delivered via LiveSubscriber::Visitor::OnObjectFragment(). When // this occurs, the session passes the FetchTask here to handle incoming // FETCH objects to pipe directly into the visitor. void OnJoiningFetchReady(std::unique_ptr<MoqtFetchTask> fetch_task); @@ -167,15 +164,15 @@ private: friend class test::MoqtSessionPeer; - friend class test::SubscribeRemoteTrackPeer; + friend class test::LiveSubscriberPeer; class PublishDoneDelegate : public quic::QuicAlarm::DelegateWithoutContext { public: - PublishDoneDelegate(quiche::QuicheWeakPtr<RemoteTrack> subscribe) + PublishDoneDelegate(quiche::QuicheWeakPtr<ObjectSubscriber> subscribe) : subscribe_(subscribe) {} void OnAlarm() override { - RemoteTrack* subscribe = subscribe_.GetIfAvailable(); + ObjectSubscriber* subscribe = subscribe_.GetIfAvailable(); if (subscribe == nullptr) { return; } @@ -183,7 +180,7 @@ } private: - quiche::QuicheWeakPtr<RemoteTrack> subscribe_; + quiche::QuicheWeakPtr<ObjectSubscriber> subscribe_; }; void MaybeSetPublishDoneAlarm(); @@ -223,14 +220,14 @@ // Class for upstream FETCH. It will notify the application using |callback| // when a FETCH_OK or REQUEST_ERROR is received. using RemoveFetchCallback = quiche::SingleUseCallback<void()>; -class UpstreamFetch : public RemoteTrack { +class UpstreamFetch : public ObjectSubscriber { public: // Standalone Fetch constructor UpstreamFetch(const MoqtFetch& fetch, const StandaloneFetch standalone, FetchResponseCallback callback, RemoveFetchCallback delete_callback) - : RemoteTrack(standalone.full_track_name, fetch.request_id, - fetch.parameters, /*request_stream=*/nullptr), + : ObjectSubscriber(standalone.full_track_name, fetch.request_id, + fetch.parameters, /*request_stream=*/nullptr), group_order_(fetch.parameters.group_order.value_or( MoqtDeliveryOrder::kAscending)), start_(standalone.start_location), @@ -243,8 +240,8 @@ UpstreamFetch(const MoqtFetch& fetch, FullTrackName full_track_name, FetchResponseCallback callback, RemoveFetchCallback delete_callback) - : RemoteTrack(full_track_name, fetch.request_id, fetch.parameters, - /*request_stream=*/nullptr), + : ObjectSubscriber(full_track_name, fetch.request_id, fetch.parameters, + /*request_stream=*/nullptr), group_order_(fetch.parameters.group_order.value_or( MoqtDeliveryOrder::kAscending)), relative_groups_( @@ -258,8 +255,8 @@ JoiningFetchAbsolute absolute_joining, FetchResponseCallback callback, RemoveFetchCallback delete_callback) - : RemoteTrack(full_track_name, fetch.request_id, fetch.parameters, - /*request_stream=*/nullptr), + : ObjectSubscriber(full_track_name, fetch.request_id, fetch.parameters, + /*request_stream=*/nullptr), group_order_(fetch.parameters.group_order.value_or( MoqtDeliveryOrder::kAscending)), start_(Location(absolute_joining.joining_start, 0)),
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_object_subscriber_test.cc similarity index 91% rename from quiche/quic/moqt/moqt_track_test.cc rename to quiche/quic/moqt/moqt_object_subscriber_test.cc index bb50891..490001b 100644 --- a/quiche/quic/moqt/moqt_track_test.cc +++ b/quiche/quic/moqt/moqt_object_subscriber_test.cc
@@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. -#include "quiche/quic/moqt/moqt_track.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include <memory> #include <optional> @@ -43,33 +43,33 @@ } // namespace -class SubscribeRemoteTrackPeer { +class LiveSubscriberPeer { public: - static MoqtFetchTask* GetFetchTask(SubscribeRemoteTrack* track) { + static MoqtFetchTask* GetFetchTask(LiveSubscriber* track) { return track->fetch_task_.get(); } - static quic::QuicAlarm* GetPublishDoneAlarm(SubscribeRemoteTrack* track) { + static quic::QuicAlarm* GetPublishDoneAlarm(LiveSubscriber* track) { return track->publish_done_alarm_.get(); } }; -class SubscribeRemoteTrackTest : public quic::test::QuicTest { +class LiveSubscriberTest : public quic::test::QuicTest { public: - SubscribeRemoteTrackTest() : track_(subscribe_, &visitor_, &stream_) { + LiveSubscriberTest() : track_(subscribe_, &visitor_, &stream_) { stream_.BindStream(&wt_stream_); } - MockSubscribeRemoteTrackVisitor visitor_; + MockLiveSubscriberVisitor visitor_; MoqtSubscribe subscribe_ = {/*request_id=*/1, FullTrackName("foo", "bar"), MessageParameters(Location(2, 0))}; MockBidiStream stream_; webtransport::test::MockStream wt_stream_; - SubscribeRemoteTrack track_; + LiveSubscriber track_; quic::MockClock clock_; quic::test::MockAlarmFactory alarm_factory_; }; -TEST_F(SubscribeRemoteTrackTest, Queries) { +TEST_F(LiveSubscriberTest, Queries) { EXPECT_EQ(track_.full_track_name(), FullTrackName("foo", "bar")); EXPECT_EQ(track_.request_id(), 1); EXPECT_FALSE(track_.track_alias().has_value()); @@ -79,18 +79,18 @@ EXPECT_EQ(track_.track_alias(), 1); } -TEST_F(SubscribeRemoteTrackTest, AllowError) { +TEST_F(LiveSubscriberTest, AllowError) { EXPECT_TRUE(track_.ErrorIsAllowed()); track_.OnObjectOrOk(); EXPECT_FALSE(track_.ErrorIsAllowed()); } -TEST_F(SubscribeRemoteTrackTest, Windows) { +TEST_F(LiveSubscriberTest, Windows) { EXPECT_TRUE(track_.InWindow(Location(2, 0))); EXPECT_FALSE(track_.InWindow(Location(1, 25))); } -TEST_F(SubscribeRemoteTrackTest, OnPublishDoneReadyToClose) { +TEST_F(LiveSubscriberTest, OnPublishDoneReadyToClose) { track_.OnStreamOpened(); track_.OnStreamClosed(true, std::nullopt); EXPECT_CALL(visitor_, OnPublishDone); @@ -98,7 +98,7 @@ track_.OnPublishDone(1, &clock_, &alarm_factory_); } -TEST_F(SubscribeRemoteTrackTest, OnPublishDoneAllStreamsCloseLater) { +TEST_F(LiveSubscriberTest, OnPublishDoneAllStreamsCloseLater) { track_.OnStreamOpened(); EXPECT_CALL(visitor_, OnPublishDone).Times(0); EXPECT_CALL(wt_stream_, Writev).Times(0); @@ -110,14 +110,13 @@ track_.OnStreamClosed(true, std::nullopt); } -TEST_F(SubscribeRemoteTrackTest, OnPublishDoneTimesOut) { +TEST_F(LiveSubscriberTest, OnPublishDoneTimesOut) { track_.OnStreamOpened(); EXPECT_CALL(visitor_, OnPublishDone).Times(0); EXPECT_CALL(wt_stream_, Writev).Times(0); track_.OnPublishDone(2, &clock_, &alarm_factory_); track_.OnStreamClosed(true, std::nullopt); // No streams are open; timer set. - quic::QuicAlarm* alarm = - SubscribeRemoteTrackPeer::GetPublishDoneAlarm(&track_); + quic::QuicAlarm* alarm = LiveSubscriberPeer::GetPublishDoneAlarm(&track_); EXPECT_NE(alarm, nullptr); EXPECT_TRUE(alarm->IsSet()); EXPECT_CALL(visitor_, OnPublishDone); @@ -125,7 +124,7 @@ alarm_factory_.FireAlarm(alarm); } -TEST_F(SubscribeRemoteTrackTest, JoiningFetchMultiObject) { +TEST_F(LiveSubscriberTest, JoiningFetchMultiObject) { auto fetch_task = std::make_unique<MockFetchTask>(); MockFetchTask* task_ptr = fetch_task.get(); track_.OnJoiningFetchReady(std::move(fetch_task)); @@ -154,14 +153,14 @@ }) .WillOnce(testing::Return(MoqtFetchTask::GetNextObjectResult::kPending)); task_ptr->CallObjectsAvailableCallback(); - EXPECT_NE(SubscribeRemoteTrackPeer::GetFetchTask(&track_), nullptr); + EXPECT_NE(LiveSubscriberPeer::GetFetchTask(&track_), nullptr); EXPECT_CALL(*task_ptr, GetNextObject) .WillOnce(testing::Return(MoqtFetchTask::GetNextObjectResult::kEof)); task_ptr->CallObjectsAvailableCallback(); - EXPECT_EQ(SubscribeRemoteTrackPeer::GetFetchTask(&track_), nullptr); + EXPECT_EQ(LiveSubscriberPeer::GetFetchTask(&track_), nullptr); } -TEST_F(SubscribeRemoteTrackTest, JoiningFetchFragmented) { +TEST_F(LiveSubscriberTest, JoiningFetchFragmented) { auto fetch_task = std::make_unique<MockFetchTask>(); MockFetchTask* task_ptr = fetch_task.get(); track_.OnJoiningFetchReady(std::move(fetch_task)); @@ -192,7 +191,7 @@ task_ptr->CallObjectsAvailableCallback(); } -TEST_F(SubscribeRemoteTrackTest, JoiningFetchEmptyPayload) { +TEST_F(LiveSubscriberTest, JoiningFetchEmptyPayload) { auto fetch_task = std::make_unique<MockFetchTask>(); MockFetchTask* task_ptr = fetch_task.get(); track_.OnJoiningFetchReady(std::move(fetch_task)); @@ -214,16 +213,16 @@ task_ptr->CallObjectsAvailableCallback(); } -TEST_F(SubscribeRemoteTrackTest, JoiningFetchError) { +TEST_F(LiveSubscriberTest, JoiningFetchError) { auto fetch_task = std::make_unique<MockFetchTask>(); MockFetchTask* task_ptr = fetch_task.get(); track_.OnJoiningFetchReady(std::move(fetch_task)); - EXPECT_NE(SubscribeRemoteTrackPeer::GetFetchTask(&track_), nullptr); + EXPECT_NE(LiveSubscriberPeer::GetFetchTask(&track_), nullptr); EXPECT_CALL(*task_ptr, GetNextObject) .WillOnce(testing::Return(MoqtFetchTask::GetNextObjectResult::kError)); task_ptr->CallObjectsAvailableCallback(); - EXPECT_EQ(SubscribeRemoteTrackPeer::GetFetchTask(&track_), nullptr); + EXPECT_EQ(LiveSubscriberPeer::GetFetchTask(&track_), nullptr); } class UpstreamFetchTest : public quic::test::QuicTest {
diff --git a/quiche/quic/moqt/moqt_publish_stream.cc b/quiche/quic/moqt/moqt_publish_stream.cc index 1ffe56c..191e43a 100644 --- a/quiche/quic/moqt/moqt_publish_stream.cc +++ b/quiche/quic/moqt/moqt_publish_stream.cc
@@ -19,18 +19,18 @@ #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" -#include "quiche/quic/moqt/moqt_subscription.h" -#include "quiche/quic/moqt/moqt_track.h" namespace moqt { -MoqtPublishPublisherStream::MoqtPublishPublisherStream( +MoqtPublishRequestStream::MoqtPublishRequestStream( MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, - SubscriptionPublisher::RemoveCallback stream_deleted_callback, + LivePublisher::RemoveCallback stream_deleted_callback, SessionErrorCallback session_error_callback, MoqtResponseCallback response_callback) : MoqtBidiStreamBase(framer, message_parser, @@ -38,14 +38,14 @@ response_callback_(std::move(response_callback)), stream_deleted_callback_(std::move(stream_deleted_callback)) {} -MoqtPublishPublisherStream::~MoqtPublishPublisherStream() { +MoqtPublishRequestStream::~MoqtPublishRequestStream() { if (publisher_ != nullptr) { publisher_->IgnoreResetAllStreams(); } Detach(); } -void MoqtPublishPublisherStream::OnStreamBound() { +void MoqtPublishRequestStream::OnStreamBound() { stream_parser()->set_allow_fin(true); publisher_->parameters().largest_object = publisher_->publisher().largest_location(); @@ -59,7 +59,7 @@ publisher_->publisher().extensions().default_publisher_group_order(); } -absl::Status MoqtPublishPublisherStream::OnRawControlMessage( +absl::Status MoqtPublishRequestStream::OnRawControlMessage( const MoqtRawControlMessage& message) { return ControlMessageDispatcher::DispatchControlMessage( *this, message_parser(), message, "publish publisher"); @@ -67,7 +67,7 @@ // TODO(martinduke): When we allow the publisher to send REQUEST_UPDATE, // REQUEST_OK and REQUEST_ERROR processing need to check the request ID. -absl::Status MoqtPublishPublisherStream::OnControlMessage( +absl::Status MoqtPublishRequestStream::OnControlMessage( const MoqtRequestOk& message) { if (message.request_id != publisher_->request_id()) { OnFatalError(absl::InvalidArgumentError( @@ -83,7 +83,7 @@ return absl::OkStatus(); } -absl::Status MoqtPublishPublisherStream::OnControlMessage( +absl::Status MoqtPublishRequestStream::OnControlMessage( const MoqtRequestError& message) { if (message.request_id != publisher_->request_id()) { OnFatalError(absl::InvalidArgumentError( @@ -95,7 +95,7 @@ return absl::OkStatus(); } -absl::Status MoqtPublishPublisherStream::OnControlMessage( +absl::Status MoqtPublishRequestStream::OnControlMessage( const MoqtRequestUpdate& message) { MessageParameters in_parameters = message.parameters, out_parameters; out_parameters.largest_object = publisher_->publisher().largest_location(); @@ -108,15 +108,15 @@ return absl::OkStatus(); } -MoqtPublishSubscriberStream::MoqtPublishSubscriberStream( +MoqtPublishResponseStream::MoqtPublishResponseStream( MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, const quic::QuicClock* absl_nonnull clock, quic::QuicAlarmFactory* absl_nonnull alarm_factory, SessionErrorCallback session_error_callback, const MoqtIncomingPublishCallback* absl_nonnull incoming_publish_callback, - SubscribeRemoteTrack::AddCallback add_callback, - SubscribeRemoteTrack::RemoveCallback remove_callback) + LiveSubscriber::AddCallback add_callback, + LiveSubscriber::RemoveCallback remove_callback) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), clock_(clock), @@ -126,19 +126,19 @@ remove_callback_(std::move(remove_callback)), weak_ptr_factory_(this) {} -absl::Status MoqtPublishSubscriberStream::OnRawControlMessage( +absl::Status MoqtPublishResponseStream::OnRawControlMessage( const MoqtRawControlMessage& message) { return ControlMessageDispatcher::DispatchControlMessage( *this, message_parser(), message, "publish subscriber"); } -absl::Status MoqtPublishSubscriberStream::OnControlMessage( +absl::Status MoqtPublishResponseStream::OnControlMessage( const MoqtPublish& message) { if (add_callback_ == nullptr) { // Two PUBLISH messages for the same stream. return absl::InvalidArgumentError("Multiple PUBLISH on the same stream"); } - subscriber_ = std::make_unique<SubscribeRemoteTrack>(message, nullptr, this); + subscriber_ = std::make_unique<LiveSubscriber>(message, nullptr, this); if (!std::move(add_callback_)(subscriber_.get())) { add_callback_ = nullptr; return SendRequestError(message.request_id, @@ -154,7 +154,7 @@ [weakptr = weak_ptr_factory_.Create(), request_id = message.request_id]( const std::variant<MessageParameters, MoqtRequestErrorInfo> response) { - MoqtPublishSubscriberStream* stream = weakptr.GetIfAvailable(); + MoqtPublishResponseStream* stream = weakptr.GetIfAvailable(); if (stream == nullptr) { return; } @@ -191,7 +191,7 @@ return absl::OkStatus(); } -absl::Status MoqtPublishSubscriberStream::OnControlMessage( +absl::Status MoqtPublishResponseStream::OnControlMessage( const MoqtRequestUpdate& message) { if (subscriber_ == nullptr) { // Stream is already closing. @@ -202,19 +202,19 @@ return absl::OkStatus(); } -absl::Status MoqtPublishSubscriberStream::OnControlMessage( +absl::Status MoqtPublishResponseStream::OnControlMessage( const MoqtRequestOk& message) { // TODO(martinduke): Implement REQUEST_UPDATE. return absl::OkStatus(); } -absl::Status MoqtPublishSubscriberStream::OnControlMessage( +absl::Status MoqtPublishResponseStream::OnControlMessage( const MoqtRequestError& message) { // TODO(martinduke): Implement REQUEST_UPDATE. return absl::OkStatus(); } -absl::Status MoqtPublishSubscriberStream::OnControlMessage( +absl::Status MoqtPublishResponseStream::OnControlMessage( const MoqtPublishDone& message) { if (subscriber_ == nullptr) { // PUBLISH_DONE can be sent before the subscriber rejects the track.
diff --git a/quiche/quic/moqt/moqt_publish_stream.h b/quiche/quic/moqt/moqt_publish_stream.h index 43e6dbe..bb0202d 100644 --- a/quiche/quic/moqt/moqt_publish_stream.h +++ b/quiche/quic/moqt/moqt_publish_stream.h
@@ -17,29 +17,29 @@ #include "quiche/quic/moqt/moqt_bidi_stream.h" #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" -#include "quiche/quic/moqt/moqt_subscription.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/common/quiche_weak_ptr.h" namespace moqt { -class MoqtPublishPublisherStream : public MoqtBidiStreamBase { +class MoqtPublishRequestStream : public MoqtBidiStreamBase { public: // Order of operations: // 1. Call this constructor // 2. Call SetPublisher() // 3. Call Webtransport::Stream::SetVisitor() // 4. Call this::BindStream() - MoqtPublishPublisherStream( + MoqtPublishRequestStream( MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, - SubscriptionPublisher::RemoveCallback stream_deleted_callback, + LivePublisher::RemoveCallback stream_deleted_callback, SessionErrorCallback session_error_callback, MoqtResponseCallback response_callback); - ~MoqtPublishPublisherStream(); + ~MoqtPublishRequestStream(); // MoqtBidiStreamBase overrides. void OnStreamBound() override; @@ -53,7 +53,7 @@ return absl::OkStatus(); } - void SetPublisher(std::unique_ptr<SubscriptionPublisher> publisher) { + void SetPublisher(std::unique_ptr<LivePublisher> publisher) { publisher_ = std::move(publisher); } @@ -61,7 +61,7 @@ if (stream_deleted_callback_ == nullptr) { return; } - SubscriptionPublisher::RemoveCallback callback = + LivePublisher::RemoveCallback callback = std::move(stream_deleted_callback_); stream_deleted_callback_ = nullptr; std::move(callback)(publisher_.get()); @@ -71,23 +71,23 @@ private: MoqtResponseCallback response_callback_; - std::unique_ptr<SubscriptionPublisher> publisher_; + std::unique_ptr<LivePublisher> publisher_; absl::flat_hash_map<uint64_t, MoqtResponseCallback> pending_updates_; - SubscriptionPublisher::RemoveCallback stream_deleted_callback_; + LivePublisher::RemoveCallback stream_deleted_callback_; }; -class MoqtPublishSubscriberStream : public MoqtBidiStreamBase { +class MoqtPublishResponseStream : public MoqtBidiStreamBase { public: - MoqtPublishSubscriberStream( + MoqtPublishResponseStream( MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, const quic::QuicClock* absl_nonnull clock, quic::QuicAlarmFactory* absl_nonnull alarm_factory, SessionErrorCallback session_error_callback, const MoqtIncomingPublishCallback* absl_nonnull incoming_publish_callback, - SubscribeRemoteTrack::AddCallback add_callback, - SubscribeRemoteTrack::RemoveCallback remove_callback); - ~MoqtPublishSubscriberStream() { Detach(); } + LiveSubscriber::AddCallback add_callback, + LiveSubscriber::RemoveCallback remove_callback); + ~MoqtPublishResponseStream() { Detach(); } // MoqtBidiStreamBase overrides. void OnStreamBound() override { @@ -102,12 +102,11 @@ absl::Status OnControlMessage(const MoqtRequestError& message); absl::Status OnControlMessage(const MoqtPublishDone& message); - SubscribeRemoteTrack* track() { return subscriber_.get(); } + LiveSubscriber* track() { return subscriber_.get(); } void Detach() override { if (remove_callback_ != nullptr) { - SubscribeRemoteTrack::RemoveCallback callback = - std::move(remove_callback_); + LiveSubscriber::RemoveCallback callback = std::move(remove_callback_); remove_callback_ = nullptr; std::move(callback)(subscriber_.get()); } @@ -118,14 +117,14 @@ uint64_t request_id_; SubscribeVisitor* absl_nullable subscribe_visitor_ = nullptr; bool in_destructor_ = false; - std::unique_ptr<SubscribeRemoteTrack> subscriber_; + std::unique_ptr<LiveSubscriber> subscriber_; absl::flat_hash_map<uint64_t, MoqtResponseCallback> pending_updates_; const quic::QuicClock* clock_; quic::QuicAlarmFactory* alarm_factory_; const MoqtIncomingPublishCallback* incoming_publish_callback_; - SubscribeRemoteTrack::AddCallback add_callback_; - SubscribeRemoteTrack::RemoveCallback remove_callback_; - quiche::QuicheWeakPtrFactory<MoqtPublishSubscriberStream> weak_ptr_factory_; + LiveSubscriber::AddCallback add_callback_; + LiveSubscriber::RemoveCallback remove_callback_; + quiche::QuicheWeakPtrFactory<MoqtPublishResponseStream> weak_ptr_factory_; }; } // namespace moqt
diff --git a/quiche/quic/moqt/moqt_publish_stream_test.cc b/quiche/quic/moqt/moqt_publish_stream_test.cc index 9de992f..e8f596d 100644 --- a/quiche/quic/moqt/moqt_publish_stream_test.cc +++ b/quiche/quic/moqt/moqt_publish_stream_test.cc
@@ -20,15 +20,15 @@ #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/quic/moqt/moqt_session_interface.h" -#include "quiche/quic/moqt/moqt_subscription.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/test_tools/mock_moqt_session.h" #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h" @@ -43,18 +43,16 @@ namespace moqt::test { -class SubscriptionPublisherPeer { +class LivePublisherPeer { public: - static const MessageParameters& parameters( - const SubscriptionPublisher& publisher) { + static const MessageParameters& parameters(const LivePublisher& publisher) { return publisher.parameters_; } }; -class SubscribeRemoteTrackPeer { +class LiveSubscriberPeer { public: - static const MessageParameters& parameters( - const SubscribeRemoteTrack& track) { + static const MessageParameters& parameters(const LiveSubscriber& track) { return track.const_parameters(); } }; @@ -70,28 +68,28 @@ constexpr uint64_t kTrackAlias = 10; const FullTrackName kTrackName("foo", "bar"); -class MoqtPublishPublisherStreamTest : public quiche::test::QuicheTest { +class MoqtPublishRequestStreamTest : public quiche::test::QuicheTest { public: - MoqtPublishPublisherStreamTest() + MoqtPublishRequestStreamTest() : framer_(/*using_webtrans=*/true, quic::Perspective::IS_CLIENT), message_parser_(kDefaultMoqtVersion, /*uses_web_transport=*/true, quic::Perspective::IS_CLIENT), track_publisher_(std::make_shared<TestTrackPublisher>(kTrackName)) { // Construct the stream visitor. - stream_ = std::make_unique<MoqtPublishPublisherStream>( + stream_ = std::make_unique<MoqtPublishRequestStream>( &framer_, message_parser_, deleted_callback_.AsStdFunction(), error_callback_.AsStdFunction(), [this](std::variant<MessageParameters, MoqtRequestErrorInfo> response) { response_ = response; }); - // Construct the SubscriptionPublisher. + // Construct the LivePublisher. parameters_.set_forward(true); parameters_.delivery_timeout = quic::QuicTimeDelta::FromSeconds(1); parameters_.group_order = MoqtDeliveryOrder::kAscending; EXPECT_CALL(visitor_, session).WillRepeatedly(Return(&webtrans_)); - auto publisher = std::make_unique<SubscriptionPublisher>( + auto publisher = std::make_unique<LivePublisher>( framer_, track_publisher_, stream_.get(), kRequestId, kTrackAlias, parameters_, visitor_.weak_ptr_factory_.Create(), /*is_publish=*/true); @@ -104,7 +102,7 @@ MoqtControlMessageParser message_parser_; webtransport::test::MockStream mock_stream_; std::shared_ptr<TestTrackPublisher> track_publisher_; - testing::MockFunction<void(SubscriptionPublisher*)> deleted_callback_; + testing::MockFunction<void(LivePublisher*)> deleted_callback_; testing::StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>> error_callback_; MockSessionToPublisherInterface visitor_; @@ -113,20 +111,20 @@ MoqtTraceRecorder trace_recorder_; MessageParameters parameters_; - std::unique_ptr<MoqtPublishPublisherStream> stream_; - SubscriptionPublisher* publisher_; // Raw pointer + std::unique_ptr<MoqtPublishRequestStream> stream_; + LivePublisher* publisher_; // Raw pointer std::optional<std::variant<MessageParameters, MoqtRequestErrorInfo>> response_; }; -TEST_F(MoqtPublishPublisherStreamTest, OnStreamBoundSendsPublish) { +TEST_F(MoqtPublishRequestStreamTest, OnStreamBoundSendsPublish) { EXPECT_CALL(mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kPublish), _)) .WillOnce(Return(absl::OkStatus())); stream_->BindStream(&mock_stream_); // Calls OnStreamBound } -TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestOk) { +TEST_F(MoqtPublishRequestStreamTest, ReceiveRequestOk) { EXPECT_CALL(mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kPublish), _)) .WillOnce(Return(absl::OkStatus())); @@ -148,14 +146,14 @@ // Verify publisher parameters were updated. const MessageParameters& pub_params = - SubscriptionPublisherPeer::parameters(*publisher_); + LivePublisherPeer::parameters(*publisher_); EXPECT_EQ(pub_params.delivery_timeout, request_ok.parameters.delivery_timeout); // Group order cannot be updated. EXPECT_EQ(pub_params.group_order, parameters_.group_order); } -TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestError) { +TEST_F(MoqtPublishRequestStreamTest, ReceiveRequestError) { EXPECT_CALL(mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kPublish), _)) .WillOnce(Return(absl::OkStatus())); @@ -177,7 +175,7 @@ EXPECT_EQ(resp_error.reason_phrase, request_error.reason_phrase); } -TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestUpdate) { +TEST_F(MoqtPublishRequestStreamTest, ReceiveRequestUpdate) { EXPECT_CALL(mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kPublish), _)) .WillOnce(Return(absl::OkStatus())); @@ -202,7 +200,7 @@ // Verify publisher parameters were updated. const MessageParameters& pub_params = - SubscriptionPublisherPeer::parameters(*publisher_); + LivePublisherPeer::parameters(*publisher_); EXPECT_EQ(pub_params.delivery_timeout, request_update.parameters.delivery_timeout); EXPECT_EQ(pub_params.subscriber_priority, @@ -216,7 +214,7 @@ EXPECT_EQ(pub_params.subscription_filter->start(), Location(1, 3)); } -TEST_F(MoqtPublishPublisherStreamTest, ReceiveObjectAck) { +TEST_F(MoqtPublishRequestStreamTest, ReceiveObjectAck) { EXPECT_CALL(mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kPublish), _)) .WillOnce(Return(absl::OkStatus())); @@ -232,7 +230,7 @@ QUICHE_EXPECT_OK(stream_->OnControlMessage(ack)); } -TEST_F(MoqtPublishPublisherStreamTest, Detach) { +TEST_F(MoqtPublishRequestStreamTest, Detach) { EXPECT_CALL(deleted_callback_, Call(publisher_)); stream_->Detach(); // Verifying second detach is a no-op @@ -240,15 +238,15 @@ stream_->Detach(); } -class MoqtPublishSubscriberStreamTest : public quiche::test::QuicheTest { +class MoqtPublishResponseStreamTest : public quiche::test::QuicheTest { public: - MoqtPublishSubscriberStreamTest() + MoqtPublishResponseStreamTest() : framer_(/*using_webtrans=*/true, quic::Perspective::IS_SERVER), message_parser_(kDefaultMoqtVersion, /*uses_web_transport=*/true, quic::Perspective::IS_SERVER), incoming_publish_callback_( incoming_publish_callback_mock_.AsStdFunction()) { - stream_ = std::make_unique<MoqtPublishSubscriberStream>( + stream_ = std::make_unique<MoqtPublishResponseStream>( &framer_, message_parser_, &mock_clock_, &mock_alarm_factory_, error_callback_.AsStdFunction(), &incoming_publish_callback_, mock_add_callback_.AsStdFunction(), @@ -275,16 +273,16 @@ incoming_publish_callback_mock_; MoqtIncomingPublishCallback incoming_publish_callback_; - testing::MockFunction<bool(SubscribeRemoteTrack*)> mock_add_callback_; - testing::MockFunction<void(SubscribeRemoteTrack*)> mock_remove_callback_; + testing::MockFunction<bool(LiveSubscriber*)> mock_add_callback_; + testing::MockFunction<void(LiveSubscriber*)> mock_remove_callback_; - StrictMock<MockSubscribeRemoteTrackVisitor> mock_subscribe_visitor_; + StrictMock<MockLiveSubscriberVisitor> mock_subscribe_visitor_; MoqtResponseCallback captured_response_callback_; webtransport::test::MockStream mock_stream_; - std::unique_ptr<MoqtPublishSubscriberStream> stream_; + std::unique_ptr<MoqtPublishResponseStream> stream_; }; -TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishAndAccept) { +TEST_F(MoqtPublishResponseStreamTest, ReceivePublishAndAccept) { EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _)) .WillOnce( [](const FullTrackName&, @@ -297,9 +295,9 @@ captured_response_callback_ = std::move(callback); return &mock_subscribe_visitor_; }); - SubscribeRemoteTrack* captured_subscriber = nullptr; + LiveSubscriber* captured_subscriber = nullptr; EXPECT_CALL(mock_add_callback_, Call(NotNull())) - .WillOnce([&](SubscribeRemoteTrack* subscriber) { + .WillOnce([&](LiveSubscriber* subscriber) { captured_subscriber = subscriber; return true; }); @@ -322,12 +320,12 @@ // Verify subscriber parameters were updated. const MessageParameters& sub_params = - SubscribeRemoteTrackPeer::parameters(*captured_subscriber); + LiveSubscriberPeer::parameters(*captured_subscriber); EXPECT_EQ(sub_params.delivery_timeout, response_parameters.delivery_timeout); EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishAndReject) { +TEST_F(MoqtPublishResponseStreamTest, ReceivePublishAndReject) { MoqtPublish publish = DefaultPublish(); // Callback returns nullptr (rejection). EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true)); @@ -340,7 +338,7 @@ QUICHE_EXPECT_OK(stream_->OnControlMessage(publish)); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceiveTwoPublishOnStream) { +TEST_F(MoqtPublishResponseStreamTest, ReceiveTwoPublishOnStream) { MoqtPublish publish = DefaultPublish(); EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true)); EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) @@ -361,7 +359,7 @@ EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName)); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDuplicate) { +TEST_F(MoqtPublishResponseStreamTest, ReceivePublishDuplicate) { MoqtPublish publish = DefaultPublish(); publish.request_id = kRequestId + 2; publish.full_track_name = kTrackName; @@ -375,13 +373,13 @@ QUICHE_EXPECT_OK(stream_->OnControlMessage(publish)); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceiveRequestUpdate) { +TEST_F(MoqtPublishResponseStreamTest, ReceiveRequestUpdate) { MoqtPublish publish = DefaultPublish(); EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) .WillOnce(Return(&mock_subscribe_visitor_)); - SubscribeRemoteTrack* captured_subscriber = nullptr; + LiveSubscriber* captured_subscriber = nullptr; EXPECT_CALL(mock_add_callback_, Call(NotNull())) - .WillOnce([&](SubscribeRemoteTrack* track) { + .WillOnce([&](LiveSubscriber* track) { captured_subscriber = track; return true; }); @@ -407,13 +405,13 @@ // Verify subscriber parameters were updated. ASSERT_NE(captured_subscriber, nullptr); const MessageParameters& sub_params = - SubscribeRemoteTrackPeer::parameters(*captured_subscriber); + LiveSubscriberPeer::parameters(*captured_subscriber); EXPECT_EQ(sub_params.delivery_timeout, request_update.parameters.delivery_timeout); EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName)); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDone) { +TEST_F(MoqtPublishResponseStreamTest, ReceivePublishDone) { MoqtPublish publish = DefaultPublish(); EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) .WillOnce(Return(&mock_subscribe_visitor_)); @@ -443,7 +441,7 @@ QUICHE_EXPECT_OK(stream_->OnControlMessage(publish_done)); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishAndRejectCallback) { +TEST_F(MoqtPublishResponseStreamTest, ReceivePublishAndRejectCallback) { MoqtPublish publish = DefaultPublish(); EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) .WillOnce([this](const FullTrackName&, const MessageParameters&, @@ -471,7 +469,7 @@ std::move(captured_response_callback_)(error_info); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDoneOnRejectedStream) { +TEST_F(MoqtPublishResponseStreamTest, ReceivePublishDoneOnRejectedStream) { MoqtPublish publish = DefaultPublish(); // Callback returns nullptr (rejection). EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true)); @@ -491,7 +489,7 @@ QUICHE_EXPECT_OK(stream_->OnControlMessage(publish_done)); } -TEST_F(MoqtPublishSubscriberStreamTest, DuplicatePublishOnSameStream) { +TEST_F(MoqtPublishResponseStreamTest, DuplicatePublishOnSameStream) { MoqtPublish publish = DefaultPublish(); EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true)); EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) @@ -515,7 +513,7 @@ EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName)); } -TEST_F(MoqtPublishSubscriberStreamTest, DuplicatePublishOnDifferentStreams) { +TEST_F(MoqtPublishResponseStreamTest, DuplicatePublishOnDifferentStreams) { MoqtPublish publish1 = DefaultPublish(); EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true)); EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) @@ -530,8 +528,8 @@ // Second stream testing::MockFunction<void(MoqtError, absl::string_view)> error_callback2; - testing::MockFunction<bool(SubscribeRemoteTrack*)> mock_add_callback2; - testing::MockFunction<void(SubscribeRemoteTrack*)> mock_remove_callback2; + testing::MockFunction<bool(LiveSubscriber*)> mock_add_callback2; + testing::MockFunction<void(LiveSubscriber*)> mock_remove_callback2; testing::MockFunction<SubscribeVisitor*( const FullTrackName&, const MessageParameters&, const TrackExtensions&, MoqtResponseCallback)> @@ -540,7 +538,7 @@ incoming_publish_callback_mock2.AsStdFunction(); webtransport::test::MockStream mock_stream2; - MoqtPublishSubscriberStream stream2( + MoqtPublishResponseStream stream2( &framer_, message_parser_, &mock_clock_, &mock_alarm_factory_, error_callback2.AsStdFunction(), &incoming_publish_callback2, mock_add_callback2.AsStdFunction(), @@ -572,7 +570,7 @@ EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName)); } -TEST_F(MoqtPublishSubscriberStreamTest, ReceiveRequestOkAndErrorTodo) { +TEST_F(MoqtPublishResponseStreamTest, ReceiveRequestOkAndErrorTodo) { MoqtRequestOk request_ok; QUICHE_EXPECT_OK(stream_->OnControlMessage(request_ok)); @@ -580,7 +578,7 @@ QUICHE_EXPECT_OK(stream_->OnControlMessage(request_error)); } -TEST_F(MoqtPublishSubscriberStreamTest, TrackAndDetach) { +TEST_F(MoqtPublishResponseStreamTest, TrackAndDetach) { MoqtPublish publish = DefaultPublish(); EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _)) .WillOnce(Return(&mock_subscribe_visitor_));
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index 9dd7d62..d03dc55 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -33,10 +33,12 @@ #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" #include "quiche/quic/moqt/moqt_namespace_stream.h" #include "quiche/quic/moqt/moqt_object.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_publish_stream.h" @@ -44,8 +46,6 @@ #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/quic/moqt/moqt_session_interface.h" #include "quiche/quic/moqt/moqt_subscribe_stream.h" -#include "quiche/quic/moqt/moqt_subscription.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/moqt_uni_stream.h" #include "quiche/quic/platform/api/quic_logging.h" @@ -216,7 +216,7 @@ << message.object_id << " priority " << message.publisher_priority << " length " << payload->size(); - SubscribeRemoteTrack* track = SubscribeByAlias(message.track_alias); + LiveSubscriber* track = SubscribeByAlias(message.track_alias); if (track == nullptr) { return; } @@ -313,8 +313,8 @@ "SUBSCRIBE_NAMESPACE already outstanding for namespace"}); return nullptr; } - std::unique_ptr<MoqtNamespaceSubscriberStream> state = - std::make_unique<MoqtNamespaceSubscriberStream>( + std::unique_ptr<MoqtSubscribeNamespaceRequestStream> state = + std::make_unique<MoqtSubscribeNamespaceRequestStream>( &framer_, ControlMessageParser(), next_request_id_, [weakptr = GetWeakPtr()](const TrackNamespace& prefix) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); @@ -330,7 +330,7 @@ } }, std::move(response_callback)); - MoqtNamespaceSubscriberStream* state_ptr = state.get(); + MoqtSubscribeNamespaceRequestStream* state_ptr = state.get(); if (session_->CanOpenNextOutgoingBidirectionalStream()) { webtransport::Stream* stream = session_->OpenOutgoingBidirectionalStream(); state->BindStream(stream); @@ -503,7 +503,7 @@ session->Error(code, reason); }, name, visitor, parameters, - [weakptr = GetWeakPtr()](SubscribeRemoteTrack* track) { + [weakptr = GetWeakPtr()](LiveSubscriber* track) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); if (session == nullptr || !track->track_alias().has_value()) { return false; @@ -512,7 +512,7 @@ *track->track_alias(), track); return success; }, - [weakptr = GetWeakPtr()](SubscribeRemoteTrack* track) { + [weakptr = GetWeakPtr()](LiveSubscriber* track) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); if (session == nullptr) { return; @@ -594,9 +594,9 @@ MoqtRequestErrorInfo{RequestErrorCode::kDuplicateSubscription, std::nullopt, "PUBLISH is coming"}); } - auto stream_visitor = std::make_unique<MoqtPublishPublisherStream>( + auto stream_visitor = std::make_unique<MoqtPublishRequestStream>( &framer_, ControlMessageParser(), - [weak_session = GetWeakPtr()](SubscriptionPublisher* publisher) { + [weak_session = GetWeakPtr()](LivePublisher* publisher) { MoqtSession* session = MoqtSessionFromWeakPtr(weak_session); if (session == nullptr) { return; @@ -613,14 +613,14 @@ session->Error(code, reason); }, std::move(response_callback)); - auto publish_state = std::make_unique<SubscriptionPublisher>( + auto publish_state = std::make_unique<LivePublisher>( framer_, publisher, stream_visitor.get(), next_request_id_, next_local_track_alias_, parameters, weak_ptr_factory_for_publishers_.Create(), true); - SubscriptionPublisher* publisher_ptr = publish_state.get(); + LivePublisher* publisher_ptr = publish_state.get(); stream_visitor->SetPublisher(std::move(publish_state)); webtransport::Stream* stream = session_->OpenOutgoingBidirectionalStream(); - MoqtPublishPublisherStream* stream_visitor_ptr = stream_visitor.get(); + MoqtPublishRequestStream* stream_visitor_ptr = stream_visitor.get(); stream->SetVisitor(std::move(stream_visitor)); stream_visitor_ptr->BindStream(stream); next_request_id_ += 2; @@ -673,7 +673,7 @@ name, visitor, [this, track_name = name](std::unique_ptr<MoqtFetchTask> fetch_task) { // Move the fetch_task to the subscribe to plumb into its visitor. - SubscribeRemoteTrack* subscribe = SubscribeByName(track_name); + LiveSubscriber* subscribe = SubscribeByName(track_name); if (subscribe == nullptr || subscribe->is_fetch()) { fetch_task.release(); return; @@ -810,7 +810,7 @@ return true; } -SubscribeRemoteTrack* MoqtSession::SubscribeByAlias(uint64_t track_alias) { +LiveSubscriber* MoqtSession::SubscribeByAlias(uint64_t track_alias) { auto it = subscribe_by_alias_.find(track_alias); if (it == subscribe_by_alias_.end()) { return nullptr; @@ -818,8 +818,7 @@ return it->second; } -SubscribeRemoteTrack* MoqtSession::SubscribeByName( - const FullTrackName& track_name) { +LiveSubscriber* MoqtSession::SubscribeByName(const FullTrackName& track_name) { auto it = subscribe_by_name_.find(track_name); if (it == subscribe_by_name_.end()) { return nullptr; @@ -929,33 +928,35 @@ break; } case MoqtMessageType::kSubscribeNamespace: { - auto namespace_stream = std::make_unique<MoqtNamespacePublisherStream>( - &session_->framer_, session_->ControlMessageParser(), - [weakptr = session_->GetWeakPtr()](const TrackNamespace& prefix) { - MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); - if (session != nullptr) { - return session->incoming_subscribe_namespace_.SubscribeNamespace( - prefix); - } - return true; - }, - [weakptr = session_->GetWeakPtr()](const TrackNamespace& prefix) { - MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); - if (session != nullptr) { - session->incoming_subscribe_namespace_.UnsubscribeNamespace( - prefix); - } - }, - [weakptr = session_->GetWeakPtr()](MoqtError code, - absl::string_view reason) { - MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); - if (session != nullptr) { - session->Error(code, reason); - } - }, - session_->callbacks_.incoming_subscribe_namespace_callback); + auto namespace_stream = + std::make_unique<MoqtSubscribeNamespaceResponseStream>( + &session_->framer_, session_->ControlMessageParser(), + [weakptr = session_->GetWeakPtr()](const TrackNamespace& prefix) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session != nullptr) { + return session->incoming_subscribe_namespace_ + .SubscribeNamespace(prefix); + } + return true; + }, + [weakptr = session_->GetWeakPtr()](const TrackNamespace& prefix) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session != nullptr) { + session->incoming_subscribe_namespace_.UnsubscribeNamespace( + prefix); + } + }, + [weakptr = session_->GetWeakPtr()](MoqtError code, + absl::string_view reason) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session != nullptr) { + session->Error(code, reason); + } + }, + session_->callbacks_.incoming_subscribe_namespace_callback); namespace_stream->BindStream(std::move(parser_)); - MoqtNamespacePublisherStream* temp_stream = namespace_stream.get(); + MoqtSubscribeNamespaceResponseStream* temp_stream = + namespace_stream.get(); stream_->SetVisitor(std::move(namespace_stream)); // The UnknownBidiStream object is deleted; no class access after this // point. @@ -963,7 +964,7 @@ break; } case MoqtMessageType::kPublish: { - auto publish_stream = std::make_unique<MoqtPublishSubscriberStream>( + auto publish_stream = std::make_unique<MoqtPublishResponseStream>( &session_->framer_, session_->ControlMessageParser(), session_->callbacks_.clock, session_->alarm_factory(), [weakptr = session_->GetWeakPtr()](MoqtError code, @@ -974,7 +975,7 @@ } }, &session_->callbacks_.incoming_publish_callback, - [weakptr = session_->GetWeakPtr()](SubscribeRemoteTrack* track) { + [weakptr = session_->GetWeakPtr()](LiveSubscriber* track) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); if (session == nullptr) { return false; @@ -1007,7 +1008,7 @@ QUICHE_DCHECK(name_inserted); return true; }, - [weakptr = session_->GetWeakPtr()](SubscribeRemoteTrack* track) { + [weakptr = session_->GetWeakPtr()](LiveSubscriber* track) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); if (session != nullptr) { session->subscribe_by_name_.erase(track->full_track_name()); @@ -1017,7 +1018,7 @@ } }); publish_stream->BindStream(std::move(parser_)); - MoqtPublishSubscriberStream* temp_stream = publish_stream.get(); + MoqtPublishResponseStream* temp_stream = publish_stream.get(); stream_->SetVisitor(std::move(publish_stream)); // The UnknownBidiStream object is deleted; no class access after this // point. @@ -1028,8 +1029,7 @@ auto subscribe_stream = std::make_unique<MoqtSubscribeResponseStream>( &session_->framer_, session_->ControlMessageParser(), session_->next_local_track_alias_++, - [weakptr = - session_->GetWeakPtr()](SubscriptionPublisher* subscription) { + [weakptr = session_->GetWeakPtr()](LivePublisher* subscription) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); if (session == nullptr) { return true; @@ -1043,8 +1043,7 @@ subscription->publisher().GetTrackName(), subscription); return success2; }, - [weakptr = - session_->GetWeakPtr()](SubscriptionPublisher* subscription) { + [weakptr = session_->GetWeakPtr()](LivePublisher* subscription) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); if (session == nullptr) { return; @@ -1553,9 +1552,9 @@ return absl::OkStatus(); } -void MoqtSession::OnMalformedTrack(RemoteTrack* track) { +void MoqtSession::OnMalformedTrack(ObjectSubscriber* track) { if (!track->is_fetch()) { - auto* subscribe = absl::down_cast<SubscribeRemoteTrack*>(track); + auto* subscribe = absl::down_cast<LiveSubscriber*>(track); if (subscribe->visitor() != nullptr) { subscribe->visitor()->OnMalformedTrack(track->full_track_name()); }
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index fd45f9d..66660be 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -28,16 +28,16 @@ #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_publisher.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/quic/moqt/moqt_session_interface.h" -#include "quiche/quic/moqt/moqt_subscription.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/moqt_uni_stream.h" #include "quiche/quic/moqt/session_namespace_tree.h" @@ -169,19 +169,19 @@ // Called when the incoming track is malformed per Section 2.5 of // draft-ietf-moqt-moq-transport-12. Unsubscribe and notify the application so // the error can be propagated downstream, if necessary. - void OnMalformedTrack(RemoteTrack* track); - quiche::QuicheWeakPtr<RemoteTrack> GetSubscribe( + void OnMalformedTrack(ObjectSubscriber* track); + quiche::QuicheWeakPtr<ObjectSubscriber> GetSubscribe( uint64_t track_alias) override { - RemoteTrack* track = SubscribeByAlias(track_alias); + ObjectSubscriber* track = SubscribeByAlias(track_alias); if (track == nullptr) { - return quiche::QuicheWeakPtr<RemoteTrack>(); + return quiche::QuicheWeakPtr<ObjectSubscriber>(); } return track->weak_ptr(); } - quiche::QuicheWeakPtr<RemoteTrack> GetFetch(uint64_t request_id) { + quiche::QuicheWeakPtr<ObjectSubscriber> GetFetch(uint64_t request_id) { auto it = fetch_by_id_.find(request_id); if (it == fetch_by_id_.end()) { - return quiche::QuicheWeakPtr<RemoteTrack>(); + return quiche::QuicheWeakPtr<ObjectSubscriber>(); } return it->second->weak_ptr(); } @@ -399,8 +399,8 @@ // Returns false if creation failed. [[nodiscard]] bool OpenDataStream(PublishedFetch* fetch, webtransport::SendOrder send_order); - SubscribeRemoteTrack* SubscribeByAlias(uint64_t track_alias); - SubscribeRemoteTrack* SubscribeByName(const FullTrackName& track_name); + LiveSubscriber* SubscribeByAlias(uint64_t track_alias); + LiveSubscriber* SubscribeByName(const FullTrackName& track_name); UpstreamFetch* FetchById(uint64_t request_id); // Checks that a subscribe ID from a SUBSCRIBE or FETCH is valid, and throws @@ -416,7 +416,7 @@ if (!SupportsObjectAck()) { return; } - SubscribeRemoteTrack* track = SubscribeByName(track_name); + LiveSubscriber* track = SubscribeByName(track_name); if (track != nullptr) { track->SendObjectAck(group_id, object_id, delta_from_deadline); } @@ -498,9 +498,9 @@ // Upstream FETCHes, indexed by request_id. Do not erase. absl::flat_hash_map<uint64_t, std::unique_ptr<UpstreamFetch>> fetch_by_id_; // All outgoing SUBSCRIBE and incoming PUBLISH, indexed by track_alias. - absl::flat_hash_map<uint64_t, SubscribeRemoteTrack*> subscribe_by_alias_; + absl::flat_hash_map<uint64_t, LiveSubscriber*> subscribe_by_alias_; // All outgoing SUBSCRIBE and incoming PUBLISH, indexed by track name. - absl::flat_hash_map<FullTrackName, SubscribeRemoteTrack*> subscribe_by_name_; + absl::flat_hash_map<FullTrackName, LiveSubscriber*> subscribe_by_name_; // The next subscribe ID that the local endpoint can send. uint64_t next_request_id_ = 0; @@ -510,14 +510,12 @@ // All open incoming subscriptions, indexed by track name, used to check for // duplicates. - absl::flat_hash_map<FullTrackName, SubscriptionPublisher*> - subscribed_track_names_; + absl::flat_hash_map<FullTrackName, LivePublisher*> subscribed_track_names_; // Application object representing the publisher for all of the tracks that // can be subscribed to via this connection. Must outlive this object. MoqtPublisher* publisher_; // Subscriptions for local tracks by the remote peer, indexed by request ID. - absl::flat_hash_map<uint64_t, SubscriptionPublisher*> - published_subscriptions_; + absl::flat_hash_map<uint64_t, LivePublisher*> published_subscriptions_; // Keeps track of all request IDs that have queued outgoing data streams. // The first element is the highest priority (lowest integer). absl::btree_multimap<MoqtTrackPriority, uint64_t>
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index afff2a7..d4a6b97 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -31,11 +31,11 @@ #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" #include "quiche/quic/moqt/moqt_object.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_publisher.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/quic/moqt/moqt_session_interface.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h" #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h" @@ -262,7 +262,7 @@ webtransport::test::MockSession& session, webtransport::test::MockStream* stream, std::unique_ptr<webtransport::StreamVisitor>& visitor, - MockSubscribeRemoteTrackVisitor* track_visitor) { + MockLiveSubscriberVisitor* track_visitor) { MoqtFramer framer(true, quic::Perspective::IS_SERVER); std::optional<PublishedObjectMetadata> previous_object; if (visitor != nullptr) { @@ -319,7 +319,7 @@ } } - MockSubscribeRemoteTrackVisitor remote_track_visitor_; + MockLiveSubscriberVisitor remote_track_visitor_; MoqtKnownTrackPublisher publisher_; webtransport::test::MockSession mock_session_; MockSessionCallbacks session_callbacks_; @@ -613,8 +613,8 @@ EXPECT_CALL(mock_bidi_stream_, Writev(ControlMessageOfType(MoqtMessageType::kSubscribeOk), _)); listener->OnSubscribeAccepted(); - EXPECT_TRUE(MoqtSessionPeer::RequestIdIsSubscriptionPublisher( - &session_, kDefaultPeerRequestId)); + EXPECT_TRUE(MoqtSessionPeer::RequestIdIsLivePublisher(&session_, + kDefaultPeerRequestId)); } TEST_F(MoqtSessionTest, AsynchronousSubscribeReturnsError) { @@ -746,7 +746,7 @@ // Peer unsubscribes. bidi_wrapper_->stream().Reset(kResetCodeCancelled); bidi_wrapper_ = nullptr; - EXPECT_FALSE(MoqtSessionPeer::RequestIdIsSubscriptionPublisher(&session_, 1)); + EXPECT_FALSE(MoqtSessionPeer::RequestIdIsLivePublisher(&session_, 1)); // Subscribe again, succeeds. request.request_id = 3; @@ -1815,8 +1815,8 @@ webtransport::test::MockStream control_stream; std::unique_ptr<MoqtBidiStreamTestWrapper> control_wrapper = MoqtSessionPeer::CreateControlStream(&session_, &control_stream); - ASSERT_TRUE(MoqtSessionPeer::RequestIdIsSubscriptionPublisher( - &session_, subscribe.request_id)); + ASSERT_TRUE(MoqtSessionPeer::RequestIdIsLivePublisher(&session_, + subscribe.request_id)); MoqtFetch fetch = DefaultFetch(); fetch.request_id = 3; fetch.fetch = JoiningFetchRelative(1, 2); @@ -1836,8 +1836,8 @@ SetLargestId(track, Location(4, 10)); ReceiveSubscribeSynchronousOk(track, subscribe, bidi_wrapper_.get()); - ASSERT_TRUE(MoqtSessionPeer::RequestIdIsSubscriptionPublisher( - &session_, subscribe.request_id)); + ASSERT_TRUE(MoqtSessionPeer::RequestIdIsLivePublisher(&session_, + subscribe.request_id)); webtransport::test::MockStream control_stream; std::unique_ptr<MoqtBidiStreamTestWrapper> control_wrapper = MoqtSessionPeer::CreateControlStream(&session_, &control_stream);
diff --git a/quiche/quic/moqt/moqt_subscribe_stream.cc b/quiche/quic/moqt/moqt_subscribe_stream.cc index 473de89..408c7cc 100644 --- a/quiche/quic/moqt/moqt_subscribe_stream.cc +++ b/quiche/quic/moqt/moqt_subscribe_stream.cc
@@ -18,13 +18,13 @@ #include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_publisher.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" -#include "quiche/quic/moqt/moqt_subscription.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/common/quiche_weak_ptr.h" namespace moqt { @@ -34,13 +34,13 @@ const MoqtControlMessageParser& message_parser, uint64_t request_id, SessionErrorCallback session_error_callback, const FullTrackName& name, SubscribeVisitor* absl_nonnull visitor, const MessageParameters& parameters, - SubscribeRemoteTrack::AddCallback add_callback, - SubscribeRemoteTrack::RemoveCallback remove_callback, + LiveSubscriber::AddCallback add_callback, + LiveSubscriber::RemoveCallback remove_callback, const quic::QuicClock* absl_nonnull clock, quic::QuicAlarmFactory* absl_nonnull alarm_factory) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), - track_(std::make_unique<SubscribeRemoteTrack>( + track_(std::make_unique<LiveSubscriber>( MoqtSubscribe{request_id, name, parameters}, visitor, this)), add_callback_(std::move(add_callback)), remove_callback_(std::move(remove_callback)), @@ -132,7 +132,7 @@ void MoqtSubscribeRequestStream::Detach() { if (remove_callback_ != nullptr) { - SubscribeRemoteTrack::RemoveCallback remove_callback = + LiveSubscriber::RemoveCallback remove_callback = std::move(remove_callback_); remove_callback_ = nullptr; std::move(remove_callback)(track_.get()); @@ -143,8 +143,8 @@ MoqtSubscribeResponseStream::MoqtSubscribeResponseStream( MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, uint64_t track_alias, - SubscriptionPublisher::AddCallback add_callback, - SubscriptionPublisher::RemoveCallback remove_callback, + LivePublisher::AddCallback add_callback, + LivePublisher::RemoveCallback remove_callback, SessionErrorCallback session_error_callback, quiche::QuicheWeakPtr<SessionToPublisherInterface> session) : MoqtBidiStreamBase(framer, message_parser, @@ -178,7 +178,7 @@ return SendRequestError(message.request_id, RequestErrorCode::kDoesNotExist, std::nullopt, "not found", /*fin=*/true); } - subscription_ = std::make_unique<SubscriptionPublisher>( + subscription_ = std::make_unique<LivePublisher>( *framer(), track_publisher, this, message.request_id, track_alias_, message.parameters, session_, false); if (add_callback_ != nullptr) { @@ -210,8 +210,7 @@ void MoqtSubscribeResponseStream::Detach() { if (remove_callback_ != nullptr && subscription_ != nullptr) { - SubscriptionPublisher::RemoveCallback remove_callback = - std::move(remove_callback_); + LivePublisher::RemoveCallback remove_callback = std::move(remove_callback_); remove_callback_ = nullptr; std::move(remove_callback)(subscription_.get()); }
diff --git a/quiche/quic/moqt/moqt_subscribe_stream.h b/quiche/quic/moqt/moqt_subscribe_stream.h index 3b6f5aa..ba216f9 100644 --- a/quiche/quic/moqt/moqt_subscribe_stream.h +++ b/quiche/quic/moqt/moqt_subscribe_stream.h
@@ -15,12 +15,12 @@ #include "quiche/quic/moqt/moqt_bidi_stream.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" -#include "quiche/quic/moqt/moqt_subscription.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/common/quiche_weak_ptr.h" namespace moqt { @@ -33,8 +33,8 @@ SessionErrorCallback session_error_callback, const FullTrackName& name, SubscribeVisitor* absl_nonnull visitor, const MessageParameters& parameters, - SubscribeRemoteTrack::AddCallback add_callback, - SubscribeRemoteTrack::RemoveCallback remove_callback, + LiveSubscriber::AddCallback add_callback, + LiveSubscriber::RemoveCallback remove_callback, const quic::QuicClock* absl_nonnull clock, quic::QuicAlarmFactory* absl_nonnull alarm_factory); ~MoqtSubscribeRequestStream() { Detach(); } @@ -48,13 +48,13 @@ absl::Status OnControlMessage(const MoqtSubscribeOk& message); absl::Status OnControlMessage(const MoqtPublishDone& message); - SubscribeRemoteTrack* track() const { return track_.get(); } + LiveSubscriber* track() const { return track_.get(); } void Detach() override; private: - std::unique_ptr<SubscribeRemoteTrack> track_; - SubscribeRemoteTrack::AddCallback add_callback_; - SubscribeRemoteTrack::RemoveCallback remove_callback_; + std::unique_ptr<LiveSubscriber> track_; + LiveSubscriber::AddCallback add_callback_; + LiveSubscriber::RemoveCallback remove_callback_; const quic::QuicClock* clock_; quic::QuicAlarmFactory* alarm_factory_; }; @@ -64,8 +64,8 @@ MoqtSubscribeResponseStream( MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, uint64_t track_alias, - SubscriptionPublisher::AddCallback add_callback, - SubscriptionPublisher::RemoveCallback remove_callback, + LivePublisher::AddCallback add_callback, + LivePublisher::RemoveCallback remove_callback, SessionErrorCallback session_error_callback, quiche::QuicheWeakPtr<SessionToPublisherInterface> session); ~MoqtSubscribeResponseStream() { @@ -103,9 +103,9 @@ } uint64_t track_alias_; - std::unique_ptr<SubscriptionPublisher> subscription_; - SubscriptionPublisher::AddCallback add_callback_; - SubscriptionPublisher::RemoveCallback remove_callback_; + std::unique_ptr<LivePublisher> subscription_; + LivePublisher::AddCallback add_callback_; + LivePublisher::RemoveCallback remove_callback_; quiche::QuicheWeakPtr<SessionToPublisherInterface> session_; };
diff --git a/quiche/quic/moqt/moqt_subscribe_stream_test.cc b/quiche/quic/moqt/moqt_subscribe_stream_test.cc index 300dafb..3927e3f 100644 --- a/quiche/quic/moqt/moqt_subscribe_stream_test.cc +++ b/quiche/quic/moqt/moqt_subscribe_stream_test.cc
@@ -18,14 +18,14 @@ #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/quic/moqt/moqt_session_interface.h" -#include "quiche/quic/moqt/moqt_subscription.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/test_tools/mock_moqt_session.h" #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h" #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h" @@ -68,11 +68,11 @@ MessageParameters parameters_; testing::StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>> error_callback_; - testing::MockFunction<bool(SubscribeRemoteTrack*)> mock_add_callback_; - testing::MockFunction<void(SubscribeRemoteTrack*)> mock_remove_callback_; + testing::MockFunction<bool(LiveSubscriber*)> mock_add_callback_; + testing::MockFunction<void(LiveSubscriber*)> mock_remove_callback_; quic::MockClock mock_clock_; quic::test::MockAlarmFactory mock_alarm_factory_; - StrictMock<MockSubscribeRemoteTrackVisitor> mock_subscribe_visitor_; + StrictMock<MockLiveSubscriberVisitor> mock_subscribe_visitor_; webtransport::test::MockStream mock_stream_; std::unique_ptr<MoqtSubscribeRequestStream> stream_; }; @@ -229,8 +229,8 @@ FullTrackName kTrackName{"foo", "bar"}; std::shared_ptr<TestTrackPublisher> track_publisher_; testing::MockFunction<void(MoqtError, absl::string_view)> error_callback_; - testing::MockFunction<bool(SubscriptionPublisher*)> mock_add_callback_; - testing::MockFunction<void(SubscriptionPublisher*)> mock_remove_callback_; + testing::MockFunction<bool(LivePublisher*)> mock_add_callback_; + testing::MockFunction<void(LivePublisher*)> mock_remove_callback_; MockSessionToPublisherInterface visitor_; webtransport::test::MockSession webtrans_; webtransport::test::MockStream mock_stream_;
diff --git a/quiche/quic/moqt/moqt_uni_stream.cc b/quiche/quic/moqt/moqt_uni_stream.cc index 26ab6c2..6320e27 100644 --- a/quiche/quic/moqt/moqt_uni_stream.cc +++ b/quiche/quic/moqt/moqt_uni_stream.cc
@@ -24,10 +24,10 @@ #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_object.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_publisher.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/common/platform/api/quiche_bug_tracker.h" #include "quiche/common/quiche_buffer_allocator.h" @@ -84,7 +84,7 @@ OutgoingSubgroupStream::OutgoingSubgroupStream( MoqtFramer framer, webtransport::Stream* absl_nonnull stream, DataStreamIndex index, uint64_t first_object, - quiche::QuicheWeakPtr<SubscriptionPublisherInterface> visitor, + quiche::QuicheWeakPtr<LivePublisherInterface> visitor, std::shared_ptr<MoqtTrackPublisher> absl_nonnull track_publisher, webtransport::StreamPriority priority, uint64_t track_alias, MoqtTraceRecorder* absl_nonnull trace_recorder) @@ -110,7 +110,7 @@ if (delivery_timeout_alarm_ != nullptr) { delivery_timeout_alarm_->PermanentCancel(); } - SubscriptionPublisherInterface* visitor = visitor_.GetIfAvailable(); + LivePublisherInterface* visitor = visitor_.GetIfAvailable(); if (visitor != nullptr) { visitor->OnDataStreamDestroyed(index_); } @@ -120,14 +120,14 @@ void OutgoingSubgroupStream::OnStopSendingReceived( webtransport::StreamErrorCode error_code) { - SubscriptionPublisherInterface* visitor = visitor_.GetIfAvailable(); + LivePublisherInterface* visitor = visitor_.GetIfAvailable(); if (visitor != nullptr) { visitor->OnSubgroupAbandoned(index_.group, index_.subgroup, error_code); } } void OutgoingSubgroupStream::DeliveryTimeoutDelegate::OnAlarm() { - SubscriptionPublisherInterface* visitor = stream_->visitor_.GetIfAvailable(); + LivePublisherInterface* visitor = stream_->visitor_.GetIfAvailable(); if (visitor != nullptr) { visitor->OnStreamTimeout(stream_->index_); } @@ -135,7 +135,7 @@ } void OutgoingSubgroupStream::SendObjects() { - SubscriptionPublisherInterface* visitor = visitor_.GetIfAvailable(); + LivePublisherInterface* visitor = visitor_.GetIfAvailable(); if (visitor == nullptr) { return; } @@ -241,7 +241,7 @@ absl::Status status = webtransport::SendFinOnStream(stream()); QUICHE_BUG_IF(OutgoingSubgroupStream_fin_failed, !status.ok()) << "Writing pure FIN failed."; - SubscriptionPublisherInterface* visitor = visitor_.GetIfAvailable(); + LivePublisherInterface* visitor = visitor_.GetIfAvailable(); if (visitor == nullptr) { return; } @@ -255,7 +255,7 @@ if (delivery_timeout_alarm_ != nullptr) { return; } - SubscriptionPublisherInterface* visitor = visitor_.GetIfAvailable(); + LivePublisherInterface* visitor = visitor_.GetIfAvailable(); if (visitor == nullptr) { return; } @@ -360,8 +360,7 @@ return; } // It's a subscribe. - auto subscribe = - absl::down_cast<SubscribeRemoteTrack*>(track_.GetIfAvailable()); + auto subscribe = absl::down_cast<LiveSubscriber*>(track_.GetIfAvailable()); if (subscribe == nullptr) { return; } @@ -412,7 +411,7 @@ return; } Location location(message.group_id, message.object_id); - RemoteTrack* track = track_.GetIfAvailable(); + ObjectSubscriber* track = track_.GetIfAvailable(); if (track == nullptr || !track->InWindow(Location(message.group_id, message.object_id))) { // This is not an error. It can be the result of a recent REQUEST_UPDATE or @@ -442,8 +441,7 @@ no_more_objects_ = true; } } - SubscribeRemoteTrack* subscribe = - absl::down_cast<SubscribeRemoteTrack*>(track); + LiveSubscriber* subscribe = absl::down_cast<LiveSubscriber*>(track); subscribe->OnObjectOrOk(); if (visitor_ != nullptr) { PublishedObjectMetadata metadata; @@ -533,8 +531,8 @@ if (!knew_track_alias) { track_ = session_->GetSubscribe(*parser_.track_alias()); // This is a new stream for a subscribe. Notify the subscription. - SubscribeRemoteTrack* subscribe = - absl::down_cast<SubscribeRemoteTrack*>(track_.GetIfAvailable()); + LiveSubscriber* subscribe = + absl::down_cast<LiveSubscriber*>(track_.GetIfAvailable()); if (subscribe == nullptr) { stream_->SendStopSending(kResetCodeCancelled); return;
diff --git a/quiche/quic/moqt/moqt_uni_stream.h b/quiche/quic/moqt/moqt_uni_stream.h index f828e21..b681125 100644 --- a/quiche/quic/moqt/moqt_uni_stream.h +++ b/quiche/quic/moqt/moqt_uni_stream.h
@@ -21,12 +21,12 @@ #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_object.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_publisher.h" #include "quiche/quic/moqt/moqt_session_interface.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/common/platform/api/quiche_export.h" #include "quiche/common/quiche_callbacks.h" @@ -90,9 +90,9 @@ }; // This interface provides information about the subscription. -class SubscriptionPublisherInterface { +class LivePublisherInterface { public: - virtual ~SubscriptionPublisherInterface() = default; + virtual ~LivePublisherInterface() = default; virtual bool InWindow(Location) = 0; virtual bool alternate_delivery_timeout() = 0; virtual const quic::QuicClock* clock() = 0; @@ -114,7 +114,7 @@ OutgoingSubgroupStream( MoqtFramer framer, webtransport::Stream* absl_nonnull stream, DataStreamIndex index, uint64_t first_object, - quiche::QuicheWeakPtr<SubscriptionPublisherInterface> visitor, + quiche::QuicheWeakPtr<LivePublisherInterface> visitor, std::shared_ptr<MoqtTrackPublisher> absl_nonnull track_publisher, webtransport::StreamPriority priority, uint64_t track_alias, MoqtTraceRecorder* absl_nonnull trace_recorder); @@ -155,7 +155,7 @@ void SendObjects(); DataStreamIndex index_; - quiche::QuicheWeakPtr<SubscriptionPublisherInterface> visitor_; + quiche::QuicheWeakPtr<LivePublisherInterface> visitor_; MoqtDataStreamType type_; uint64_t track_alias_; @@ -199,10 +199,11 @@ public: virtual ~SessionToUniStreamInterface() = default; virtual bool deliver_partial_objects() const = 0; - virtual void OnMalformedTrack(RemoteTrack* name) = 0; - virtual quiche::QuicheWeakPtr<RemoteTrack> GetSubscribe( + virtual void OnMalformedTrack(ObjectSubscriber* name) = 0; + virtual quiche::QuicheWeakPtr<ObjectSubscriber> GetSubscribe( uint64_t track_alias) = 0; - virtual quiche::QuicheWeakPtr<RemoteTrack> GetFetch(uint64_t request_id) = 0; + virtual quiche::QuicheWeakPtr<ObjectSubscriber> GetFetch( + uint64_t request_id) = 0; virtual void Error(MoqtError error_code, absl::string_view reason) = 0; }; @@ -251,7 +252,7 @@ webtransport::Stream* stream_; SubscribeVisitor* visitor_ = nullptr; // Once the subscribe ID is identified, set it here. - quiche::QuicheWeakPtr<RemoteTrack> track_; + quiche::QuicheWeakPtr<ObjectSubscriber> track_; MoqtDataParser parser_; std::string partial_object_; uint64_t bytes_received_this_object_ = 0;
diff --git a/quiche/quic/moqt/moqt_uni_stream_test.cc b/quiche/quic/moqt/moqt_uni_stream_test.cc index 65c8ad2..4265330 100644 --- a/quiche/quic/moqt/moqt_uni_stream_test.cc +++ b/quiche/quic/moqt/moqt_uni_stream_test.cc
@@ -22,8 +22,8 @@ #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" #include "quiche/quic/moqt/moqt_object.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h" #include "quiche/quic/moqt/test_tools/moqt_session_peer.h" @@ -59,10 +59,9 @@ return object; } -class MockSubscriptionPublisherInterface - : public SubscriptionPublisherInterface { +class MockLivePublisherInterface : public LivePublisherInterface { public: - MockSubscriptionPublisherInterface() : weak_ptr_factory_(this) {} + MockLivePublisherInterface() : weak_ptr_factory_(this) {} MOCK_METHOD(bool, InWindow, (Location), (override)); MOCK_METHOD(bool, alternate_delivery_timeout, (), (override)); @@ -75,13 +74,12 @@ (uint64_t, uint64_t, webtransport::StreamErrorCode), (override)); MOCK_METHOD(void, OnDataStreamDestroyed, (DataStreamIndex), (override)); - quiche::QuicheWeakPtr<SubscriptionPublisherInterface> GetWeakPtr() { + quiche::QuicheWeakPtr<LivePublisherInterface> GetWeakPtr() { return weak_ptr_factory_.Create(); } private: - quiche::QuicheWeakPtrFactory<SubscriptionPublisherInterface> - weak_ptr_factory_; + quiche::QuicheWeakPtrFactory<LivePublisherInterface> weak_ptr_factory_; }; class OutgoingSubgroupStreamTest : public quic::test::QuicTest { @@ -125,7 +123,7 @@ StrictMock<webtransport::test::MockStream> mock_stream_; DataStreamIndex index_; std::shared_ptr<StrictMock<MockTrackPublisher>> track_publisher_; - StrictMock<MockSubscriptionPublisherInterface> visitor_; + StrictMock<MockLivePublisherInterface> visitor_; MoqtTraceRecorder trace_recorder_; TrackExtensions track_extensions_; quic::MockClock mock_clock_; @@ -475,10 +473,10 @@ ~MockSessionToUniStreamInterface() override = default; MOCK_METHOD(bool, deliver_partial_objects, (), (const, override)); - MOCK_METHOD(void, OnMalformedTrack, (RemoteTrack*), (override)); - MOCK_METHOD(quiche::QuicheWeakPtr<RemoteTrack>, GetSubscribe, (uint64_t), + MOCK_METHOD(void, OnMalformedTrack, (ObjectSubscriber*), (override)); + MOCK_METHOD(quiche::QuicheWeakPtr<ObjectSubscriber>, GetSubscribe, (uint64_t), (override)); - MOCK_METHOD(quiche::QuicheWeakPtr<RemoteTrack>, GetFetch, (uint64_t), + MOCK_METHOD(quiche::QuicheWeakPtr<ObjectSubscriber>, GetFetch, (uint64_t), (override)); MOCK_METHOD(void, Error, (MoqtError, absl::string_view), (override)); }; @@ -491,8 +489,8 @@ subscribe_message_(1, ftn_, MessageParameters()) { EXPECT_CALL(session_, deliver_partial_objects()) .WillRepeatedly(Return(false)); - track_ = std::make_unique<SubscribeRemoteTrack>(subscribe_message_, - &visitor_, nullptr); + track_ = std::make_unique<LiveSubscriber>(subscribe_message_, &visitor_, + nullptr); track_->set_track_alias(2); CreateStream(); } @@ -520,8 +518,8 @@ quic::MockClock mock_clock_; FullTrackName ftn_; MoqtSubscribe subscribe_message_; - testing::NiceMock<MockSubscribeRemoteTrackVisitor> visitor_; - std::unique_ptr<SubscribeRemoteTrack> track_; + testing::NiceMock<MockLiveSubscriberVisitor> visitor_; + std::unique_ptr<LiveSubscriber> track_; std::unique_ptr<IncomingDataStream> stream_; }; @@ -702,7 +700,7 @@ mock_stream_.Receive( absl::string_view(reinterpret_cast<const char*>(&alias), 1), false); EXPECT_CALL(session_, GetSubscribe(2)) - .WillOnce(Return(quiche::QuicheWeakPtr<RemoteTrack>())); + .WillOnce(Return(quiche::QuicheWeakPtr<ObjectSubscriber>())); stream_->OnCanRead(); EXPECT_TRUE(mock_stream_.was_reset()); } @@ -765,7 +763,7 @@ char fetch_bytes[] = {0x05, 0x03}; mock_stream_.Receive(absl::string_view(fetch_bytes, 2), false); EXPECT_CALL(session_, GetFetch(3)) - .WillOnce(Return(quiche::QuicheWeakPtr<RemoteTrack>())); + .WillOnce(Return(quiche::QuicheWeakPtr<ObjectSubscriber>())); stream_->OnCanRead(); EXPECT_TRUE(mock_stream_.was_reset()); }
diff --git a/quiche/quic/moqt/test_tools/mock_moqt_session.h b/quiche/quic/moqt/test_tools/mock_moqt_session.h index 8583e26..2cd1df2 100644 --- a/quiche/quic/moqt/test_tools/mock_moqt_session.h +++ b/quiche/quic/moqt/test_tools/mock_moqt_session.h
@@ -22,6 +22,7 @@ #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" #include "quiche/quic/moqt/moqt_parser.h" @@ -29,7 +30,6 @@ #include "quiche/quic/moqt/moqt_publisher.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/quic/moqt/moqt_session_interface.h" -#include "quiche/quic/moqt/moqt_subscription.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/common/platform/api/quiche_test.h"
diff --git a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h index 5915dfc..5b46d41 100644 --- a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h +++ b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h
@@ -187,7 +187,7 @@ }; // TODO(martinduke): Rename to MockSubscribeVisitor. -class MockSubscribeRemoteTrackVisitor : public SubscribeVisitor { +class MockLiveSubscriberVisitor : public SubscribeVisitor { public: MOCK_METHOD(void, OnReply, (const FullTrackName& full_track_name,
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h index c7bdb73..65dce82 100644 --- a/quiche/quic/moqt/test_tools/moqt_session_peer.h +++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -23,13 +23,13 @@ #include "quiche/quic/moqt/moqt_bidi_stream.h" #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" +#include "quiche/quic/moqt/moqt_live_publisher.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" +#include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session.h" #include "quiche/quic/moqt/moqt_session_interface.h" -#include "quiche/quic/moqt/moqt_subscription.h" -#include "quiche/quic/moqt/moqt_track.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/moqt_uni_stream.h" #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h" @@ -109,8 +109,8 @@ return new_stream; } - static bool RequestIdIsSubscriptionPublisher(MoqtSession* session, - uint64_t request_id) { + static bool RequestIdIsLivePublisher(MoqtSession* session, + uint64_t request_id) { return session->published_subscriptions_.contains(request_id); } @@ -157,8 +157,7 @@ return session->callbacks_.clock->ApproximateNow(); } - static quic::QuicAlarm* GetPublishDoneAlarm( - SubscribeRemoteTrack* subscription) { + static quic::QuicAlarm* GetPublishDoneAlarm(LiveSubscriber* subscription) { return subscription->publish_done_alarm_.get(); }
diff --git a/quiche/quic/moqt/tools/chat_client.cc b/quiche/quic/moqt/tools/chat_client.cc index 329f8a4..558470c 100644 --- a/quiche/quic/moqt/tools/chat_client.cc +++ b/quiche/quic/moqt/tools/chat_client.cc
@@ -201,7 +201,7 @@ queue_->AddObject(std::move(message_slice), /*key=*/true); } -void ChatClient::RemoteTrackVisitor::OnReply( +void ChatClient::ObjectSubscriberVisitor::OnReply( const FullTrackName& full_track_name, std::variant<SubscribeOkData, MoqtRequestErrorInfo> response) { auto it = client_->other_users_.find(full_track_name); @@ -223,7 +223,7 @@ response); } -void ChatClient::RemoteTrackVisitor::OnObjectFragment( +void ChatClient::ObjectSubscriberVisitor::OnObjectFragment( const FullTrackName& full_track_name, const PublishedObjectMetadata& metadata, absl::string_view object, uint64_t offset) {
diff --git a/quiche/quic/moqt/tools/chat_client.h b/quiche/quic/moqt/tools/chat_client.h index 6e229d5..b41f898 100644 --- a/quiche/quic/moqt/tools/chat_client.h +++ b/quiche/quic/moqt/tools/chat_client.h
@@ -100,9 +100,9 @@ quic::QuicEventLoop* event_loop() { return event_loop_; } - class QUICHE_EXPORT RemoteTrackVisitor : public moqt::SubscribeVisitor { + class QUICHE_EXPORT ObjectSubscriberVisitor : public moqt::SubscribeVisitor { public: - RemoteTrackVisitor(ChatClient* client) : client_(client) {} + ObjectSubscriberVisitor(ChatClient* client) : client_(client) {} void OnReply( const moqt::FullTrackName& full_track_name, @@ -149,7 +149,7 @@ // Related to subscriptions/publish_namespaces // TODO: One for each subscribe - RemoteTrackVisitor remote_track_visitor_; + ObjectSubscriberVisitor remote_track_visitor_; // General state variables // The event loop to use for this client.