MoQT Upstream FETCH data streams.
Also, added track_alias() query to Parser API.
PiperOrigin-RevId: 710104119
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
