Factor out the logic for queueing outgoing control messages from MoqtBidiStreamBase into a separate MoqtControlMessageQueue class. PiperOrigin-RevId: 927417268
diff --git a/build/source_list.bzl b/build/source_list.bzl index 5d80b7b..3a37cb3 100644 --- a/build/source_list.bzl +++ b/build/source_list.bzl
@@ -1581,6 +1581,7 @@ moqt_hdrs = [ "quic/moqt/moqt_bidi_stream.h", "quic/moqt/moqt_bitrate_adjuster.h", + "quic/moqt/moqt_control_message_queue.h", "quic/moqt/moqt_error.h", "quic/moqt/moqt_fetch_task.h", "quic/moqt/moqt_framer.h", @@ -1619,6 +1620,7 @@ moqt_srcs = [ "quic/moqt/moqt_bidi_stream.cc", "quic/moqt/moqt_bitrate_adjuster.cc", + "quic/moqt/moqt_control_message_queue.cc", "quic/moqt/moqt_error.cc", "quic/moqt/moqt_framer.cc", "quic/moqt/moqt_key_value_pair.cc", @@ -1653,6 +1655,7 @@ moqt_test_srcs = [ "quic/moqt/moqt_bidi_stream_test.cc", "quic/moqt/moqt_bitrate_adjuster_test.cc", + "quic/moqt/moqt_control_message_queue_test.cc", "quic/moqt/moqt_framer_test.cc", "quic/moqt/moqt_integration_test.cc", "quic/moqt/moqt_key_value_pair_test.cc",
diff --git a/build/source_list.gni b/build/source_list.gni index 962bbfd..e2457fe 100644 --- a/build/source_list.gni +++ b/build/source_list.gni
@@ -1585,6 +1585,7 @@ moqt_hdrs = [ "src/quiche/quic/moqt/moqt_bidi_stream.h", "src/quiche/quic/moqt/moqt_bitrate_adjuster.h", + "src/quiche/quic/moqt/moqt_control_message_queue.h", "src/quiche/quic/moqt/moqt_error.h", "src/quiche/quic/moqt/moqt_fetch_task.h", "src/quiche/quic/moqt/moqt_framer.h", @@ -1623,6 +1624,7 @@ moqt_srcs = [ "src/quiche/quic/moqt/moqt_bidi_stream.cc", "src/quiche/quic/moqt/moqt_bitrate_adjuster.cc", + "src/quiche/quic/moqt/moqt_control_message_queue.cc", "src/quiche/quic/moqt/moqt_error.cc", "src/quiche/quic/moqt/moqt_framer.cc", "src/quiche/quic/moqt/moqt_key_value_pair.cc", @@ -1658,6 +1660,7 @@ moqt_test_srcs = [ "src/quiche/quic/moqt/moqt_bidi_stream_test.cc", "src/quiche/quic/moqt/moqt_bitrate_adjuster_test.cc", + "src/quiche/quic/moqt/moqt_control_message_queue_test.cc", "src/quiche/quic/moqt/moqt_framer_test.cc", "src/quiche/quic/moqt/moqt_integration_test.cc", "src/quiche/quic/moqt/moqt_key_value_pair_test.cc",
diff --git a/build/source_list.json b/build/source_list.json index 75b9c7f..6cf4e9a 100644 --- a/build/source_list.json +++ b/build/source_list.json
@@ -1584,6 +1584,7 @@ "moqt_hdrs": [ "quiche/quic/moqt/moqt_bidi_stream.h", "quiche/quic/moqt/moqt_bitrate_adjuster.h", + "quiche/quic/moqt/moqt_control_message_queue.h", "quiche/quic/moqt/moqt_error.h", "quiche/quic/moqt/moqt_fetch_task.h", "quiche/quic/moqt/moqt_framer.h", @@ -1622,6 +1623,7 @@ "moqt_srcs": [ "quiche/quic/moqt/moqt_bidi_stream.cc", "quiche/quic/moqt/moqt_bitrate_adjuster.cc", + "quiche/quic/moqt/moqt_control_message_queue.cc", "quiche/quic/moqt/moqt_error.cc", "quiche/quic/moqt/moqt_framer.cc", "quiche/quic/moqt/moqt_key_value_pair.cc", @@ -1657,6 +1659,7 @@ "moqt_test_srcs": [ "quiche/quic/moqt/moqt_bidi_stream_test.cc", "quiche/quic/moqt/moqt_bitrate_adjuster_test.cc", + "quiche/quic/moqt/moqt_control_message_queue_test.cc", "quiche/quic/moqt/moqt_framer_test.cc", "quiche/quic/moqt/moqt_integration_test.cc", "quiche/quic/moqt/moqt_key_value_pair_test.cc",
diff --git a/quiche/quic/moqt/moqt_bidi_stream.cc b/quiche/quic/moqt/moqt_bidi_stream.cc index 3d6b9c6..46a4d78 100644 --- a/quiche/quic/moqt/moqt_bidi_stream.cc +++ b/quiche/quic/moqt/moqt_bidi_stream.cc
@@ -4,7 +4,6 @@ #include "quiche/quic/moqt/moqt_bidi_stream.h" -#include <array> #include <cstdint> #include <optional> #include <utility> @@ -12,7 +11,6 @@ #include "absl/status/status.h" #include "absl/status/statusor.h" #include "absl/strings/string_view.h" -#include "absl/types/span.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" @@ -20,10 +18,6 @@ #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/common/platform/api/quiche_bug_tracker.h" #include "quiche/common/platform/api/quiche_logging.h" -#include "quiche/common/quiche_buffer_allocator.h" -#include "quiche/common/quiche_mem_slice.h" -#include "quiche/web_transport/stream_helpers.h" -#include "quiche/web_transport/web_transport.h" namespace moqt { @@ -57,37 +51,10 @@ << "OnCanWrite() called when no stream is bound"; return; } - webtransport::Stream* stream = stream_parser_->stream(); - if (pending_messages_.empty() && fin_queued_) { - absl::Status status = webtransport::SendFinOnStream(*stream); - if (!status.ok()) { - OnFatalError(status); - } - return; + absl::Status status = outgoing_message_queue_.OnCanWrite(); + if (!status.ok()) { + OnFatalError(status); } - while (!pending_messages_.empty() && stream->CanWrite()) { - absl::Status status = - SendMessage(std::move(pending_messages_.front()), - fin_queued_ && pending_messages_.size() == 1); - pending_messages_.pop_front(); - if (!status.ok()) { - OnFatalError(status); - return; - } - } -} - -absl::Status MoqtBidiStreamBase::SendOrBufferMessage( - quiche::QuicheBuffer message, bool fin) { - if (fin_queued_) { - return absl::InternalError( - "Trying to send data when a FIN has been already queued"); - } - if (stream() == nullptr || !stream()->CanWrite()) { - fin_queued_ = fin; - return AddToQueue(std::move(message)); - } - return SendMessage(std::move(message), fin); } absl::Status MoqtBidiStreamBase::SendRequestOk( @@ -129,21 +96,4 @@ std::move(session_error_callback_)(*error_code, status.message()); } -absl::Status MoqtBidiStreamBase::AddToQueue(quiche::QuicheBuffer message) { - if (pending_messages_.size() == kMaxPendingMessages) { - return absl::ResourceExhaustedError( - "Not enough flow credit on the control stream"); - } - pending_messages_.push_back(std::move(message)); - return absl::OkStatus(); -} - -absl::Status MoqtBidiStreamBase::SendMessage(quiche::QuicheBuffer message, - bool fin) { - webtransport::StreamWriteOptions options; - options.set_send_fin(fin); - std::array write_vector = {quiche::QuicheMemSlice(std::move(message))}; - return stream()->Writev(absl::MakeSpan(write_vector), options); -} - } // namespace moqt
diff --git a/quiche/quic/moqt/moqt_bidi_stream.h b/quiche/quic/moqt/moqt_bidi_stream.h index 29f98f4..1dd88aa 100644 --- a/quiche/quic/moqt/moqt_bidi_stream.h +++ b/quiche/quic/moqt/moqt_bidi_stream.h
@@ -5,7 +5,6 @@ #ifndef QUICHE_QUIC_MOQT_MOQT_BIDI_STREAM_H #define QUICHE_QUIC_MOQT_MOQT_BIDI_STREAM_H -#include <cstddef> #include <cstdint> #include <memory> #include <optional> @@ -18,6 +17,7 @@ #include "absl/strings/str_cat.h" #include "absl/strings/string_view.h" #include "quiche/quic/core/quic_time.h" +#include "quiche/quic/moqt/moqt_control_message_queue.h" #include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_framer.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" @@ -26,7 +26,6 @@ #include "quiche/common/platform/api/quiche_logging.h" #include "quiche/common/quiche_buffer_allocator.h" #include "quiche/common/quiche_callbacks.h" -#include "quiche/common/quiche_circular_deque.h" #include "quiche/web_transport/web_transport.h" namespace moqt { @@ -47,9 +46,6 @@ // as it might not yet exist due to flow control limits. class MoqtBidiStreamBase : public webtransport::StreamVisitor { public: - // Maximum amount of messages buffered on top of the QUIC send buffer. - static constexpr size_t kMaxPendingMessages = 100; - MoqtBidiStreamBase(MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, BidiStreamDeletedCallback stream_deleted_callback, @@ -65,12 +61,14 @@ std::unique_ptr<MoqtControlStreamParser> absl_nonnull parser) { QUICHE_DCHECK(stream_parser_ == nullptr); stream_parser_ = std::move(parser); + outgoing_message_queue_.SetStream(stream_parser_->stream()); OnStreamBound(); } // Binds a WebTransport stream `stream` to this object. void BindStream(webtransport::Stream* absl_nonnull stream) { QUICHE_DCHECK(stream_parser_ == nullptr); stream_parser_ = std::make_unique<MoqtControlStreamParser>(stream); + outgoing_message_queue_.SetStream(stream); OnStreamBound(); } @@ -81,12 +79,12 @@ void OnCanRead() override; void OnCanWrite() override; - bool QueueIsFull() const { - return pending_messages_.size() == kMaxPendingMessages; - } + bool QueueIsFull() const { return outgoing_message_queue_.QueueIsFull(); } absl::Status SendOrBufferMessage(quiche::QuicheBuffer message, - bool fin = false); + bool fin = false) { + return outgoing_message_queue_.SendOrBufferMessage(std::move(message), fin); + } void SendOrBufferMessageOrFatal(quiche::QuicheBuffer message, bool fin = false) { CheckStatus(SendOrBufferMessage(std::move(message), fin)); @@ -102,10 +100,7 @@ absl::Status SendRequestError(uint64_t request_id, MoqtRequestErrorInfo info, bool fin = false); - void Fin() { - fin_queued_ = true; - OnCanWrite(); - } + void Fin() { CheckStatus(outgoing_message_queue_.Fin()); } void Reset(webtransport::StreamErrorCode error) { webtransport::Stream* stream = stream_parser_->stream(); if (stream != nullptr) { @@ -164,9 +159,6 @@ private: friend class test::MoqtBidiStreamTestWrapper; - absl::Status AddToQueue(quiche::QuicheBuffer message); - absl::Status SendMessage(quiche::QuicheBuffer message, bool fin); - // CanDispatch<S, M> indicates whether `S` has a method with signature // absl::Status OnControlMessage(const M&); template <typename Subclass, typename Message, typename = void> @@ -181,8 +173,7 @@ MoqtFramer* absl_nonnull framer_; std::unique_ptr<MoqtControlStreamParser> absl_nullable stream_parser_; MoqtControlMessageParser message_parser_; - quiche::QuicheCircularDeque<quiche::QuicheBuffer> pending_messages_; - bool fin_queued_ = false; + MoqtControlMessageQueue outgoing_message_queue_; BidiStreamDeletedCallback stream_deleted_callback_; SessionErrorCallback session_error_callback_; };
diff --git a/quiche/quic/moqt/moqt_bidi_stream_test.cc b/quiche/quic/moqt/moqt_bidi_stream_test.cc index b2dc6db..bb85966 100644 --- a/quiche/quic/moqt/moqt_bidi_stream_test.cc +++ b/quiche/quic/moqt/moqt_bidi_stream_test.cc
@@ -5,28 +5,18 @@ #include "quiche/quic/moqt/moqt_bidi_stream.h" #include <memory> -#include <optional> #include "absl/status/status.h" #include "absl/strings/string_view.h" -#include "absl/types/span.h" #include "quiche/quic/core/quic_types.h" #include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_framer.h" -#include "quiche/quic/moqt/moqt_key_value_pair.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_interface.h" -#include "quiche/quic/moqt/test_tools/moqt_framer_utils.h" #include "quiche/common/platform/api/quiche_test.h" -#include "quiche/common/quiche_mem_slice.h" -#include "quiche/common/test_tools/quiche_test_utils.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" - -using ::testing::_; -using ::testing::Return; namespace moqt::test { @@ -70,44 +60,6 @@ webtransport::test::MockStream mock_stream_; }; -TEST_F(MoqtBidiStreamTest, MessageBufferedThenSent) { - stream_->BindStream(&mock_stream_); - EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(false)); - EXPECT_CALL(mock_stream_, Writev).Times(0); - QUICHE_EXPECT_OK(stream_->SendRequestOk(0, MessageParameters())); - QUICHE_EXPECT_OK(stream_->SendRequestError(2, RequestErrorCode::kUnauthorized, - std::nullopt, - - "bad request")); - stream_->Fin(); - { - testing::InSequence seq; - EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); - EXPECT_CALL(mock_stream_, - Writev(ControlMessageOfType(MoqtMessageType::kRequestOk), _)); - EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); - EXPECT_CALL(mock_stream_, - Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _)) - .WillOnce([](absl::Span<quiche::QuicheMemSlice>, - const webtransport::StreamWriteOptions& options) { - EXPECT_TRUE(options.send_fin()); - return absl::OkStatus(); - }); - } - stream_->OnCanWrite(); -} - -TEST_F(MoqtBidiStreamTest, FinSentWhenDrained) { - stream_->BindStream(&mock_stream_); - EXPECT_CALL(mock_stream_, Writev) - .WillOnce([](absl::Span<quiche::QuicheMemSlice>, - const webtransport::StreamWriteOptions& options) { - EXPECT_TRUE(options.send_fin()); - return absl::OkStatus(); - }); - stream_->Fin(); -} - TEST_F(MoqtBidiStreamTest, Reset) { stream_->BindStream(&mock_stream_); EXPECT_CALL(mock_stream_, ResetWithUserCode(1234)); @@ -119,22 +71,6 @@ stream_.reset(); } -TEST_F(MoqtBidiStreamTest, PendingQueueFull) { - stream_->BindStream(&mock_stream_); - EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(false)); - for (int i = 0; i < MoqtBidiStreamBase::kMaxPendingMessages; ++i) { - EXPECT_FALSE(stream_->QueueIsFull()); - QUICHE_EXPECT_OK(stream_->SendOrBufferMessage( - framer_.SerializeRequestUpdate(MoqtRequestUpdate{}))); - } - EXPECT_TRUE(stream_->QueueIsFull()); - EXPECT_EQ(stream_ - ->SendOrBufferMessage( - framer_.SerializeRequestUpdate(MoqtRequestUpdate{})) - .code(), - absl::StatusCode::kResourceExhausted); -} - TEST_F(MoqtBidiStreamTest, DispatchControlMessage) { webtransport::test::InMemoryStream stream(0); stream_->BindStream(&stream);
diff --git a/quiche/quic/moqt/moqt_control_message_queue.cc b/quiche/quic/moqt/moqt_control_message_queue.cc new file mode 100644 index 0000000..b2b984f --- /dev/null +++ b/quiche/quic/moqt/moqt_control_message_queue.cc
@@ -0,0 +1,77 @@ +// Copyright 2026 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/moqt/moqt_control_message_queue.h" + +#include <array> +#include <utility> + +#include "absl/status/status.h" +#include "absl/types/span.h" +#include "quiche/common/quiche_buffer_allocator.h" +#include "quiche/common/quiche_mem_slice.h" +#include "quiche/web_transport/stream_helpers.h" +#include "quiche/web_transport/web_transport.h" + +namespace moqt { + +absl::Status MoqtControlMessageQueue::OnCanWrite() { + if (stream_ == nullptr) { + return absl::InternalError("OnCanWrite() called when no stream is bound"); + } + if (pending_messages_.empty() && fin_queued_) { + return webtransport::SendFinOnStream(*stream_); + } + while (!pending_messages_.empty() && stream_->CanWrite()) { + absl::Status status = + SendMessage(*stream_, std::move(pending_messages_.front()), + fin_queued_ && pending_messages_.size() == 1); + pending_messages_.pop_front(); + if (!status.ok()) { + return status; + } + } + return absl::OkStatus(); +} + +absl::Status MoqtControlMessageQueue::SendOrBufferMessage( + quiche::QuicheBuffer message, bool fin) { + if (fin_queued_) { + return absl::InternalError( + "Trying to send data when a FIN has been already queued"); + } + if (stream_ == nullptr || !stream_->CanWrite()) { + fin_queued_ = fin; + return AddToQueue(std::move(message)); + } + return SendMessage(*stream_, std::move(message), fin); +} + +absl::Status MoqtControlMessageQueue::Fin() { + fin_queued_ = true; + if (stream_ != nullptr) { + return OnCanWrite(); + } + return absl::OkStatus(); +} + +absl::Status MoqtControlMessageQueue::AddToQueue(quiche::QuicheBuffer message) { + if (pending_messages_.size() == kMaxPendingMessages) { + return absl::ResourceExhaustedError( + "Not enough flow credit on the control stream"); + } + pending_messages_.push_back(std::move(message)); + return absl::OkStatus(); +} + +absl::Status MoqtControlMessageQueue::SendMessage(webtransport::Stream& stream, + quiche::QuicheBuffer message, + bool fin) { + webtransport::StreamWriteOptions options; + options.set_send_fin(fin); + std::array write_vector = {quiche::QuicheMemSlice(std::move(message))}; + return stream.Writev(absl::MakeSpan(write_vector), options); +} + +} // namespace moqt
diff --git a/quiche/quic/moqt/moqt_control_message_queue.h b/quiche/quic/moqt/moqt_control_message_queue.h new file mode 100644 index 0000000..d82be23 --- /dev/null +++ b/quiche/quic/moqt/moqt_control_message_queue.h
@@ -0,0 +1,58 @@ +// Copyright 2026 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#ifndef QUICHE_QUIC_MOQT_MOQT_CONTROL_MESSAGE_QUEUE_H_ +#define QUICHE_QUIC_MOQT_MOQT_CONTROL_MESSAGE_QUEUE_H_ + +#include <cstddef> + +#include "absl/base/nullability.h" +#include "absl/status/status.h" +#include "quiche/common/platform/api/quiche_export.h" +#include "quiche/common/quiche_buffer_allocator.h" +#include "quiche/common/quiche_circular_deque.h" +#include "quiche/web_transport/web_transport.h" + +namespace moqt { + +// MoqtControlMessageQueue manages the queueing and sending of outgoing control +// messages on a WebTransport stream. +class QUICHE_EXPORT MoqtControlMessageQueue { + public: + // Maximum amount of messages buffered on top of the QUIC send buffer. + static constexpr size_t kMaxPendingMessages = 100; + + MoqtControlMessageQueue() = default; + explicit MoqtControlMessageQueue(webtransport::Stream* absl_nullable stream) + : stream_(stream) {} + + webtransport::Stream* absl_nullable stream() const { return stream_; } + void SetStream(webtransport::Stream* absl_nonnull stream) { + stream_ = stream; + } + + bool QueueIsFull() const { + return pending_messages_.size() == kMaxPendingMessages; + } + + absl::Status SendOrBufferMessage(quiche::QuicheBuffer message, + bool fin = false); + absl::Status Fin(); + + // Dequeues all pending writes. + absl::Status OnCanWrite(); + + private: + absl::Status AddToQueue(quiche::QuicheBuffer message); + static absl::Status SendMessage(webtransport::Stream& stream, + quiche::QuicheBuffer message, bool fin); + + webtransport::Stream* absl_nullable stream_ = nullptr; + quiche::QuicheCircularDeque<quiche::QuicheBuffer> pending_messages_; + bool fin_queued_ = false; +}; + +} // namespace moqt + +#endif // QUICHE_QUIC_MOQT_MOQT_CONTROL_MESSAGE_QUEUE_H_
diff --git a/quiche/quic/moqt/moqt_control_message_queue_test.cc b/quiche/quic/moqt/moqt_control_message_queue_test.cc new file mode 100644 index 0000000..2b32b2e --- /dev/null +++ b/quiche/quic/moqt/moqt_control_message_queue_test.cc
@@ -0,0 +1,130 @@ +// Copyright 2026 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/moqt/moqt_control_message_queue.h" + +#include "absl/status/status.h" +#include "absl/strings/string_view.h" +#include "absl/types/span.h" +#include "quiche/common/platform/api/quiche_test.h" +#include "quiche/common/quiche_buffer_allocator.h" +#include "quiche/common/quiche_mem_slice.h" +#include "quiche/common/simple_buffer_allocator.h" +#include "quiche/common/test_tools/quiche_test_utils.h" +#include "quiche/web_transport/test_tools/mock_web_transport.h" +#include "quiche/web_transport/web_transport.h" + +using ::testing::Return; + +namespace moqt::test { +namespace { + +class MoqtControlMessageQueueTest : public quiche::test::QuicheTest { + public: + MoqtControlMessageQueueTest() : queue_(&mock_stream_) {} + + quiche::QuicheBuffer MakeMessage(absl::string_view payload) { + return quiche::QuicheBuffer::Copy(quiche::SimpleBufferAllocator::Get(), + payload); + } + + webtransport::test::MockStream mock_stream_; + MoqtControlMessageQueue queue_; +}; + +TEST_F(MoqtControlMessageQueueTest, MessageBufferedThenSent) { + EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(false)); + EXPECT_CALL(mock_stream_, Writev).Times(0); + QUICHE_EXPECT_OK(queue_.SendOrBufferMessage(MakeMessage("message1"))); + QUICHE_EXPECT_OK(queue_.SendOrBufferMessage(MakeMessage("message2"))); + QUICHE_EXPECT_OK(queue_.Fin()); + { + testing::InSequence seq; + EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> slices, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(slices.size(), 1u); + EXPECT_EQ(slices[0].AsStringView(), "message1"); + EXPECT_FALSE(options.send_fin()); + return absl::OkStatus(); + }); + EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> slices, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(slices.size(), 1u); + EXPECT_EQ(slices[0].AsStringView(), "message2"); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + } + QUICHE_EXPECT_OK(queue_.OnCanWrite()); +} + +TEST_F(MoqtControlMessageQueueTest, FinSentWhenEmpty) { + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> slices, + const webtransport::StreamWriteOptions& options) { + EXPECT_TRUE(slices.empty()); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + QUICHE_EXPECT_OK(queue_.Fin()); +} + +TEST_F(MoqtControlMessageQueueTest, PendingQueueFull) { + EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(false)); + for (int i = 0; i < MoqtControlMessageQueue::kMaxPendingMessages; ++i) { + EXPECT_FALSE(queue_.QueueIsFull()); + QUICHE_EXPECT_OK(queue_.SendOrBufferMessage(MakeMessage("msg"))); + } + EXPECT_TRUE(queue_.QueueIsFull()); + EXPECT_EQ(queue_.SendOrBufferMessage(MakeMessage("msg")).code(), + absl::StatusCode::kResourceExhausted); +} + +TEST_F(MoqtControlMessageQueueTest, SendWhenCanWrite) { + EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> slices, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(slices.size(), 1u); + EXPECT_EQ(slices[0].AsStringView(), "immediate"); + EXPECT_FALSE(options.send_fin()); + return absl::OkStatus(); + }); + QUICHE_EXPECT_OK( + queue_.SendOrBufferMessage(MakeMessage("immediate"), /*fin=*/false)); +} + +TEST_F(MoqtControlMessageQueueTest, ErrorWhenFinAlreadyQueued) { + EXPECT_CALL(mock_stream_, Writev).WillOnce(Return(absl::OkStatus())); + QUICHE_EXPECT_OK(queue_.Fin()); + EXPECT_EQ(queue_.SendOrBufferMessage(MakeMessage("msg")).code(), + absl::StatusCode::kInternal); +} + +TEST_F(MoqtControlMessageQueueTest, OnCanWriteWithoutStream) { + MoqtControlMessageQueue queue(nullptr); + EXPECT_EQ(queue.OnCanWrite().code(), absl::StatusCode::kInternal); +} + +TEST_F(MoqtControlMessageQueueTest, BufferFinWithoutStream) { + MoqtControlMessageQueue queue(nullptr); + QUICHE_EXPECT_OK(queue.Fin()); + queue.SetStream(&mock_stream_); + + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> slices, + const webtransport::StreamWriteOptions& options) { + EXPECT_TRUE(slices.empty()); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + QUICHE_EXPECT_OK(queue.OnCanWrite()); +} + +} // namespace +} // namespace moqt::test