Support the "has first object in the subgroup" bit in the subgroup header. PiperOrigin-RevId: 927552535
diff --git a/quiche/quic/moqt/moqt_framer_test.cc b/quiche/quic/moqt/moqt_framer_test.cc index 79f74fb..482e5c4 100644 --- a/quiche/quic/moqt/moqt_framer_test.cc +++ b/quiche/quic/moqt/moqt_framer_test.cc
@@ -265,7 +265,8 @@ }; TEST_F(MoqtFramerSimpleTest, GroupMiddler) { - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(1, 1, true, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(1, 1, true, false, true); auto header = std::make_unique<StreamHeaderSubgroupMessage>(type); auto buffer1 = SerializeObject( framer_, std::get<MoqtObject>(header->structured_data()), "foo", type, 0); @@ -324,13 +325,15 @@ std::string(kDefaultExtensionBlob.data(), kDefaultExtensionBlob.size()), /*object_status=*/MoqtObjectStatus::kObjectDoesNotExist, /*subgroup_id=*/8, + /*first_object_in_subgroup=*/true, /*payload_length=*/3, }; quiche::QuicheBuffer buffer; std::optional<PublishedObjectMetadata> previous; EXPECT_QUIC_BUG( buffer = framer_.SerializeObjectHeader( - object, MoqtDataStreamType::Subgroup(8, 0, false, false), previous), + object, MoqtDataStreamType::Subgroup(8, 0, false, false, true), + previous), "Object metadata is invalid"); EXPECT_TRUE(buffer.empty()); } @@ -345,6 +348,7 @@ std::string(kDefaultExtensionBlob), /*object_status=*/MoqtObjectStatus::kNormal, /*subgroup_id=*/std::nullopt, + /*first_object_in_subgroup=*/std::nullopt, /*payload_length=*/3, }; quiche::QuicheBuffer buffer;
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h index d8c0ac8..edfe38b 100644 --- a/quiche/quic/moqt/moqt_messages.h +++ b/quiche/quic/moqt/moqt_messages.h
@@ -44,6 +44,7 @@ static constexpr uint64_t kExtensions = 0x01; static constexpr uint64_t kEndOfGroup = 0x08; static constexpr uint64_t kDefaultPriority = 0x20; + static constexpr uint64_t kHasFirstObject = 0x40; // These two cannot simultaneously be true; static constexpr uint64_t kFirstObjectId = 0x02; static constexpr uint64_t kSubgroupId = 0x04; @@ -59,7 +60,7 @@ return std::nullopt; } if (value > (kSubgroup | kExtensions | kEndOfGroup | kDefaultPriority | - kFirstObjectId | kSubgroupId)) { + kFirstObjectId | kSubgroupId | kHasFirstObject)) { // Reserved bits. return std::nullopt; } @@ -70,11 +71,9 @@ } static MoqtDataStreamType Fetch() { return MoqtDataStreamType(kFetch); } static MoqtDataStreamType Padding() { return MoqtDataStreamType(kPadding); } - static MoqtDataStreamType Subgroup(uint64_t subgroup_id, - uint64_t first_object_id, - bool no_extension_headers, - bool default_priority, - bool end_of_group = false) { + static MoqtDataStreamType Subgroup( + uint64_t subgroup_id, uint64_t first_object_id, bool no_extension_headers, + bool default_priority, bool has_first_object, bool end_of_group = false) { uint64_t value = kSubgroup; if (!no_extension_headers) { value |= kExtensions; @@ -85,6 +84,9 @@ if (default_priority) { value |= kDefaultPriority; } + if (has_first_object) { + value |= kHasFirstObject; + } if (subgroup_id == 0) { return MoqtDataStreamType(value); } @@ -117,6 +119,9 @@ bool HasDefaultPriority() const { return IsSubgroup() && (value_ & kDefaultPriority); } + bool HasFirstObject() const { + return IsSubgroup() && (value_ & kHasFirstObject); + } uint64_t value() const { return value_; } MoqtDataStreamType& operator=(const MoqtDataStreamType& other) = default; @@ -252,7 +257,8 @@ MoqtPriority publisher_priority; std::string extension_headers; // Raw, unparsed extension headers. MoqtObjectStatus object_status; - std::optional<uint64_t> subgroup_id; // Only for subgroup objects. + std::optional<uint64_t> subgroup_id; // Only for subgroup objects. + std::optional<bool> first_object_in_subgroup; // Only for subgroup objects. uint64_t payload_length; };
diff --git a/quiche/quic/moqt/moqt_object.h b/quiche/quic/moqt/moqt_object.h index 895a360..996a5e7 100644 --- a/quiche/quic/moqt/moqt_object.h +++ b/quiche/quic/moqt/moqt_object.h
@@ -29,6 +29,11 @@ std::string extensions; MoqtObjectStatus status = MoqtObjectStatus::kNormal; MoqtPriority publisher_priority = kDefaultPublisherPriority; + // `first_object_in_subgroup` is only available in objects communicated via + // a subscription. It is not available for the fetched objects; however, the + // subscriber can guarantee that all subgroups have a start object by setting + // the FETCH range appropriately. + std::optional<bool> first_object_in_subgroup; // The length of the entire payload, which might include data that is not // present in an encompassing PublishedObject or CachedObject. uint64_t payload_length;
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.cc b/quiche/quic/moqt/moqt_outgoing_queue.cc index 71b3e39..9472a89 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue.cc +++ b/quiche/quic/moqt/moqt_outgoing_queue.cc
@@ -79,6 +79,7 @@ "", status, default_publisher_priority(), + queue_.back().empty(), payload.length(), clock_->ApproximateNow()}; queue_.back().push_back(
diff --git a/quiche/quic/moqt/moqt_outgoing_queue_test.cc b/quiche/quic/moqt/moqt_outgoing_queue_test.cc index b091536..c1265f9 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue_test.cc +++ b/quiche/quic/moqt/moqt_outgoing_queue_test.cc
@@ -62,6 +62,7 @@ AnyOf(MoqtObjectStatus::kNormal, MoqtObjectStatus::kEndOfGroup, MoqtObjectStatus::kEndOfTrack))))); + EXPECT_EQ(object->metadata.first_object_in_subgroup, sequence.object == 0); if (object->metadata.status == MoqtObjectStatus::kNormal) { PublishObject(object->metadata.location.group, object->metadata.location.object,
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index 19e3377..13bfe26 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -1285,6 +1285,10 @@ if (num_objects_read_ == 0 && type_.SubgroupIsFirstObjectId()) { metadata_.subgroup_id = metadata_.object_id; } + if (type_.IsSubgroup()) { + metadata_.first_object_in_subgroup = + type_.HasFirstObject() && num_objects_read_ == 0; + } if (type_.AreExtensionHeadersPresent()) { return kExtensionSize; }
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc index e3c09f0..7b9d69c 100644 --- a/quiche/quic/moqt/moqt_parser_test.cc +++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -463,7 +463,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(1, 1, true, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(1, 1, true, false, true); auto message = std::make_unique<StreamHeaderSubgroupMessage>(type); EXPECT_TRUE(message->SetPayloadLength(14)); message->set_wire_image_size(message->total_message_size() - 11); @@ -499,7 +500,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(2, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(2, 1, false, false, true); auto message = std::make_unique<StreamHeaderSubgroupMessage>(type); EXPECT_TRUE(message->SetPayloadLength(payload_length)); @@ -533,7 +535,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(2, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(2, 1, false, false, true); auto message = std::make_unique<StreamHeaderSubgroupMessage>(type); // first part @@ -557,7 +560,8 @@ MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); // first part - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(0, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(0, 1, false, false, true); auto message1 = std::make_unique<StreamHeaderSubgroupMessage>(type); stream.Receive(message1->PacketSample(), false); parser.ReadAllData(); @@ -583,7 +587,8 @@ MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); // first part - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(0, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(0, 1, false, false, true); auto message1 = std::make_unique<StreamHeaderSubgroupMessage>(type); message1->ExpandVarints(); stream.Receive(message1->PacketSample(), false); @@ -954,7 +959,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(0, 1, true, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(0, 1, true, false, true); auto message = std::make_unique<StreamHeaderSubgroupMessage>(type); stream.Receive( message->PacketSample().substr(0, message->total_message_size() - 1), @@ -972,7 +978,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(0, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(0, 1, false, false, true); auto message = std::make_unique<StreamHeaderSubgroupMessage>(type); // Read up to the extension body and then FIN. stream.Receive(message->PacketSample().substr(0, 7), true); @@ -989,7 +996,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtParserTestVisitor data_visitor; MoqtDataParser parser(&stream, &data_visitor); - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(1, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(1, 1, false, false, true); auto message = std::make_unique<StreamHeaderSubgroupMessage>(type); stream.Receive( message->PacketSample().substr(0, message->total_message_size() - 1), @@ -1600,7 +1608,8 @@ }; TEST_F(MoqtDataParserStateMachineTest, ReadAll) { - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(0, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(0, 1, false, false, true); stream_.Receive(StreamHeaderSubgroupMessage(type).PacketSample()); stream_.Receive(StreamMiddlerSubgroupMessage(type).PacketSample()); parser_.ReadAllData(); @@ -1614,7 +1623,8 @@ } TEST_F(MoqtDataParserStateMachineTest, ReadObjects) { - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(0, 1, true, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(0, 1, true, false, true); stream_.Receive(StreamHeaderSubgroupMessage(type).PacketSample()); stream_.Receive(StreamMiddlerSubgroupMessage(type).PacketSample(), /*fin=*/true); @@ -1629,7 +1639,8 @@ } TEST_F(MoqtDataParserStateMachineTest, ReadTypeThenObjects) { - MoqtDataStreamType type = MoqtDataStreamType::Subgroup(1, 1, false, false); + MoqtDataStreamType type = + MoqtDataStreamType::Subgroup(1, 1, false, false, true); stream_.Receive(StreamHeaderSubgroupMessage(type).PacketSample()); stream_.Receive(StreamMiddlerSubgroupMessage(type).PacketSample(), /*fin=*/true); @@ -1742,7 +1753,8 @@ TEST_F(MoqtDataParserStateMachineTest, IntegerOverflowObjectId) { MoqtDataStreamType type = MoqtDataStreamType::Subgroup( - 0, 1, /*no_extension_headers=*/true, /*default_priority=*/false); + 0, 1, /*no_extension_headers=*/true, /*default_priority=*/false, + /*has_first_object=*/true); stream_.Receive(StreamHeaderSubgroupMessage(type).PacketSample()); char buffer[32]; quic::QuicDataWriter writer(sizeof(buffer), buffer); @@ -1757,4 +1769,53 @@ "Integer overflow when parsing object ID"); } +TEST_F(MoqtDataParserStateMachineTest, SubgroupHasFirstObjectTrue) { + MoqtDataStreamType type = MoqtDataStreamType::Subgroup( + 0, 1, /*no_extension_headers=*/true, /*default_priority=*/false, + /*has_first_object=*/true); + stream_.Receive(StreamHeaderSubgroupMessage(type).PacketSample()); + stream_.Receive(StreamMiddlerSubgroupMessage(type).PacketSample(), + /*fin=*/true); + parser_.ReadAtMostOneObject(); + ASSERT_EQ(visitor_.messages_received(), 1); + ASSERT_TRUE(visitor_.last_message().has_value()); + EXPECT_EQ(visitor_.last_message()->first_object_in_subgroup, true); + parser_.ReadAtMostOneObject(); + ASSERT_EQ(visitor_.messages_received(), 2); + ASSERT_TRUE(visitor_.last_message().has_value()); + EXPECT_EQ(visitor_.last_message()->first_object_in_subgroup, false); + EXPECT_EQ(visitor_.parsing_error(), std::nullopt); + EXPECT_TRUE(visitor_.fin_received()); +} + +TEST_F(MoqtDataParserStateMachineTest, SubgroupHasFirstObjectFalse) { + MoqtDataStreamType type = MoqtDataStreamType::Subgroup( + 0, 1, /*no_extension_headers=*/true, /*default_priority=*/false, + /*has_first_object=*/false); + stream_.Receive(StreamHeaderSubgroupMessage(type).PacketSample()); + stream_.Receive(StreamMiddlerSubgroupMessage(type).PacketSample(), + /*fin=*/true); + parser_.ReadAtMostOneObject(); + ASSERT_EQ(visitor_.messages_received(), 1); + ASSERT_TRUE(visitor_.last_message().has_value()); + EXPECT_EQ(visitor_.last_message()->first_object_in_subgroup, false); + parser_.ReadAtMostOneObject(); + ASSERT_EQ(visitor_.messages_received(), 2); + ASSERT_TRUE(visitor_.last_message().has_value()); + EXPECT_EQ(visitor_.last_message()->first_object_in_subgroup, false); + EXPECT_EQ(visitor_.parsing_error(), std::nullopt); + EXPECT_TRUE(visitor_.fin_received()); +} + +TEST_F(MoqtDataParserStateMachineTest, FetchFirstObjectMissing) { + StreamHeaderFetchMessage header; + stream_.Receive(header.PacketSample()); + parser_.ReadStreamType(); + ASSERT_EQ(visitor_.messages_received(), 0); + parser_.ReadAtMostOneObject(); + ASSERT_EQ(visitor_.messages_received(), 1); + ASSERT_TRUE(visitor_.last_message().has_value()); + EXPECT_FALSE(visitor_.last_message()->first_object_in_subgroup.has_value()); +} + } // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc index 32ba149..e12227d 100644 --- a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc +++ b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc
@@ -83,7 +83,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, subgroup, "", status, 128, - payload.length()}, + location.object == 0, payload.length()}, payload, /*offset=*/0); std::optional<PublishedObject> object = publisher_.GetCachedObject(location.group, subgroup, location.object); @@ -173,7 +173,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{Location(group, 0), 0, "", - MoqtObjectStatus::kEndOfGroup, 128, 0}, + MoqtObjectStatus::kEndOfGroup, 128, true, 0}, "", /*offset=*/0); } } @@ -189,7 +189,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "", MoqtObjectStatus::kNormal, 128, - 6}, + location.object == 0, 6}, "object", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -205,7 +205,8 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{first_location, 0, "", - MoqtObjectStatus::kEndOfTrack, 128, 0}, + MoqtObjectStatus::kEndOfTrack, 128, + first_location.object == 0, 0}, "", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -221,7 +222,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 1, "", MoqtObjectStatus::kEndOfGroup, - 128, 6}, + 128, location.object == 0, 6}, "object", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -237,7 +238,8 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{first_location, 1, "", - MoqtObjectStatus::kEndOfGroup, 128, 0}, + MoqtObjectStatus::kEndOfGroup, 128, + first_location.object == 0, 0}, "", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -253,7 +255,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "", MoqtObjectStatus::kNormal, 200, - 6}, + location.object == 0, 6}, "object", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -271,7 +273,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "", MoqtObjectStatus::kNormal, 128, - 6}, + location.object == 0, 6}, "object", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -287,7 +289,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{first_location, 0, "", MoqtObjectStatus::kNormal, - 128, 6}, + 128, first_location.object == 0, 6}, "object", /*offset=*/0); // Object is simply ignored; track is not malformed. EXPECT_FALSE(track_deleted_); @@ -369,7 +371,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "foo", MoqtObjectStatus::kNormal, - 128, 6}, + 128, location.object == 0, 6}, "object", /*offset=*/0); // Exact duplicate is ignored. It doesn't matter that the arrival time // changed. @@ -379,7 +381,8 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "foo", MoqtObjectStatus::kNormal, - 128, 6, quic::QuicTime::Infinite()}, + 128, location.object == 0, 6, + quic::QuicTime::Infinite()}, "object", /*offset=*/0); } @@ -392,7 +395,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "foo", MoqtObjectStatus::kNormal, - 128, 6}, + 128, location.object == 0, 6}, "object", /*offset=*/0); // Priority change; malformed track. EXPECT_CALL(listener_, OnNewObjectAvailable).Times(0); @@ -400,7 +403,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "foo", MoqtObjectStatus::kNormal, 64, - 6}, + location.object == 0, 6}, "object", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -414,7 +417,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "foo", MoqtObjectStatus::kNormal, - 128, 7}, + 128, location.object == 0, 7}, "payload", /*offset=*/0); // Payload change; malformed track. EXPECT_CALL(listener_, OnNewObjectAvailable).Times(0); @@ -422,7 +425,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, 0, "foo", MoqtObjectStatus::kNormal, - 128, 6}, + 128, location.object == 0, 6}, "foobar", /*offset=*/0); EXPECT_TRUE(track_deleted_); } @@ -469,7 +472,7 @@ publisher_.OnObjectFragment( kTrackName, PublishedObjectMetadata{location, std::nullopt, "", - MoqtObjectStatus::kNormal, 128, 6}, + MoqtObjectStatus::kNormal, 128, std::nullopt, 6}, "object", /*offset=*/0); std::optional<PublishedObject> object = publisher_.GetCachedObject(location.group, std::nullopt, 0); @@ -481,8 +484,10 @@ Location location = kLargestLocation.Next(); uint64_t subgroup = 0; // Total size is 15 bytes. - PublishedObjectMetadata metadata = { - location, subgroup, "", MoqtObjectStatus::kNormal, 128, 15}; + PublishedObjectMetadata metadata = {location, subgroup, + "", MoqtObjectStatus::kNormal, + 128, location.object == 0, + 15}; // Fragment 1 arrives. EXPECT_CALL(listener_, @@ -524,7 +529,8 @@ SubscribeAndOk(); Location location = kLargestLocation.Next(); PublishedObjectMetadata metadata = { - location, std::nullopt, "", MoqtObjectStatus::kNormal, 128, 10}; + location, std::nullopt, "", MoqtObjectStatus::kNormal, + 128, std::nullopt, 10}; // Fragment length mismatch. EXPECT_QUICHE_BUG( publisher_.OnObjectFragment(kTrackName, metadata, "short", 0), @@ -540,8 +546,10 @@ Location location = kLargestLocation.Next(); uint64_t subgroup = 0; // Total size is 15 bytes. - PublishedObjectMetadata metadata = { - location, subgroup, "", MoqtObjectStatus::kNormal, 128, 15}; + PublishedObjectMetadata metadata = {location, subgroup, + "", MoqtObjectStatus::kNormal, + 128, location.object == 0, + 15}; // Fragment 1 arrives (first 10 bytes). EXPECT_CALL(listener_,
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index 6ddd7f8..15eb900 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -70,7 +70,7 @@ constexpr uint64_t kDefaultLocalRequestId = 0; constexpr uint64_t kDefaultPeerRequestId = 1; const MoqtDataStreamType kDefaultSubgroupStreamType = - MoqtDataStreamType::Subgroup(2, 4, false, false); + MoqtDataStreamType::Subgroup(2, 4, false, false, true); constexpr MoqtPriority kDefaultPublisherPriority = 0x80; const TrackExtensions kNoExtensions; @@ -208,8 +208,9 @@ } quiche::QuicheBuffer buffer = framer.SerializeObjectHeader( object, - MoqtDataStreamType::Subgroup(*object.subgroup_id, object.object_id, - false, false), + MoqtDataStreamType::Subgroup( + *object.subgroup_id, object.object_id, false, false, + object.first_object_in_subgroup.value_or(true)), previous_object); size_t data_read = 0; if (visitor == nullptr) { // It's the first object in the stream @@ -1222,6 +1223,7 @@ /*extension_headers=*/"", /*object_status=*/MoqtObjectStatus::kNormal, /*subgroup_id=*/std::nullopt, + /*first_object_in_subgroup=*/std::nullopt, /*payload_length=*/8, }; char datagram[] = {0x00, 0x02, 0x00, 0x00, 0x00, 0x64, 0x65, @@ -1326,7 +1328,7 @@ .WillOnce(Return(PublishedObject{ PublishedObjectMetadata{ Location(0, 0), 0, "", MoqtObjectStatus::kNormal, - kLocalDefaultPriority, 8, MoqtSessionPeer::Now(&session_)}, + kLocalDefaultPriority, true, 8, MoqtSessionPeer::Now(&session_)}, PayloadFromString("deadbeef")})) .WillOnce(Return(std::nullopt)); EXPECT_CALL(mock_stream_, Writev) @@ -1341,9 +1343,10 @@ // Send a datagram with the default priority. EXPECT_CALL(*track, GetCachedObject(_, _, _, _)) .WillOnce(Return(PublishedObject{ - PublishedObjectMetadata{ - Location(0, 1), std::nullopt, "", MoqtObjectStatus::kNormal, - kLocalDefaultPriority, 8, MoqtSessionPeer::Now(&session_)}, + PublishedObjectMetadata{Location(0, 1), std::nullopt, "", + MoqtObjectStatus::kNormal, + kLocalDefaultPriority, std::nullopt, 8, + MoqtSessionPeer::Now(&session_)}, PayloadFromString("deadbeef")})); EXPECT_CALL(mock_session_, SendOrQueueDatagram) .WillOnce([](absl::string_view datagram) { @@ -1357,9 +1360,10 @@ // Non-default priority EXPECT_CALL(*track, GetCachedObject(_, _, _, _)) .WillOnce(Return(PublishedObject{ - PublishedObjectMetadata{ - Location(0, 2), std::nullopt, "", MoqtObjectStatus::kNormal, - kLocalDefaultPriority + 1, 8, MoqtSessionPeer::Now(&session_)}, + PublishedObjectMetadata{Location(0, 2), std::nullopt, "", + MoqtObjectStatus::kNormal, + kLocalDefaultPriority + 1, std::nullopt, 8, + MoqtSessionPeer::Now(&session_)}, PayloadFromString("deadbeef")})); EXPECT_CALL(mock_session_, SendOrQueueDatagram) .WillOnce([](absl::string_view datagram) { @@ -1610,7 +1614,7 @@ // Trigger stream opening (calls SetObjectAvailableCallback with lambda1). // Setting the stream visitor will cause a second call to the callback. PublishedObjectMetadata metadata = { - Location(0, 0), 0, "", MoqtObjectStatus::kNormal, 128, 10}; + Location(0, 0), 0, "", MoqtObjectStatus::kNormal, 128, true, 10}; EXPECT_CALL(*fetch_task, GetNextObject) .WillOnce([&](PublishedObject& output) { output.metadata = metadata; @@ -1894,6 +1898,7 @@ /*extension_headers=*/"", /*status=*/MoqtObjectStatus::kNormal, /*subgroup=*/0, + /*first_object_in_subgroup=*/true, /*payload_length=*/3, }; MoqtFramer framer(true, quic::Perspective::IS_SERVER); @@ -2124,6 +2129,7 @@ /*extension_headers=*/"", /*status=*/MoqtObjectStatus::kNormal, /*subgroup=*/0, + /*first_object_in_subgroup=*/true, /*payload_length=*/3, }; MoqtFramer framer(true, quic::Perspective::IS_SERVER); @@ -2197,6 +2203,7 @@ /*extension_headers=*/"", /*status=*/MoqtObjectStatus::kNormal, /*subgroup=*/0, + /*first_object_in_subgroup=*/true, /*payload_length=*/3, }; MoqtFramer framer(true, quic::Perspective::IS_SERVER); @@ -2544,6 +2551,7 @@ /*extension_headers=*/"", /*object_status=*/MoqtObjectStatus::kEndOfGroup, /*subgroup_id=*/0, + /*first_object_in_subgroup=*/true, /*payload_length=*/0, }; EXPECT_CALL(mock_stream_, GetStreamId()) @@ -2588,6 +2596,7 @@ /*extension_headers=*/"", /*object_status=*/MoqtObjectStatus::kEndOfGroup, /*subgroup_id=*/0, + /*first_object_in_subgroup=*/true, /*payload_length=*/0, }; EXPECT_CALL(mock_stream_, GetStreamId()) @@ -2755,7 +2764,7 @@ EXPECT_CALL(*track, GetCachedObject(0, Optional(1), 0, 0)) .WillOnce(Return(PublishedObject{ PublishedObjectMetadata{Location(0, 0), 1, "", - MoqtObjectStatus::kNormal, 0x80, 0, + MoqtObjectStatus::kNormal, 0x80, true, 0, MoqtSessionPeer::Now(&session_)}, PayloadFromString(""), false})); EXPECT_CALL(*track, GetCachedObject(0, Optional(1), 1, 0))
diff --git a/quiche/quic/moqt/moqt_subscription_test.cc b/quiche/quic/moqt/moqt_subscription_test.cc index 227b8cb..37e1da9 100644 --- a/quiche/quic/moqt/moqt_subscription_test.cc +++ b/quiche/quic/moqt/moqt_subscription_test.cc
@@ -109,6 +109,9 @@ object.metadata.status = MoqtObjectStatus::kNormal; object.metadata.publisher_priority = publisher_priority; object.metadata.extensions = "extensions"; + object.metadata.first_object_in_subgroup = + subgroup.has_value() ? std::optional<bool>(location.object == 0) + : std::nullopt; object.metadata.payload_length = 8; object.payload.push_back(quiche::QuicheMemSlice::Copy("deadbeef")); return object; @@ -344,7 +347,7 @@ TEST_F(SubscriptionPublisherTest, UpdatePriorityWithActiveStreams) { CreateStream( Location(1, 0), 0, 127, - {0x11, static_cast<uint8_t>(kTrackAlias), 0x01, 0x7f, 0x00, 0x0a}); + {0x51, static_cast<uint8_t>(kTrackAlias), 0x01, 0x7f, 0x00, 0x0a}); MessageParameters new_params; new_params.subscriber_priority = 20; EXPECT_CALL(mock_uni_stream_, SetPriority);
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_track_test.cc index 010157d..9e15a7f 100644 --- a/quiche/quic/moqt/moqt_track_test.cc +++ b/quiche/quic/moqt/moqt_track_test.cc
@@ -306,7 +306,8 @@ PublishedObject object; EXPECT_EQ(fetch_task_->GetNextObject(object), MoqtFetchTask::GetNextObjectResult::kPending); - MoqtObject new_object = {1, 3, 0, 128, "", MoqtObjectStatus::kNormal, 0, 6}; + MoqtObject new_object = {1, 3, 0, 128, "", MoqtObjectStatus::kNormal, + 0, true, 6}; bool got_object = false; fetch_task_->SetObjectAvailableCallback([&]() { got_object = true; @@ -341,7 +342,8 @@ TEST_F(UpstreamFetchTest, ObjectRetrievalEmptyPayload) { fetch_.OnFetchResult(Location(3, 50), absl::OkStatus(), nullptr); - MoqtObject moqt_obj = {1, 3, 0, 128, "", MoqtObjectStatus::kEndOfGroup, 0, 0}; + MoqtObject moqt_obj = {1, 3, 0, 128, "", MoqtObjectStatus::kEndOfGroup, + 0, true, 0}; fetch_.task()->NewObject(moqt_obj); fetch_.task()->NotifyNewObject(); fetch_.OnStreamOpened([]() {}); @@ -370,7 +372,7 @@ fetch_.OnFetchResult(largest, absl::OkStatus(), nullptr); fetch_.OnStreamOpened([]() {}); - MoqtObject obj1 = {1, 3, 49, 128, "", MoqtObjectStatus::kNormal, 0, 1}; + MoqtObject obj1 = {1, 3, 49, 128, "", MoqtObjectStatus::kNormal, 0, false, 1}; fetch_.task()->NewObject(obj1); fetch_.task()->AppendPayloadToObject("a"); fetch_.task()->NotifyNewObject(); @@ -382,7 +384,7 @@ EXPECT_EQ(fetch_task_->GetNextObject(out), MoqtFetchTask::GetNextObjectResult::kPending); - MoqtObject obj2 = {1, 3, 50, 128, "", MoqtObjectStatus::kNormal, 0, 1}; + MoqtObject obj2 = {1, 3, 50, 128, "", MoqtObjectStatus::kNormal, 0, false, 1}; fetch_.task()->NewObject(obj2); fetch_.task()->AppendPayloadToObject("b"); fetch_.task()->NotifyNewObject();
diff --git a/quiche/quic/moqt/moqt_uni_stream.cc b/quiche/quic/moqt/moqt_uni_stream.cc index 0f0456e..26ab6c2 100644 --- a/quiche/quic/moqt/moqt_uni_stream.cc +++ b/quiche/quic/moqt/moqt_uni_stream.cc
@@ -29,6 +29,7 @@ #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" #include "quiche/common/quiche_mem_slice.h" #include "quiche/common/quiche_weak_ptr.h" @@ -149,6 +150,10 @@ << "Received non-empty object with no payload"; return; } + QUICHE_BUG_IF(OutgoingSubgroupStream_SendObjects_no_first_object, + !object->metadata.first_object_in_subgroup.has_value()) + << "first_object_in_subgroup has to be set on all objects set via " + "subscription"; QUICHE_DCHECK_EQ(object->metadata.location.group, index_.group); QUICHE_DCHECK(object->metadata.subgroup == index_.subgroup); if (!visitor->InWindow(object->metadata.location)) { @@ -175,7 +180,8 @@ type_ = MoqtDataStreamType::Subgroup( index_.subgroup, next_object_, false, object->metadata.publisher_priority == - publisher_->extensions().default_publisher_priority()); + publisher_->extensions().default_publisher_priority(), + object->metadata.first_object_in_subgroup.value_or(true)); } uint64_t start_offset = already_delivered_; already_delivered_ += @@ -446,6 +452,7 @@ metadata.extensions = message.extension_headers; metadata.status = message.object_status; metadata.publisher_priority = message.publisher_priority; + metadata.first_object_in_subgroup = message.first_object_in_subgroup; metadata.payload_length = message.payload_length; metadata.arrival_time = clock_->Now(); visitor_->OnObjectFragment(track->full_track_name(), metadata, payload,
diff --git a/quiche/quic/moqt/moqt_uni_stream_test.cc b/quiche/quic/moqt/moqt_uni_stream_test.cc index 9b446ea..d69c198 100644 --- a/quiche/quic/moqt/moqt_uni_stream_test.cc +++ b/quiche/quic/moqt/moqt_uni_stream_test.cc
@@ -51,6 +51,7 @@ object.metadata.location = Location(0, 0); object.metadata.subgroup = 0; object.metadata.status = MoqtObjectStatus::kNormal; + object.metadata.first_object_in_subgroup = true; object.metadata.arrival_time = quic::QuicTime::Zero(); object.metadata.payload_length = 7; object.payload.push_back(quiche::QuicheMemSlice::Copy("payload")); @@ -463,8 +464,9 @@ 0x80, // publisher_priority "", // extension_headers MoqtObjectStatus::kNormal, - 0, // subgroup_id - 0, // payload_length + 0, // subgroup_id + true, // first_object_in_subgroup + 0, // payload_length }; class MockSessionToUniStreamInterface : public SessionToUniStreamInterface { @@ -539,7 +541,7 @@ } TEST_F(IncomingDataStreamTest, DestructorAfterObject) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); EXPECT_CALL(visitor_, OnObjectFragment); stream_->OnObjectMessage(kDefaultObject, "", true); @@ -548,7 +550,7 @@ } TEST_F(IncomingDataStreamTest, DestructorAfterFin) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); EXPECT_CALL(visitor_, OnObjectFragment); stream_->OnObjectMessage(kDefaultObject, "", true); @@ -570,7 +572,7 @@ } TEST_F(IncomingDataStreamTest, OnObjectMessage) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); MoqtObject object = kDefaultObject; object.payload_length = 8; @@ -592,7 +594,7 @@ } TEST_F(IncomingDataStreamTest, OnObjectMessageBufferPartialObject) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); MoqtObject object = kDefaultObject; object.payload_length = 6; @@ -611,7 +613,7 @@ TEST_F(IncomingDataStreamTest, OnObjectMessageDontBufferPartialObject) { EXPECT_CALL(session_, deliver_partial_objects()).WillRepeatedly(Return(true)); - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); MoqtObject object = kDefaultObject; object.payload_length = 6; @@ -673,7 +675,8 @@ MoqtObject sent_object = MoqtObject( /*request_id=*/0, /*group_id=*/0, /*object_id=*/0, /*publisher_priority=*/0x80, /*extension_headers=*/"", - MoqtObjectStatus::kNormal, /*subgroup_id=*/0, /*payload_length=*/12); + MoqtObjectStatus::kNormal, /*subgroup_id=*/0, + /*first_object_in_subgroup=*/true, /*payload_length=*/12); stream_->OnObjectMessage(sent_object, "foo", false); task->NotifyNewObject(); EXPECT_EQ(objects_available_callbacks, 1); @@ -703,7 +706,7 @@ } TEST_F(IncomingDataStreamTest, OnObjectMessageInvalidTrack) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); uint8_t alias = 2; mock_stream_.Receive( absl::string_view(reinterpret_cast<const char*>(&alias), 1), false); @@ -714,7 +717,7 @@ } TEST_F(IncomingDataStreamTest, OnObjectMessageNotInWindow) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); MessageParameters parameters; parameters.set_forward(false); @@ -724,7 +727,7 @@ } TEST_F(IncomingDataStreamTest, OnObjectMessageMissingSubgroupId) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); MoqtObject object = kDefaultObject; object.subgroup_id = std::nullopt; @@ -733,7 +736,7 @@ } TEST_F(IncomingDataStreamTest, ObjectAfterTrackEnd) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); MoqtObject object = kDefaultObject; object.object_status = MoqtObjectStatus::kEndOfTrack; @@ -748,7 +751,7 @@ } TEST_F(IncomingDataStreamTest, ObjectAfterGroupEnd) { - ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80)); + ProcessStreamType(MoqtDataStreamType::Subgroup(0, 0, false, 0x80, true)); ProcessAlias(2); MoqtObject object = kDefaultObject; object.object_status = MoqtObjectStatus::kEndOfGroup;
diff --git a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h index d106fb2..5915dfc 100644 --- a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h +++ b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h
@@ -155,6 +155,7 @@ metadata.extensions = ""; metadata.status = MoqtObjectStatus::kNormal; metadata.publisher_priority = 128; + metadata.first_object_in_subgroup = location.object == 0; metadata.payload_length = payload.length(); metadata.arrival_time = arrival_time; auto it = objects_.find(location);
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h index a19cc8a..4a01ac2 100644 --- a/quiche/quic/moqt/test_tools/moqt_test_message.h +++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -71,15 +71,18 @@ inline std::vector<MoqtDataStreamType> AllMoqtDataStreamTypes() { std::vector<MoqtDataStreamType> types; + types.reserve(256); types.push_back(MoqtDataStreamType::Fetch()); uint64_t first_object_id = 1; for (uint64_t subgroup_id : {0, 1, 2}) { for (bool no_extension_headers : {true, false}) { for (bool default_priority : {true, false}) { - for (bool end_of_group : {false, true}) { - types.push_back(MoqtDataStreamType::Subgroup( - subgroup_id, first_object_id, no_extension_headers, - default_priority, end_of_group)); + for (bool has_first_object : {true, false}) { + for (bool end_of_group : {false, true}) { + types.push_back(MoqtDataStreamType::Subgroup( + subgroup_id, first_object_id, no_extension_headers, + default_priority, has_first_object, end_of_group)); + } } } } @@ -275,15 +278,16 @@ } protected: - MoqtObject object_ = { - /*track_alias=*/4, - /*group_id*/ 5, - /*object_id=*/6, - /*publisher_priority=*/7, - std::string(kDefaultExtensionBlob), - /*object_status=*/MoqtObjectStatus::kNormal, - /*subgroup_id=*/8, - /*payload_length=*/3, + MoqtObject object_{ + .track_alias = 4, + .group_id = 5, + .object_id = 6, + .publisher_priority = 7, + .extension_headers = std::string(kDefaultExtensionBlob), + .object_status = MoqtObjectStatus::kNormal, + .subgroup_id = 8, + .first_object_in_subgroup = false, + .payload_length = 3, }; }; @@ -891,7 +895,10 @@ private: uint8_t raw_packet_[4] = { - 0x0a, 0x00, 0x01, 0x03, // request_id = 3 + 0x0a, + 0x00, + 0x01, + 0x03, // request_id = 3 }; MoqtUnsubscribe unsubscribe_ = { @@ -1638,7 +1645,9 @@ private: uint8_t raw_packet_[4] = { - 0x17, 0x00, 0x01, + 0x17, + 0x00, + 0x01, 0x01, // request_id = 1 }; @@ -1669,7 +1678,9 @@ private: uint8_t raw_packet_[4] = { - 0x1a, 0x00, 0x01, + 0x1a, + 0x00, + 0x01, 0x0b, // max_request_id = 11 };