diff --git a/quiche/quic/moqt/moqt_parser.h b/quiche/quic/moqt/moqt_parser.h
index 1b4de80..b05fdcc 100644
--- a/quiche/quic/moqt/moqt_parser.h
+++ b/quiche/quic/moqt/moqt_parser.h
@@ -202,6 +202,13 @@
   // Returns the type of the unidirectional stream, if already known.
   std::optional<MoqtDataStreamType> stream_type() const { return type_; }
 
+  // Returns the track alias, if already known.
+  std::optional<uint64_t> track_alias() const {
+    return (next_input_ == kStreamType || next_input_ == kTrackAlias)
+               ? std::optional<uint64_t>()
+               : metadata_.track_alias;
+  }
+
  private:
   friend class test::MoqtDataParserPeer;
 
diff --git a/quiche/quic/moqt/moqt_publisher.h b/quiche/quic/moqt/moqt_publisher.h
index 010db7f..0f378ed 100644
--- a/quiche/quic/moqt/moqt_publisher.h
+++ b/quiche/quic/moqt/moqt_publisher.h
@@ -77,7 +77,9 @@
 
   // Sets the callback that is called when GetNextObject() has previously
   // returned kPending, but now a new object (or potentially an error or an
-  // end-of-fetch) is available.
+  // end-of-fetch) is available. The application is responsible for calling
+  // GetNextObject() until it gets kPending; no further callback will occur
+  // until then.
   virtual void SetObjectAvailableCallback(
       ObjectsAvailableCallback callback) = 0;
 
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index 8496188..a7473b1 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -36,6 +36,7 @@
 #include "quiche/quic/platform/api/quic_logging.h"
 #include "quiche/common/platform/api/quiche_bug_tracker.h"
 #include "quiche/common/platform/api/quiche_logging.h"
+#include "quiche/common/platform/api/quiche_mem_slice.h"
 #include "quiche/common/quiche_buffer_allocator.h"
 #include "quiche/common/quiche_stream.h"
 #include "quiche/common/simple_buffer_allocator.h"
@@ -959,7 +960,6 @@
     subscribe->visitor()->OnReply(track->full_track_name(), message.largest_id,
                                   std::nullopt);
   }
-  subscribe->OnObjectOrOk();
 }
 
 void MoqtSession::ControlStream::OnSubscribeErrorMessage(
@@ -1385,19 +1385,114 @@
     return;
   }
   track->OnObjectOrOk();
-  SubscribeRemoteTrack* subscribe = static_cast<SubscribeRemoteTrack*>(track);
-  if (subscribe->visitor() != nullptr) {
-    subscribe->visitor()->OnObjectFragment(
-        track->full_track_name(),
-        FullSequence{message.group_id, message.subgroup_id.value_or(0),
-                     message.object_id},
-        message.publisher_priority, message.object_status, payload,
-        end_of_message);
+  if (!track->is_fetch()) {
+    SubscribeRemoteTrack* subscribe = static_cast<SubscribeRemoteTrack*>(track);
+    if (subscribe->visitor() != nullptr) {
+      subscribe->visitor()->OnObjectFragment(
+          track->full_track_name(),
+          FullSequence{message.group_id, message.subgroup_id.value_or(0),
+                       message.object_id},
+          message.publisher_priority, message.object_status, payload,
+          end_of_message);
+    }
+  } else {  // FETCH
+    UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
+    UpstreamFetch::UpstreamFetchTask* task = fetch->task();
+    if (task == nullptr) {
+      // The application killed the FETCH.
+      stream_->SendStopSending(kResetCodeSubscriptionGone);
+      return;
+    }
+    if (!task->HasObject()) {
+      task->NewObject(message);
+    }
+    if (task->NeedsMorePayload() && !payload.empty()) {
+      task->AppendPayloadToObject(payload);
+    }
   }
   partial_object_.clear();
 }
 
