diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index 2aba2b2..443dd50 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -12,10 +12,12 @@
 #include <utility>
 #include <vector>
 
+
 #include "absl/algorithm/container.h"
 #include "absl/status/status.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/string_view.h"
+#include "absl/types/span.h"
 #include "quiche/quic/core/quic_types.h"
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_subscribe_windows.h"
@@ -33,6 +35,27 @@
 
 using ::quic::Perspective;
 
+MoqtSession::Stream* MoqtSession::GetControlStream() {
+  if (!control_stream_.has_value()) {
+    return nullptr;
+  }
+  webtransport::Stream* raw_stream = session_->GetStreamById(*control_stream_);
+  if (raw_stream == nullptr) {
+    return nullptr;
+  }
+  return static_cast<Stream*>(raw_stream->visitor());
+}
+
+void MoqtSession::SendControlMessage(quiche::QuicheBuffer message) {
+  Stream* control_stream = GetControlStream();
+  if (control_stream == nullptr) {
+    QUICHE_LOG(DFATAL) << "Trying to send a message on the control stream "
+                          "while it does not exist";
+    return;
+  }
+  control_stream->SendOrBufferMessage(std::move(message));
+}
+
 void MoqtSession::OnSessionReady() {
   QUICHE_DLOG(INFO) << ENDPOINT << "Underlying session ready";
   if (parameters_.perspective == Perspective::IS_SERVER) {
@@ -55,12 +78,7 @@
   if (!parameters_.using_webtrans) {
     setup.path = parameters_.path;
   }
-  quiche::QuicheBuffer serialized_setup = framer_.SerializeClientSetup(setup);
-  bool success = control_stream->Write(serialized_setup.AsStringView());
-  if (!success) {
-    Error(MoqtError::kGenericError, "Failed to write client SETUP message");
-    return;
-  }
+  SendControlMessage(framer_.SerializeClientSetup(setup));
   QUIC_DLOG(INFO) << ENDPOINT << "Send the SETUP message";
 }
 
@@ -123,12 +141,7 @@
   }
   MoqtAnnounce message;
   message.track_namespace = track_namespace;
-  bool success = session_->GetStreamById(*control_stream_)
-                     ->Write(framer_.SerializeAnnounce(message).AsStringView());
-  if (!success) {
-    Error(MoqtError::kGenericError, "Failed to write ANNOUNCE message");
-    return;
-  }
+  SendControlMessage(framer_.SerializeAnnounce(message));
   QUIC_DLOG(INFO) << ENDPOINT << "Sent ANNOUNCE message for "
                   << message.track_namespace;
   pending_outgoing_announces_[track_namespace] = std::move(announce_callback);
@@ -235,13 +248,7 @@
   } else {
     message.track_alias = next_remote_track_alias_++;
   }
-  bool success =
-      session_->GetStreamById(*control_stream_)
-          ->Write(framer_.SerializeSubscribe(message).AsStringView());
-  if (!success) {
-    Error(MoqtError::kGenericError, "Failed to write SUBSCRIBE message");
-    return false;
-  }
+  SendControlMessage(framer_.SerializeSubscribe(message));
   QUIC_DLOG(INFO) << ENDPOINT << "Sent SUBSCRIBE message for "
                   << message.track_namespace << ":" << message.track_name;
   active_subscribes_.try_emplace(message.subscribe_id, message, visitor);
@@ -457,13 +464,7 @@
     MoqtServerSetup response;
     response.selected_version = session_->parameters_.version;
     response.role = MoqtRole::kBoth;
-    bool success = stream_->Write(
-        session_->framer_.SerializeServerSetup(response).AsStringView());
-    if (!success) {
-      session_->Error(MoqtError::kGenericError,
-                      "Failed to write server SETUP message");
-      return;
-    }
+    SendOrBufferMessage(session_->framer_.SerializeServerSetup(response));
     QUIC_DLOG(INFO) << ENDPOINT << "Sent the SETUP message";
   }
   // TODO: handle role and path.
@@ -506,13 +507,8 @@
   subscribe_error.error_code = error_code;
   subscribe_error.reason_phrase = reason_phrase;
   subscribe_error.track_alias = track_alias;
-  bool success =
-      stream_->Write(session_->framer_.SerializeSubscribeError(subscribe_error)
-                         .AsStringView());
-  if (!success) {
-    session_->Error(MoqtError::kGenericError,
-                    "Failed to write SUBSCRIBE_ERROR message");
-  }
+  SendOrBufferMessage(
+      session_->framer_.SerializeSubscribeError(subscribe_error));
 }
 
 void MoqtSession::Stream::OnSubscribeMessage(const MoqtSubscribe& message) {
@@ -573,13 +569,7 @@
   }
   MoqtSubscribeOk subscribe_ok;
   subscribe_ok.subscribe_id = message.subscribe_id;
-  bool success = stream_->Write(
-      session_->framer_.SerializeSubscribeOk(subscribe_ok).AsStringView());
-  if (!success) {
-    session_->Error(MoqtError::kGenericError,
-                    "Failed to write SUBSCRIBE_OK message");
-    return;
-  }
+  SendOrBufferMessage(session_->framer_.SerializeSubscribeOk(subscribe_ok));
   QUIC_DLOG(INFO) << ENDPOINT << "Created subscription for "
                   << message.track_namespace << ":" << message.track_name;
   if (!end.has_value()) {
@@ -660,13 +650,7 @@
   }
   MoqtAnnounceOk ok;
   ok.track_namespace = message.track_namespace;
