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