-void MoqtSession::IncomingDataStream::OnCanRead() { parser_.ReadAllData(); }
+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());
+    }
+  }
+}
+
+void MoqtSession::IncomingDataStream::MaybeReadOneObject() {
+  if (!parser_.track_alias().has_value() ||
+      parser_.stream_type() != MoqtDataStreamType::kStreamHeaderFetch) {
+    QUICHE_BUG(quic_bug_read_one_object_parser_unexpected_state)
+        << "Requesting object, parser in unexpected state";
+  }
+  RemoteTrack* track = session_->RemoteTrackById(*parser_.track_alias());
+  if (track == nullptr || !track->is_fetch()) {
+    QUICHE_BUG(quic_bug_read_one_object_track_unexpected_state)
+        << "Requesting object, track in unexpected state";
+    return;
+  }
+  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
+  UpstreamFetch::UpstreamFetchTask* task = fetch->task();
+  if (task == nullptr) {
+    return;
+  }
+  if (task->HasObject() && !task->NeedsMorePayload()) {
+    return;
+  }
+  parser_.ReadAtMostOneObject();
+  // If it read an object, it called OnObjectMessage and may have altered the
+  // task's object state.
+  if (task->HasObject() && !task->NeedsMorePayload()) {
+    task->NotifyNewObject();
+  }
+}
+
+void MoqtSession::IncomingDataStream::OnCanRead() {
+  if (!parser_.stream_type().has_value()) {
+    parser_.ReadStreamType();
+    if (!parser_.stream_type().has_value()) {
+      return;
+    }
+  }
+  if (parser_.stream_type() != MoqtDataStreamType::kStreamHeaderFetch) {
+    parser_.ReadAllData();
+    return;
+  }
+  bool learned_track_alias = false;
+  if (!parser_.track_alias().has_value()) {
+    learned_track_alias = true;
+    parser_.ReadTrackAlias();
+    if (!parser_.track_alias().has_value()) {
+      return;
+    }
+  }
+  auto it = session_->upstream_by_id_.find(*parser_.track_alias());
+  if (it == session_->upstream_by_id_.end()) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Received object for a track with no FETCH";
+    // This is a not a session error because there might be an UNSUBSCRIBE in
+    // flight.
+    stream_->SendStopSending(kResetCodeSubscriptionGone);
+    return;
+  }
+  if (it->second == nullptr) {
+    QUICHE_BUG(quiche_bug_moqt_fetch_pointer_is_null)
+        << "Fetch pointer is null";
+    return;
+  }
+  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(it->second.get());
+  if (learned_track_alias) {
+    // If the task already exists (FETCH_OK has arrived), the callback will
+    // immediately execute to read the first object. Otherwise, it will only
+    // execute when the task is created or a cached object is read.
+    fetch->OnStreamOpened([this]() { MaybeReadOneObject(); });
+    return;
+  }
+  MaybeReadOneObject();
+}
 
 void MoqtSession::IncomingDataStream::OnControlMessageReceived() {
   session_->Error(MoqtError::kProtocolViolation,
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index adf4b7e..18c02e2 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -314,6 +314,7 @@
    public:
     IncomingDataStream(MoqtSession* session, webtransport::Stream* stream)
         : session_(session), stream_(stream), parser_(stream, this) {}
+    ~IncomingDataStream();
 
     // webtransport::StreamVisitor implementation.
     void OnCanRead() override;
@@ -335,6 +336,8 @@
 
     webtransport::Stream* stream() const { return stream_; }
 
+    void MaybeReadOneObject();
+
    private:
     friend class test::MoqtSessionPeer;
     void OnControlMessageReceived();
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 172a152..ae2a3ca 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -8,6 +8,7 @@
 #include <cstring>
 #include <memory>
 #include <optional>
+#include <queue>
 #include <string>
 #include <utility>
 #include <vector>
@@ -19,17 +20,22 @@
 #include "quiche/quic/core/quic_data_reader.h"
 #include "quiche/quic/core/quic_time.h"
 #include "quiche/quic/core/quic_types.h"
+#include "quiche/quic/moqt/moqt_framer.h"
 #include "quiche/quic/moqt/moqt_known_track_publisher.h"
 #include "quiche/quic/moqt/moqt_messages.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_track.h"
 #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h"
 #include "quiche/quic/moqt/test_tools/moqt_session_peer.h"
 #include "quiche/quic/moqt/tools/moqt_mock_visitor.h"
 #include "quiche/quic/platform/api/quic_test.h"
 #include "quiche/quic/test_tools/quic_test_utils.h"
+#include "quiche/common/quiche_buffer_allocator.h"
 #include "quiche/common/quiche_stream.h"
+#include "quiche/common/simple_buffer_allocator.h"
+#include "quiche/web_transport/test_tools/in_memory_stream.h"
 #include "quiche/web_transport/test_tools/mock_web_transport.h"
 #include "quiche/web_transport/web_transport.h"
 
@@ -2256,6 +2262,218 @@
   EXPECT_EQ(fetch_task->GetStatus().message(), "No username provided");
 }
 
+// The application takes objects as they arrive.
+TEST_F(MoqtSessionTest, IncomingFetchObjectsGreedyApp) {
+  webtransport::test::MockStream mock_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
+  std::unique_ptr<MoqtFetchTask> fetch_task;
+  uint64_t expected_object_id = 0;
+  session_.Fetch(
+      FullTrackName("foo", "bar"),
+      [&](std::unique_ptr<MoqtFetchTask> task) {
+        fetch_task = std::move(task);
+        fetch_task->SetObjectAvailableCallback([&]() {
+          PublishedObject object;
+          MoqtFetchTask::GetNextObjectResult result;
+          do {
+            result = fetch_task->GetNextObject(object);
+            if (result == MoqtFetchTask::GetNextObjectResult::kSuccess) {
+              EXPECT_EQ(object.sequence.object, expected_object_id);
+              ++expected_object_id;
+            }
+          } while (result != MoqtFetchTask::GetNextObjectResult::kPending);
+        });
+      },
+      FullSequence(0, 0), 4, std::nullopt, 128, std::nullopt,
+      MoqtSubscribeParameters());
+  // Build queue of packets to arrive.
+  std::queue<quiche::QuicheBuffer> headers;
+  std::queue<std::string> payloads;
+  MoqtObject object = {
+      /*subscribe_id=*/0,
+      /*group_id, object_id=*/0,
+      0,
+      /*publisher_priority=*/128,
+      /*status=*/MoqtObjectStatus::kNormal,
+      /*subgroup=*/0,
+      /*payload_length=*/3,
+  };
+  MoqtFramer framer_(quiche::SimpleBufferAllocator::Get(), true);
+  for (int i = 0; i < 4; ++i) {
+    object.object_id = i;
+    headers.push(framer_.SerializeObjectHeader(
+        object, MoqtDataStreamType::kStreamHeaderFetch, i == 0));
+    payloads.push("foo");
+  }
+
+  // Open stream, deliver two objects before FETCH_OK. Neither should be read.
+  webtransport::test::InMemoryStream data_stream(kIncomingUniStreamId);
+  data_stream.SetVisitor(
+      MoqtSessionPeer::CreateIncomingStreamVisitor(&session_, &data_stream));
+  for (int i = 0; i < 2; ++i) {
+    data_stream.Receive(headers.front().AsStringView(), false);
+    data_stream.Receive(payloads.front(), false);
+    headers.pop();
+    payloads.pop();
+  }
+  EXPECT_EQ(fetch_task, nullptr);
+  EXPECT_GT(data_stream.ReadableBytes(), 0);
+
+  // FETCH_OK arrives, objects are delivered.
+  MoqtFetchOk ok = {
+      /*subscribe_id=*/0,
+      /*group_order=*/MoqtDeliveryOrder::kAscending,
+      /*largest_id=*/FullSequence(3, 25),
+      MoqtSubscribeParameters(),
+  };
+  stream_input->OnFetchOkMessage(ok);
+  ASSERT_NE(fetch_task, nullptr);
+  EXPECT_EQ(expected_object_id, 2);
+
+  // Deliver the rest of the objects.
+  for (int i = 2; i < 4; ++i) {
+    data_stream.Receive(headers.front().AsStringView(), false);
+    data_stream.Receive(payloads.front(), false);
+    headers.pop();
+    payloads.pop();
+  }
+  EXPECT_EQ(expected_object_id, 4);
+}
+
+TEST_F(MoqtSessionTest, IncomingFetchObjectsSlowApp) {
+  webtransport::test::MockStream mock_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
+  std::unique_ptr<MoqtFetchTask> fetch_task;
+  uint64_t expected_object_id = 0;
+  bool objects_available = false;
+  session_.Fetch(
+      FullTrackName("foo", "bar"),
+      [&](std::unique_ptr<MoqtFetchTask> task) {
+        fetch_task = std::move(task);
+        fetch_task->SetObjectAvailableCallback(
+            [&]() { objects_available = true; });
+      },
+      FullSequence(0, 0), 4, std::nullopt, 128, std::nullopt,
+      MoqtSubscribeParameters());
+  // Build queue of packets to arrive.
+  std::queue<quiche::QuicheBuffer> headers;
+  std::queue<std::string> payloads;
+  MoqtObject object = {
+      /*subscribe_id=*/0,
+      /*group_id, object_id=*/0,
+      0,
+      /*publisher_priority=*/128,
+      /*status=*/MoqtObjectStatus::kNormal,
+      /*subgroup=*/0,
+      /*payload_length=*/3,
+  };
+  MoqtFramer framer_(quiche::SimpleBufferAllocator::Get(), true);
+  for (int i = 0; i < 4; ++i) {
+    object.object_id = i;
+    headers.push(framer_.SerializeObjectHeader(
+        object, MoqtDataStreamType::kStreamHeaderFetch, i == 0));
+    payloads.push("foo");
+  }
+
+  // Open stream, deliver two objects before FETCH_OK. Neither should be read.
+  webtransport::test::InMemoryStream data_stream(kIncomingUniStreamId);
+  data_stream.SetVisitor(
+      MoqtSessionPeer::CreateIncomingStreamVisitor(&session_, &data_stream));
+  for (int i = 0; i < 2; ++i) {
+    data_stream.Receive(headers.front().AsStringView(), false);
+    data_stream.Receive(payloads.front(), false);
+    headers.pop();
+    payloads.pop();
+  }
+  EXPECT_EQ(fetch_task, nullptr);
+  EXPECT_GT(data_stream.ReadableBytes(), 0);
+
+  // FETCH_OK arrives, objects are available.
+  MoqtFetchOk ok = {
+      /*subscribe_id=*/0,
+      /*group_order=*/MoqtDeliveryOrder::kAscending,
+      /*largest_id=*/FullSequence(3, 25),
+      MoqtSubscribeParameters(),
+  };
+  stream_input->OnFetchOkMessage(ok);
+  ASSERT_NE(fetch_task, nullptr);
+  EXPECT_TRUE(objects_available);
+
+  // Get the objects
+  MoqtFetchTask::GetNextObjectResult result;
+  do {
+    PublishedObject new_object;
+    result = fetch_task->GetNextObject(new_object);
+    if (result == MoqtFetchTask::GetNextObjectResult::kSuccess) {
+      EXPECT_EQ(new_object.sequence.object, expected_object_id);
+      ++expected_object_id;
+    }
+  } while (result != MoqtFetchTask::GetNextObjectResult::kPending);
+  EXPECT_EQ(expected_object_id, 2);
+  objects_available = false;
+
+  // Deliver the rest of the objects.
+  for (int i = 2; i < 4; ++i) {
+    data_stream.Receive(headers.front().AsStringView(), false);
+    data_stream.Receive(payloads.front(), false);
+    headers.pop();
+    payloads.pop();
+  }
+  EXPECT_TRUE(objects_available);
+  EXPECT_EQ(expected_object_id, 2);  // Not delivered yet.
+  // Get the objects
+  do {
+    PublishedObject new_object;
+    result = fetch_task->GetNextObject(new_object);
+    if (result == MoqtFetchTask::GetNextObjectResult::kSuccess) {
+      EXPECT_EQ(new_object.sequence.object, expected_object_id);
+      ++expected_object_id;
+    }
+  } while (result != MoqtFetchTask::GetNextObjectResult::kPending);
+  EXPECT_EQ(expected_object_id, 4);
+}
+
+TEST_F(MoqtSessionTest, PartialObjectFetch) {
+  MoqtSessionParameters parameters(quic::Perspective::IS_CLIENT);
+  parameters.deliver_partial_objects = true;
+  MoqtSession session(&mock_session_, parameters,
+                      session_callbacks_.AsSessionCallbacks());
+  webtransport::test::InMemoryStream stream(kIncomingUniStreamId);
+  std::unique_ptr<MoqtFetchTask> fetch_task =
+      MoqtSessionPeer::CreateUpstreamFetch(&session, &stream);
+  UpstreamFetch::UpstreamFetchTask* task =
+      static_cast<UpstreamFetch::UpstreamFetchTask*>(fetch_task.get());
+  ASSERT_NE(task, nullptr);
+  EXPECT_FALSE(task->HasObject());
+  bool object_ready = false;
+  task->SetObjectAvailableCallback([&]() { object_ready = true; });
+  MoqtObject object = {
+      /*subscribe_id=*/0,
+      /*group_id, object_id=*/0,
+      0,
+      /*publisher_priority=*/128,
+      /*status=*/MoqtObjectStatus::kNormal,
+      /*subgroup=*/0,
+      /*payload_length=*/6,
+  };
+  MoqtFramer framer_(quiche::SimpleBufferAllocator::Get(), true);
+  quiche::QuicheBuffer header = framer_.SerializeObjectHeader(
+      object, MoqtDataStreamType::kStreamHeaderFetch, true);
+  stream.Receive(header.AsStringView(), false);
+  EXPECT_FALSE(task->HasObject());
+  EXPECT_FALSE(object_ready);
+  stream.Receive("foo", false);
+  EXPECT_TRUE(task->HasObject());
+  EXPECT_TRUE(task->NeedsMorePayload());
+  EXPECT_FALSE(object_ready);
+  stream.Receive("bar", false);
+  EXPECT_TRUE(object_ready);
+  EXPECT_TRUE(task->HasObject());
+  EXPECT_FALSE(task->NeedsMorePayload());
+}
+
 // TODO: re-enable this test once this behavior is re-implemented.
 #if 0
 TEST_F(MoqtSessionTest, SubscribeUpdateClosesSubscription) {
diff --git a/quiche/quic/moqt/moqt_track.cc b/quiche/quic/moqt/moqt_track.cc
index 31980d1..d8e31a3 100644
--- a/quiche/quic/moqt/moqt_track.cc
+++ b/quiche/quic/moqt/moqt_track.cc
@@ -4,6 +4,7 @@
 
 #include "quiche/quic/moqt/moqt_track.h"
 
+#include <cstring>
 #include <memory>
 #include <optional>
 #include <utility>
@@ -12,7 +13,10 @@
 #include "absl/strings/string_view.h"
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_publisher.h"
-#include "quiche/common/platform/api/quiche_logging.h"
+#include "quiche/common/platform/api/quiche_bug_tracker.h"
+#include "quiche/common/platform/api/quiche_mem_slice.h"
+#include "quiche/common/quiche_buffer_allocator.h"
+#include "quiche/common/simple_buffer_allocator.h"
 #include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
@@ -39,6 +43,18 @@
                                                   std::move(callback));
   task_ = task->weak_ptr();
   std::move(ok_callback_)(std::move(task));
+  if (can_read_callback_) {
+    task_.GetIfAvailable()->set_can_read_callback(
+        std::move(can_read_callback_));
+  }
+}
+
+void UpstreamFetch::OnStreamOpened(CanReadCallback can_read_callback) {
+  if (task_.IsValid()) {
+    task_.GetIfAvailable()->set_can_read_callback(std::move(can_read_callback));
+  } else {
+    can_read_callback_ = std::move(can_read_callback);
+  }
 }
 
 UpstreamFetch::UpstreamFetchTask::~UpstreamFetchTask() {
@@ -56,17 +72,47 @@
     if (eof_) {
       return kEof;
     }
+    need_object_available_callback_ = true;
     return kPending;
   }