-  bool success =
-      stream_->Write(session_->framer_.SerializeAnnounceOk(ok).AsStringView());
-  if (!success) {
-    session_->Error(MoqtError::kGenericError,
-                    "Failed to write ANNOUNCE_OK message");
-    return;
-  }
+  SendOrBufferMessage(session_->framer_.SerializeAnnounceOk(ok));
 }
 
 void MoqtSession::Stream::OnAnnounceOkMessage(const MoqtAnnounceOk& message) {
@@ -739,4 +723,17 @@
   return sequence;
 }
 
+void MoqtSession::Stream::SendOrBufferMessage(quiche::QuicheBuffer message,
+                                              bool fin) {
+  quiche::StreamWriteOptions options;
+  options.set_send_fin(fin);
+  options.set_buffer_unconditionally(true);
+  std::array<absl::string_view, 1> write_vector = {message.AsStringView()};
+  absl::Status success = stream_->Writev(absl::MakeSpan(write_vector), options);
+  if (!success.ok()) {
+    session_->Error(MoqtError::kGenericError,
+                    "Failed to write a control message");
+  }
+}
+
 }  // namespace moqt
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index 0bd0fa6..cb9163d 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -19,6 +19,7 @@
 #include "quiche/quic/moqt/moqt_parser.h"
 #include "quiche/quic/moqt/moqt_track.h"
 #include "quiche/common/platform/api/quiche_export.h"
+#include "quiche/common/quiche_buffer_allocator.h"
 #include "quiche/common/quiche_callbacks.h"
 #include "quiche/common/simple_buffer_allocator.h"
 #include "quiche/web_transport/web_transport.h"
@@ -120,8 +121,7 @@
   bool PublishObject(const FullTrackName& full_track_name, uint64_t group_id,
                      uint64_t object_id, uint64_t object_send_order,
                      MoqtForwardingPreference forwarding_preference,
-                     absl::string_view payload,
-                     bool end_of_stream);
+                     absl::string_view payload, bool end_of_stream);
   // TODO: Add an API to FIN the stream for a particular track/group/object.
   // TODO: Add an API to send partial objects.
 
@@ -174,6 +174,10 @@
 
     webtransport::Stream* stream() const { return stream_; }
 
+    // Sends a control message, or buffers it if there is insufficient flow
+    // control credit.
+    void SendOrBufferMessage(quiche::QuicheBuffer message, bool fin = false);
+
    private:
     friend class test::MoqtSessionPeer;
     void SendSubscribeError(const MoqtSubscribe& message,
@@ -191,6 +195,12 @@
     std::string partial_object_;
   };
 
+  // Returns the pointer to the control stream, or nullptr if none is present.
+  Stream* GetControlStream();
+  // Sends a message on the control stream; QUICHE_DCHECKs if no control stream
+  // is present.
+  void SendControlMessage(quiche::QuicheBuffer message);
+
   // Returns false if the SUBSCRIBE isn't sent.
   bool Subscribe(MoqtSubscribe& message, RemoteTrack::Visitor* visitor);
   // converts two MoqtLocations into absolute sequences.
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index cc24b9e..9401a24 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -34,6 +34,7 @@
 namespace {
 
 using ::testing::_;
+using ::testing::AnyNumber;
 using ::testing::Return;
 using ::testing::StrictMock;
 
@@ -65,10 +66,13 @@
 class MoqtSessionPeer {
  public:
   static std::unique_ptr<MoqtParserVisitor> CreateControlStream(
-      MoqtSession* session, webtransport::Stream* stream) {
+      MoqtSession* session, webtransport::test::MockStream* stream) {
     auto new_stream = std::make_unique<MoqtSession::Stream>(
         session, stream, /*is_control_stream=*/true);
     session->control_stream_ = kControlStreamId;
+    EXPECT_CALL(*stream, visitor())
+        .Times(AnyNumber())
+        .WillRepeatedly(Return(new_stream.get()));
     return new_stream;
   }
 
@@ -155,7 +159,9 @@
       });
   EXPECT_CALL(mock_stream, GetStreamId())
       .WillOnce(Return(webtransport::StreamId(4)));
+  EXPECT_CALL(mock_session_, GetStreamById(4)).WillOnce(Return(&mock_stream));
   bool correct_message = false;
+  EXPECT_CALL(mock_stream, visitor()).WillOnce([&] { return visitor.get(); });
   EXPECT_CALL(mock_stream, Writev(_, _))
       .WillOnce([&](absl::Span<const absl::string_view> data,
                     const quiche::StreamWriteOptions& options) {
@@ -789,7 +795,9 @@
       });
   EXPECT_CALL(mock_stream, GetStreamId())
       .WillOnce(Return(webtransport::StreamId(4)));
+  EXPECT_CALL(mock_session_, GetStreamById(4)).WillOnce(Return(&mock_stream));
   bool correct_message = false;
+  EXPECT_CALL(mock_stream, visitor()).WillOnce([&] { return visitor.get(); });
   EXPECT_CALL(mock_stream, Writev(_, _))
       .WillOnce([&](absl::Span<const absl::string_view> data,
                     const quiche::StreamWriteOptions& options) {