Refresh MoQT integration test for FETCH and fix detected errors.

PiperOrigin-RevId: 712510813

Author: martinduke

quiche/quic/moqt/moqt_integration_test.cc

@@ -16,6 +16,7 @@
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_outgoing_queue.h"
 #include "quiche/quic/moqt/moqt_priority.h"
+#include "quiche/quic/moqt/moqt_publisher.h"
 #include "quiche/quic/moqt/moqt_session.h"
 #include "quiche/quic/moqt/moqt_track.h"
 #include "quiche/quic/moqt/test_tools/moqt_simulator_harness.h"
@@ -336,68 +337,55 @@ TEST_F(MoqtIntegrationTest, SendMultipleGroups) {
   }
 }
 
-// TODO(martinduke): Restore this test when FETCH is implemented.
-#if 0
 TEST_F(MoqtIntegrationTest, FetchItemsFromPast) {
   EstablishSession();
   MoqtKnownTrackPublisher publisher;
   server_->session()->set_publisher(&publisher);
 
-  for (MoqtForwardingPreference forwarding_preference :
-       {MoqtForwardingPreference::kSubgroup,
-        MoqtForwardingPreference::kDatagram}) {
-    SCOPED_TRACE(MoqtForwardingPreferenceToString(forwarding_preference));
-    MockSubscribeRemoteTrackVisitor client_visitor;
-    std::string name =
-        absl::StrCat("pref_", static_cast<int>(forwarding_preference));
-    auto queue = std::make_shared<MoqtOutgoingQueue>(
-        FullTrackName{"test", name}, forwarding_preference);
-    publisher.Add(queue);
-    for (int i = 0; i < 100; ++i) {
-      queue->AddObject(MemSliceFromString("object"), /*key=*/true);
-    }
-
-    client_->session()->SubscribeAbsolute(FullTrackName("test", name), 0, 0,
-                                          &client_visitor);
-    int received = 0;
-    // Those won't arrive since they have expired.
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{0, 0}, _, _, _, true))
-        .Times(0);
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{0, 0}, _, _, _, true))
-        .Times(0);
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{96, 0}, _, _, _, true))
-        .Times(0);
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{96, 0}, _, _, _, true))
-        .Times(0);
-    // Those are within the "last three groups" window.
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{97, 0}, _, _, _, true))
-        .WillOnce([&] { ++received; });
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{97, 1}, _, _, _, true))
-        .WillOnce([&] { ++received; });
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{98, 0}, _, _, _, true))
-        .WillOnce([&] { ++received; });
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{98, 1}, _, _, _, true))
-        .WillOnce([&] { ++received; });
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{99, 0}, _, _, _, true))
-        .WillOnce([&] { ++received; });
-    EXPECT_CALL(client_visitor,
-                OnObjectFragment(_, FullSequence{99, 1}, _, _, _, true))
-        .Times(0);  // The current group should not be closed yet.
-    bool success = test_harness_.RunUntilWithDefaultTimeout(
-        [&]() { return received >= 5; });
-    EXPECT_TRUE(success);
+  MockSubscribeRemoteTrackVisitor client_visitor;
+  FullTrackName full_track_name("test", "fetch");
+  auto queue = std::make_shared<MoqtOutgoingQueue>(
+      full_track_name, MoqtForwardingPreference::kSubgroup);
+  publisher.Add(queue);
+  for (int i = 0; i < 100; ++i) {
+    queue->AddObject(MemSliceFromString("object"), /*key=*/true);
   }
+  std::unique_ptr<MoqtFetchTask> fetch;
+  EXPECT_TRUE(client_->session()->Fetch(
+      full_track_name,
+      [&](std::unique_ptr<MoqtFetchTask> task) { fetch = std::move(task); },
+      FullSequence{0, 0}, 99, std::nullopt, 128, std::nullopt,
+      MoqtSubscribeParameters()));
+  // Run until we get FETCH_OK.
+  bool success = test_harness_.RunUntilWithDefaultTimeout(
+      [&]() { return fetch != nullptr; });
+  EXPECT_TRUE(success);
+
+  EXPECT_TRUE(fetch->GetStatus().ok());
+  EXPECT_EQ(fetch->GetLargestId(), FullSequence(99, 0));
+  MoqtFetchTask::GetNextObjectResult result;
+  PublishedObject object;
+  FullSequence expected{97, 0};
+  do {
+    result = fetch->GetNextObject(object);
+    if (result == MoqtFetchTask::GetNextObjectResult::kEof) {
+      break;
+    }
+    EXPECT_EQ(result, MoqtFetchTask::GetNextObjectResult::kSuccess);
+    EXPECT_EQ(object.sequence, expected);
+    if (object.sequence.object == 1) {
+      EXPECT_EQ(object.status, MoqtObjectStatus::kEndOfGroup);
+      expected.object = 0;
+      ++expected.group;
+    } else {
+      EXPECT_EQ(object.status, MoqtObjectStatus::kNormal);
+      EXPECT_EQ(object.payload.AsStringView(), "object");
+      ++expected.object;
+    }
+  } while (result == MoqtFetchTask::GetNextObjectResult::kSuccess);
+  EXPECT_EQ(result, MoqtFetchTask::GetNextObjectResult::kEof);
+  EXPECT_EQ(expected, FullSequence(99, 1));
 }
-#endif
 
 TEST_F(MoqtIntegrationTest, AnnounceFailure) {
   EstablishSession();

quiche/quic/moqt/moqt_session.cc

@@ -1415,11 +1415,9 @@ void MoqtSession::IncomingDataStream::OnObjectMessage(const MoqtObject& message,
 
 MoqtSession::IncomingDataStream::~IncomingDataStream() {
   if (parser_.track_alias().has_value() &&
-      parser_.stream_type() == MoqtDataStreamType::kStreamHeaderFetch) {
-    RemoteTrack* track = session_->RemoteTrackById(*parser_.track_alias());
-    if (track != nullptr && track->is_fetch()) {
-      session_->upstream_by_id_.erase(*parser_.track_alias());
-    }
+      parser_.stream_type() == MoqtDataStreamType::kStreamHeaderFetch &&
+      track_.IsValid()) {
+    session_->upstream_by_id_.erase(*parser_.track_alias());
   }
 }
 

quiche/quic/moqt/moqt_track.cc

@@ -75,14 +75,19 @@ UpstreamFetch::UpstreamFetchTask::GetNextObject(PublishedObject& output) {
     need_object_available_callback_ = true;
     return kPending;
   }
-  quiche::QuicheMemSlice message_slice(std::move(payload_));
+  if (!payload_.empty()) {
+    quiche::QuicheMemSlice message_slice(std::move(payload_));
+    output.payload = std::move(message_slice);
+  }
   output.sequence = FullSequence(next_object_->group_id,
                                  next_object_->subgroup_id.value_or(0),
                                  next_object_->object_id);
   output.status = next_object_->object_status;
   output.publisher_priority = next_object_->publisher_priority;
-  output.payload = std::move(message_slice);
   output.fin_after_this = false;
+  if (output.sequence == largest_id_) {  // This is the last object.
+    eof_ = true;
+  }
   next_object_.reset();
   can_read_callback_();
   return kSuccess;