-  output = *std::move(next_object_);
+  quiche::QuicheMemSlice message_slice(std::move(payload_));
+  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;
   next_object_.reset();
+  can_read_callback_();
   return kSuccess;
 }
 
-void UpstreamFetch::UpstreamFetchTask::NewObject(PublishedObject& object) {
-  QUICHE_DCHECK(!next_object_.has_value());
-  next_object_ = std::move(object);
-  if (object_available_callback_) {
+void UpstreamFetch::UpstreamFetchTask::NewObject(const MoqtObject& message) {
+  next_object_ = message;
+  payload_ = quiche::QuicheBuffer(quiche::SimpleBufferAllocator::Get(),
+                                  message.payload_length);
+}
+
+void UpstreamFetch::UpstreamFetchTask::AppendPayloadToObject(
+    absl::string_view payload) {
+  QUICHE_BUG_IF(quic_bug_AppendPayloadToObjectCalledEarly,
+                !next_object_.has_value())
+      << "AppendPayloadToObject called without an object";
+  QUICHE_BUG_IF(quic_bug_AlreadyGotPayload, next_object_->payload_length == 0)
+      << "AppendPayloadToObject called after payload was already full";
+  // Copy |payload| to the right spot in the buffer.
+  memcpy(payload_.data() + payload_.size() - next_object_->payload_length,
+         payload.data(), payload.length());
+  next_object_->payload_length -= payload.length();
+}
+
+void UpstreamFetch::UpstreamFetchTask::NotifyNewObject() {
+  QUICHE_BUG_IF(quic_bug_NotifyNewObjectCalledEarly,
+                !next_object_.has_value() || next_object_->payload_length > 0)
+      << "NotifyNewObject called without a full object in store";
+  if (need_object_available_callback_ && object_available_callback_) {
+    need_object_available_callback_ = false;
     object_available_callback_();
   }
 }
diff --git a/quiche/quic/moqt/moqt_track.h b/quiche/quic/moqt/moqt_track.h
index fe158e1..6c8bab3 100644
--- a/quiche/quic/moqt/moqt_track.h
+++ b/quiche/quic/moqt/moqt_track.h
@@ -17,6 +17,7 @@
 #include "quiche/quic/moqt/moqt_priority.h"
 #include "quiche/quic/moqt/moqt_publisher.h"
 #include "quiche/quic/moqt/moqt_subscribe_windows.h"
+#include "quiche/common/quiche_buffer_allocator.h"
 #include "quiche/common/quiche_callbacks.h"
 #include "quiche/common/quiche_weak_ptr.h"
 #include "quiche/web_transport/web_transport.h"
@@ -195,13 +196,27 @@
       return weak_ptr_factory_.Create();
     }
 
-    // Manage the relationship with the data stream.
-    void OnStreamOpened(CanReadCallback callback) {
+    // MoqtSession should not use this function; use
+    // UpstreamFetch::OnStreamOpened() instead, in case the task does not exist
+    // yet.
+    void set_can_read_callback(CanReadCallback callback) {
       can_read_callback_ = std::move(callback);
+      can_read_callback_();  // Accept the first object.
     }
 
     // Called when the data stream receives a new object.
-    void NewObject(PublishedObject& object);
+    void NewObject(const MoqtObject& message);
+    void AppendPayloadToObject(absl::string_view payload);
+    // MoqtSession calls this for a hint if the object has been read.
+    bool HasObject() const { return next_object_.has_value(); }
+    bool NeedsMorePayload() const {
+      return next_object_.has_value() && next_object_->payload_length > 0;
+    }
+    // MoqtSession calls NotifyNewObject() after NewObject() because it has to
+    // exit the parser loop before the callback possibly causes another read.
+    // Furthermore, NewObject() may be a partial object, and so
+    // NotifyNewObject() is called only when the object is complete.
+    void NotifyNewObject();
 
     // Deletes callbacks to session or stream, updates the status. If |error|
     // has no value, will append an EOF to the object stream.
@@ -214,10 +229,22 @@
     absl::Status status_;
     TaskDestroyedCallback task_destroyed_callback_;
 
-    // Object delivery state.
-    std::optional<PublishedObject> next_object_;
+    // Object delivery state. The payload_length member is used to track the
+    // payload bytes not yet received. The application receives a
+    // PublishedObject that is constructed from next_object_ and payload_.
+    std::optional<MoqtObject> next_object_;
+    // Store payload separately. Will be converted into QuicheMemSlice only when
+    // complete, since QuicheMemSlice is immutable.
+    quiche::QuicheBuffer payload_;
+
+    // The task should only call object_available_callback_ when the last result
+    // was kPending. Otherwise, there can be recursive loops of
+    // GetNextObjectResult().
+    bool need_object_available_callback_ = true;
     bool eof_ = false;  // The next object is EOF.
+    // The Fetch task signals the application when it has new objects.
     ObjectsAvailableCallback object_available_callback_;
+    // The Fetch task signals the stream when it has dispensed of an object.
     CanReadCallback can_read_callback_;
 
     // Must be last.
@@ -230,9 +257,16 @@
 
   UpstreamFetchTask* task() { return task_.GetIfAvailable(); }
 
+  // Manage the relationship with the data stream.
+  void OnStreamOpened(CanReadCallback callback);
+
  private:
   quiche::QuicheWeakPtr<UpstreamFetchTask> task_;
 
+  // Before FetchTask is created, an incoming stream will register the callback
+  // here instead.
+  CanReadCallback can_read_callback_;
+
   // Initial values from Fetch() call.
   FetchResponseCallback ok_callback_;  // Will be destroyed on FETCH_OK.
 };
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_track_test.cc
index 8072535..ecf1ed3 100644
--- a/quiche/quic/moqt/moqt_track_test.cc
+++ b/quiche/quic/moqt/moqt_track_test.cc
@@ -149,15 +149,35 @@
   PublishedObject object;
   EXPECT_EQ(fetch_task_->GetNextObject(object),
             MoqtFetchTask::GetNextObjectResult::kPending);
-  PublishedObject new_object(FullSequence(3, 0),
-                             MoqtObjectStatus::kGroupDoesNotExist, 128,
-                             quiche::QuicheMemSlice(), false);
+  MoqtObject new_object = {1, 3, 0, 128, MoqtObjectStatus::kNormal, 0, 6};
+  bool got_object = false;
   fetch_task_->SetObjectAvailableCallback([&]() {
+    got_object = true;
     EXPECT_EQ(fetch_task_->GetNextObject(object),
               MoqtFetchTask::GetNextObjectResult::kSuccess);
-    EXPECT_EQ(object.sequence, new_object.sequence);
+    EXPECT_EQ(object.sequence, FullSequence(3, 0, 0));
+    EXPECT_EQ(object.payload.AsStringView(), "foobar");
   });
+  int got_read_callback = 0;
+  fetch_.OnStreamOpened([&]() { ++got_read_callback; });
+  EXPECT_FALSE(fetch_.task()->HasObject());
+  EXPECT_FALSE(fetch_.task()->NeedsMorePayload());
   fetch_.task()->NewObject(new_object);
+  EXPECT_TRUE(fetch_.task()->HasObject());
+  EXPECT_TRUE(fetch_.task()->NeedsMorePayload());
+  fetch_.task()->AppendPayloadToObject("foo");
+  EXPECT_TRUE(fetch_.task()->HasObject());
+  EXPECT_TRUE(fetch_.task()->NeedsMorePayload());
+  fetch_.task()->AppendPayloadToObject("bar");
+  EXPECT_TRUE(fetch_.task()->HasObject());
+  EXPECT_FALSE(fetch_.task()->NeedsMorePayload());
+  EXPECT_FALSE(got_object);
+  EXPECT_EQ(got_read_callback, 1);  // Call from OnStreamOpened().
+  fetch_.task()->NotifyNewObject();
+  EXPECT_FALSE(fetch_.task()->HasObject());
+  EXPECT_FALSE(fetch_.task()->NeedsMorePayload());
+  EXPECT_EQ(got_read_callback, 2);  // Call from GetNextObjectResult().
+  EXPECT_TRUE(got_object);
 }
 
 }  // namespace test
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h
index e867376..dc94e7b 100644
--- a/quiche/quic/moqt/test_tools/moqt_session_peer.h
+++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -10,6 +10,8 @@
 #include <optional>
 #include <utility>
 
