Add a method to send "End of Track" to the queue.
PiperOrigin-RevId: 742352459
diff --git a/quiche/quic/moqt/moqt_integration_test.cc b/quiche/quic/moqt/moqt_integration_test.cc
index 72b7ca4..3c8f001 100644
--- a/quiche/quic/moqt/moqt_integration_test.cc
+++ b/quiche/quic/moqt/moqt_integration_test.cc
@@ -372,6 +372,19 @@
success = test_harness_.RunUntilWithDefaultTimeout(
[&]() { return received >= 7; });
EXPECT_TRUE(success);
+
+ EXPECT_CALL(client_visitor,
+ OnObjectFragment(_, FullSequence{2, 2}, _,
+ MoqtObjectStatus::kEndOfGroup, "", true))
+ .WillOnce([&] { ++received; });
+ EXPECT_CALL(client_visitor,
+ OnObjectFragment(_, FullSequence{3, 0}, _,
+ MoqtObjectStatus::kEndOfTrack, "", true))
+ .WillOnce([&] { ++received; });
+ queue->Close();
+ success = test_harness_.RunUntilWithDefaultTimeout(
+ [&]() { return received >= 9; });
+ EXPECT_TRUE(success);
}
}
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.cc b/quiche/quic/moqt/moqt_outgoing_queue.cc
index 2c05a70..d6ba015 100644
--- a/quiche/quic/moqt/moqt_outgoing_queue.cc
+++ b/quiche/quic/moqt/moqt_outgoing_queue.cc
@@ -33,25 +33,33 @@
"flag.";
return;
}
-
- if (key) {
- if (!queue_.empty()) {
- AddRawObject(MoqtObjectStatus::kEndOfGroup, quiche::QuicheMemSlice());
- }
-
- if (queue_.size() == kMaxQueuedGroups) {
- queue_.erase(queue_.begin());
- for (MoqtObjectListener* listener : listeners_) {
- listener->OnGroupAbandoned(current_group_id_ - kMaxQueuedGroups + 1);
- }
- }
- queue_.emplace_back();
- ++current_group_id_;
+ if (closed_) {
+ QUICHE_BUG(MoqtOutgoingQueue_AddObject_closed)
+ << "Trying to send objects on a closed queue.";
+ return;
}
+ if (key) {
+ OpenNewGroup();
+ }
AddRawObject(MoqtObjectStatus::kNormal, std::move(payload));
}
+void MoqtOutgoingQueue::OpenNewGroup() {
+ if (!queue_.empty()) {
+ AddRawObject(MoqtObjectStatus::kEndOfGroup, quiche::QuicheMemSlice());
+ }
+
+ if (queue_.size() == kMaxQueuedGroups) {
+ queue_.erase(queue_.begin());
+ for (MoqtObjectListener* listener : listeners_) {
+ listener->OnGroupAbandoned(current_group_id_ - kMaxQueuedGroups + 1);
+ }
+ }
+ queue_.emplace_back();
+ ++current_group_id_;
+}
+
void MoqtOutgoingQueue::AddRawObject(MoqtObjectStatus status,
quiche::QuicheMemSlice payload) {
FullSequence sequence{current_group_id_, queue_.back().size()};
@@ -101,6 +109,9 @@
}
absl::StatusOr<MoqtTrackStatusCode> MoqtOutgoingQueue::GetTrackStatus() const {
+ if (closed_) {
+ return MoqtTrackStatusCode::kFinished;
+ }
if (queue_.empty()) {
return MoqtTrackStatusCode::kNotYetBegun;
}
@@ -193,4 +204,16 @@
return kSuccess;
}
+void MoqtOutgoingQueue::Close() {
+ if (closed_) {
+ QUICHE_BUG(MoqtOutgoingQueue_Close_twice)
+ << "Trying to close an outgoing queue that is already closed.";
+ return;
+ }
+ closed_ = true;
+
+ OpenNewGroup();
+ AddRawObject(MoqtObjectStatus::kEndOfTrack, {});
+}
+
} // namespace moqt
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.h b/quiche/quic/moqt/moqt_outgoing_queue.h
index 9715563..4687ac8 100644
--- a/quiche/quic/moqt/moqt_outgoing_queue.h
+++ b/quiche/quic/moqt/moqt_outgoing_queue.h
@@ -95,6 +95,9 @@
}
}
+ // Sends an "End of Track" object.
+ void Close();
+
private:
// The number of recent groups to keep around for newly joined subscribers.
static constexpr size_t kMaxQueuedGroups = 3;
@@ -125,7 +128,10 @@
using Group = std::vector<CachedObject>;
+ // Appends an object to the end of the current group.
void AddRawObject(MoqtObjectStatus status, quiche::QuicheMemSlice payload);
+ // Closes the current group, if there is any, and opens a new one.
+ void OpenNewGroup();
// The number of the oldest group available.
uint64_t first_group_in_queue() const {
@@ -137,6 +143,7 @@
MoqtForwardingPreference forwarding_preference_;
MoqtPriority publisher_priority_ = 128;
MoqtDeliveryOrder delivery_order_ = MoqtDeliveryOrder::kAscending;
+ bool closed_ = false;
absl::InlinedVector<Group, kMaxQueuedGroups> queue_;
uint64_t current_group_id_ = -1;
absl::flat_hash_set<MoqtObjectListener*> listeners_;