+
+#include "absl/status/status.h"
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_parser.h"
 #include "quiche/quic/moqt/moqt_priority.h"
@@ -57,6 +59,14 @@
     return new_stream;
   }
 
+  static std::unique_ptr<webtransport::StreamVisitor>
+  CreateIncomingStreamVisitor(MoqtSession* session,
+                              webtransport::Stream* stream) {
+    auto new_stream =
+        std::make_unique<MoqtSession::IncomingDataStream>(session, stream);
+    return new_stream;
+  }
+
   // In the test OnSessionReady, the session creates a stream and then passes
   // its unique_ptr to the mock webtransport stream. This function casts
   // that unique_ptr into a MoqtSession::Stream*, which is a private class of
@@ -156,6 +166,43 @@
                                                  uint64_t subscribe_id) {
     return *session->published_subscriptions_[subscribe_id]->largest_sent();
   }
+
+  // Adds an upstream fetch and a stream ready to receive data.
+  static std::unique_ptr<MoqtFetchTask> CreateUpstreamFetch(
+      MoqtSession* session, webtransport::Stream* stream) {
+    MoqtFetch fetch_message = {
+        0,
+        FullTrackName{"foo", "bar"},
+        128,
+        std::nullopt,
+        FullSequence{0, 0},
+        4,
+        std::nullopt,
+        MoqtSubscribeParameters(),
+    };
+    std::unique_ptr<MoqtFetchTask> task;
+    auto [it, success] = session->upstream_by_id_.try_emplace(
+        0, std::make_unique<UpstreamFetch>(
+               fetch_message, [&](std::unique_ptr<MoqtFetchTask> fetch_task) {
+                 task = std::move(fetch_task);
+               }));
+    QUICHE_DCHECK(success);
+    UpstreamFetch* fetch = static_cast<UpstreamFetch*>(it->second.get());
+    // Initialize the fetch task
+    fetch->OnFetchResult(
+        FullSequence{4, 10}, absl::OkStatus(),
+        [=, session_ptr = session, fetch_id = fetch_message.subscribe_id]() {
+          session_ptr->CancelFetch(fetch_id);
+        });
+    ;
+    auto mock_session =
+        static_cast<webtransport::test::MockSession*>(session->session());
+    EXPECT_CALL(*mock_session, AcceptIncomingUnidirectionalStream())
+        .WillOnce(testing::Return(stream))
+        .WillOnce(testing::Return(nullptr));
+    session->OnIncomingUnidirectionalStreamAvailable();
+    return task;
+  }
 };
 
 }  // namespace moqt::test