diff --git a/quiche/quic/moqt/moqt_bidi_stream.cc b/quiche/quic/moqt/moqt_bidi_stream.cc
index 9509c4e..b5b7740 100644
--- a/quiche/quic/moqt/moqt_bidi_stream.cc
+++ b/quiche/quic/moqt/moqt_bidi_stream.cc
@@ -23,7 +23,7 @@
 namespace moqt {
 
 void MoqtBidiStreamBase::OnCanRead() {
-  if (stream_parser_ == nullptr) {
+  if (!stream_parser_.has_value()) {
     QUICHE_BUG(MoqtBidiStreamBase_OnCanRead_no_stream)
         << "OnCanRead() called when no stream is bound";
     return;
@@ -47,7 +47,7 @@
 }
 
 void MoqtBidiStreamBase::OnCanWrite() {
-  if (stream_parser_ == nullptr) {
+  if (!stream_parser_.has_value()) {
     QUICHE_BUG(MoqtBidiStreamBase_OnCanWrite_no_stream)
         << "OnCanWrite() called when no stream is bound";
     return;
diff --git a/quiche/quic/moqt/moqt_bidi_stream.h b/quiche/quic/moqt/moqt_bidi_stream.h
index 68b55f7..13fc2d2 100644
--- a/quiche/quic/moqt/moqt_bidi_stream.h
+++ b/quiche/quic/moqt/moqt_bidi_stream.h
@@ -54,17 +54,16 @@
   ~MoqtBidiStreamBase() = default;
 
   // Binds a WebTransport stream associated with `parser` to this object.
-  void BindStream(
-      std::unique_ptr<MoqtControlStreamParser> absl_nonnull parser) {
-    QUICHE_DCHECK(stream_parser_ == nullptr);
-    stream_parser_ = std::move(parser);
+  void BindStream(MoqtStreamTypeParser parser) {
+    QUICHE_DCHECK(!stream_parser_.has_value());
+    stream_parser_.emplace(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);
+    QUICHE_DCHECK(!stream_parser_.has_value());
+    stream_parser_.emplace(stream);
     outgoing_message_queue_.SetStream(stream);
     OnStreamBound();
   }
@@ -115,9 +114,8 @@
     Detach();
   }
   void Reset(webtransport::StreamErrorCode error) {
-    webtransport::Stream* stream = stream_parser_->stream();
-    if (stream != nullptr) {
-      stream->ResetWithUserCode(error);
+    if (stream() != nullptr) {
+      stream()->ResetWithUserCode(error);
     }
     Detach();
   }
@@ -154,19 +152,21 @@
   // Terminates the MoQT session due to a fatal error encountered.
   void OnFatalError(absl::Status status);
 
-  MoqtControlStreamParser* stream_parser() { return stream_parser_.get(); }
+  MoqtControlStreamParser* stream_parser() {
+    return stream_parser_.has_value() ? &*stream_parser_ : nullptr;
+  }
   const MoqtControlMessageParser& message_parser() const {
     return message_parser_;
   }
   webtransport::Stream* stream() const {
-    return stream_parser_ != nullptr ? stream_parser_->stream() : nullptr;
+    return stream_parser_.has_value() ? stream_parser_->stream() : nullptr;
   }
 
  private:
   friend class test::MoqtBidiStreamTestWrapper;
 
   MoqtFramer* absl_nonnull framer_;
-  std::unique_ptr<MoqtControlStreamParser> absl_nullable stream_parser_;
+  std::optional<MoqtControlStreamParser> stream_parser_;
   MoqtControlMessageParser message_parser_;
   MoqtControlMessageQueue outgoing_message_queue_;
   MoqtRequestUpdateQueue request_update_queue_;
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc
index 3eec6fe..c7aa980 100644
--- a/quiche/quic/moqt/moqt_parser.cc
+++ b/quiche/quic/moqt/moqt_parser.cc
@@ -488,6 +488,54 @@
   return status;
 }
 
+MoqtStreamTypeParser::MoqtStreamTypeParser(
+    MoqtStreamTypeParser&& other) noexcept
+    : stream_(other.stream_), type_(other.type_), status_(other.status_) {
+  other.status_ = absl::InternalError("Accessing a moved-from parser");
+}
+
+MoqtStreamTypeParser& MoqtStreamTypeParser::operator=(
+    MoqtStreamTypeParser&& other) noexcept {
+  if (this != &other) {
+    stream_ = other.stream_;
+    type_ = other.type_;
+    status_ = other.status_;
+    other.status_ = absl::InternalError("Accessing a moved-from parser");
+  }
+  return *this;
+}
+
+absl::StatusOr<uint64_t> MoqtStreamTypeParser::ReadStreamType() {
+  if (!status_.ok()) {
+    return status_;
+  }
+  if (type_.has_value()) {
+    // The type has already been read; the rest of the stream should be only
+    // read by the next parser.
+    return *type_;
+  }
+  bool fin_read = false;
+  std::optional<uint64_t> type = ReadMoqVarIntFromStream(*stream_, fin_read);
+  if (fin_read && type != MoqtDataStreamType::kPadding) {
+    // Besides padding streams, all other streams require some data after the
+    // type byte.
+    status_ = absl::InvalidArgumentError(
+        "FIN received before or immediately after the stream type");
+    return status_;
+  }
+  if (!type.has_value()) {
+    return absl::UnavailableError("No complete message available");
+  }
+  type_ = *type;
+  return *type_;
+}
+
+MoqtControlStreamParser::MoqtControlStreamParser(
+    MoqtStreamTypeParser type_parser)
+    : stream_(*type_parser.stream()),
+      current_message_type_(type_parser.stream_type()),
+      fin_read_(false) {}
+
 absl::StatusOr<MoqtRawControlMessage>
 MoqtControlStreamParser::ReadNextMessage() {
   if (error_encountered_ || fin_read_) {
@@ -509,24 +557,6 @@
   return result;
 }
 
-absl::StatusOr<MoqtMessageType>
-MoqtControlStreamParser::ReadFirstMessageType() {
-  if (first_message_type_.has_value()) {
-    return static_cast<MoqtMessageType>(*first_message_type_);
-  }
-  if (error_encountered_ || fin_read_) {
-    return absl::FailedPreconditionError(
-        "Trying to read from a control stream after an error or an EOF "
-        "occurred.");
-  }
-  absl::Status read_status = ReadMessageType();
-  if (absl::IsUnavailable(read_status) && fin_read_) {
-    return absl::InvalidArgumentError("FIN received before any type");
-  }
-  QUICHE_RETURN_IF_ERROR(read_status);
-  return static_cast<MoqtMessageType>(*first_message_type_);
-}
-
 absl::Status MoqtControlStreamParser::ReadMessageType() {
   if (current_message_type_.has_value()) {
     QUICHE_BUG(MoqtControlStreamParser_ReadMessageType_bad_state)
@@ -549,9 +579,6 @@
         "Unexpected FIN on a control stream (FIN received immediately after "
         "type)");
   }
-  if (!first_message_type_.has_value()) {
-    first_message_type_ = *current_message_type_;
-  }
   return absl::OkStatus();
 }
 
@@ -1068,6 +1095,16 @@
   return absl::OkStatus();
 }
 
+MoqtDataParser::MoqtDataParser(MoqtStreamTypeParser type_parser,
+                               MoqtDataParserVisitor* visitor)
+    : stream_(*type_parser.stream()), visitor_(*visitor) {
+  if (type_parser.stream_type().has_value()) {
+    ProcessStreamType(*type_parser.stream_type());
+  } else {
+    next_input_ = kStreamType;
+  }
+}
+
 void MoqtDataParser::ParseError(absl::string_view reason) {
   if (parsing_error_) {
     return;  // Don't send multiple parse errors.
@@ -1312,25 +1349,13 @@
   }
   switch (next_input_) {
     case kStreamType: {
+      // TODO(vasilvv): Handle padding streams (which are allowed to FIN
+      // immediately after type, unlike all other types of streams).
       std::optional<uint64_t> value_read = ReadMoqVarIntNoFin();
       if (!value_read.has_value()) {
         return;
       }
-      std::optional<MoqtDataStreamType> type =
-          MoqtDataStreamType::FromValue(*value_read);
-      if (!type.has_value()) {
-        ParseError("Invalid stream type supplied");
-        return;
-      }
-      type_ = *type;
-      if (type_.IsPadding()) {
-        next_input_ = kPadding;
-        return;
-      }
-      if (type_.EndOfGroupInStream()) {
-        contains_end_of_group_ = true;
-      }
-      next_input_ = AdvanceParserState();
+      ProcessStreamType(*value_read);
       return;
     }
 
@@ -1604,4 +1629,22 @@
   return stream_.SkipBytes(0);
 }
 
+void MoqtDataParser::ProcessStreamType(uint64_t raw_type) {
+  std::optional<MoqtDataStreamType> type =
+      MoqtDataStreamType::FromValue(raw_type);
+  if (!type.has_value()) {
+    ParseError("Invalid stream type supplied");
+    return;
+  }
+  type_ = *type;
+  if (type_.IsPadding()) {
+    next_input_ = kPadding;
+    return;
+  }
+  if (type_.EndOfGroupInStream()) {
+    contains_end_of_group_ = true;
+  }
+  next_input_ = AdvanceParserState();
+}
+
 }  // namespace moqt
diff --git a/quiche/quic/moqt/moqt_parser.h b/quiche/quic/moqt/moqt_parser.h
index f8f685b..d28e203 100644
--- a/quiche/quic/moqt/moqt_parser.h
+++ b/quiche/quic/moqt/moqt_parser.h
@@ -12,9 +12,9 @@
 #include <cstdint>
 #include <optional>
 #include <string>
+#include <utility>
 
 #include "absl/base/nullability.h"
-#include "absl/cleanup/cleanup.h"
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/strings/str_cat.h"
@@ -63,12 +63,43 @@
   virtual void OnParsingError(MoqtError code, absl::string_view reason) = 0;
 };
 
+// MoqtStreamTypeParser reads the initial varint from a WebTransport stream to
+// determine its type before constructing either a control or a data stream
+// parser. Note that both of those parsers can be safely constructed from the
+// stream type parser even if the parser has not read the type yet.
+class QUICHE_EXPORT MoqtStreamTypeParser {
+ public:
+  explicit MoqtStreamTypeParser(webtransport::Stream* absl_nonnull stream)
+      : stream_(stream) {}
+  ~MoqtStreamTypeParser() = default;
+
+  // Move-only semantics to avoid a stream being accessed by two different
+  // parsers at the same time.
+  MoqtStreamTypeParser(const MoqtStreamTypeParser&) = delete;
+  MoqtStreamTypeParser& operator=(const MoqtStreamTypeParser&) = delete;
+  MoqtStreamTypeParser(MoqtStreamTypeParser&& other) noexcept;
+  MoqtStreamTypeParser& operator=(MoqtStreamTypeParser&& other) noexcept;
+
+  // Reads the first varint from the stream. Returns kUnavailable if the type
+  // has not been received yet.
+  absl::StatusOr<uint64_t> ReadStreamType();
+
+  std::optional<uint64_t> stream_type() const { return type_; }
+  webtransport::Stream* absl_nonnull stream() const { return stream_; }
+
+ private:
+  webtransport::Stream* absl_nonnull stream_;
+  std::optional<uint64_t> type_;
+  absl::Status status_ = absl::OkStatus();
+};
+
 // MoqtControlStreamParser unframes MoQT control messages from the control
 // stream without parsing the payload.
 class QUICHE_EXPORT MoqtControlStreamParser {
  public:
   explicit MoqtControlStreamParser(webtransport::Stream* absl_nonnull stream)
       : stream_(*stream) {}
+  explicit MoqtControlStreamParser(MoqtStreamTypeParser type_parser);
 
   // MoqtControlStreamParser is not movable, since reading from the same stream
   // through two different parsers would corrupt the state.
@@ -81,8 +112,6 @@
   // status if no complete message can be read; if FIN is read, `fin_read` will
   // be set to true.
   absl::StatusOr<MoqtRawControlMessage> ReadNextMessage();
-  // Reads the type of the first message on the stream.
-  absl::StatusOr<MoqtMessageType> ReadFirstMessageType();
 
   bool fin_read() const { return fin_read_; }
   webtransport::Stream* stream() const { return &stream_; }
@@ -99,7 +128,6 @@
   absl::Status ReadMessageType();
 
   webtransport::Stream& stream_;
-  std::optional<uint64_t> first_message_type_;
   std::optional<uint64_t> current_message_type_;
   std::optional<absl::Span<char>> current_message_remaining_;
   std::string current_message_;
@@ -266,6 +294,8 @@
   explicit MoqtDataParser(webtransport::Stream* stream,
                           MoqtDataParserVisitor* visitor)
       : stream_(*stream), visitor_(*visitor) {}
+  MoqtDataParser(MoqtStreamTypeParser type_parser,
+                 MoqtDataParserVisitor* visitor);
 
   // Reads all of the available objects on the stream.
   void ReadAllData();
@@ -344,6 +374,7 @@
   // Checks if we have encountered a FIN without data.  If so, processes it and
   // returns true.
   bool CheckForFinWithoutData();
+  void ProcessStreamType(uint64_t raw_type);
 
   void ParseError(absl::string_view reason);
 
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc
index d598c3d..3407da4 100644
--- a/quiche/quic/moqt/moqt_parser_test.cc
+++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -909,8 +909,6 @@
               StatusIs(absl::StatusCode::kInvalidArgument));
   EXPECT_THAT(parser.ReadNextMessage().status(),
               StatusIs(absl::StatusCode::kFailedPrecondition));
-  EXPECT_THAT(parser.ReadFirstMessageType().status(),
-              StatusIs(absl::StatusCode::kFailedPrecondition));
 }
 
 TEST_F(MoqtMessageSpecificTest, CannotAccessAfterError2) {
@@ -921,8 +919,6 @@
               StatusIs(absl::StatusCode::kInvalidArgument));
   EXPECT_THAT(parser.ReadNextMessage().status(),
               StatusIs(absl::StatusCode::kFailedPrecondition));
-  EXPECT_THAT(parser.ReadFirstMessageType(),
-              IsOkAndHolds(MoqtMessageType::kSubscribe));
 }
 
 TEST_F(MoqtMessageSpecificTest, FinMidType) {
@@ -1070,30 +1066,6 @@
                        HasSubstr("FIN on a control stream")));
 }
 
-TEST_F(MoqtMessageSpecificTest, ControlStreamReadType) {
-  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
-  MoqtControlStreamParser parser(&stream);
-  stream.Receive("\x03", false);
-  absl::StatusOr<MoqtMessageType> type = parser.ReadFirstMessageType();
-  EXPECT_THAT(type, IsOkAndHolds(MoqtMessageType::kSubscribe));
-}
-
-TEST_F(MoqtMessageSpecificTest, ControlStreamFinBeforeType) {
-  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
-  MoqtControlStreamParser parser(&stream);
-  stream.Receive("", true);
-  absl::StatusOr<MoqtMessageType> type = parser.ReadFirstMessageType();
-  EXPECT_EQ(type.status().code(), absl::StatusCode::kInvalidArgument);
-}
-
-TEST_F(MoqtMessageSpecificTest, ControlStreamFinInTheMiddleOfType) {
-  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
-  MoqtControlStreamParser parser(&stream);
-  stream.Receive("\xff", true);
-  absl::StatusOr<MoqtMessageType> type = parser.ReadFirstMessageType();
-  EXPECT_EQ(type.status().code(), absl::StatusCode::kInvalidArgument);
-}
-
 TEST_F(MoqtMessageSpecificTest, InvalidObjectStatus) {
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtParserTestVisitor data_visitor;
@@ -1817,4 +1789,105 @@
   EXPECT_FALSE(visitor_.last_message()->first_object_in_subgroup.has_value());
 }
 
+TEST_F(MoqtMessageSpecificTest, StreamTypeParserToControlStream) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  stream.Receive("\x03", false);
+  absl::StatusOr<uint64_t> type = type_parser.ReadStreamType();
+  EXPECT_THAT(type,
+              IsOkAndHolds(static_cast<uint64_t>(MoqtMessageType::kSubscribe)));
+  EXPECT_NE(type_parser.stream(), nullptr);
+
+  MoqtControlStreamParser control_parser(std::move(type_parser));
+  EXPECT_THAT(
+      type_parser.ReadStreamType(),  // NOLINT(bugprone-use-after-move)
+      StatusIs(absl::StatusCode::kInternal, HasSubstr("moved-from parser")));
+  EXPECT_EQ(control_parser.stream(), &stream);
+}
+
+TEST_F(MoqtMessageSpecificTest, StreamTypeParserToDataStream) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  stream.Receive("\x10", false);
+  absl::StatusOr<uint64_t> type = type_parser.ReadStreamType();
+  EXPECT_THAT(type, IsOkAndHolds(0x10));
+  EXPECT_THAT(type_parser.ReadStreamType(),
+              IsOkAndHolds(0x10));  // Second read should read the cached value.
+  EXPECT_NE(type_parser.stream(), nullptr);
+
+  MoqtParserTestVisitor data_visitor;
+  MoqtDataParser data_parser(std::move(type_parser), &data_visitor);
+  EXPECT_THAT(
+      type_parser.ReadStreamType(),  // NOLINT(bugprone-use-after-move)
+      StatusIs(absl::StatusCode::kInternal, HasSubstr("moved-from parser")));
+  EXPECT_EQ(data_parser.stream_type(), MoqtDataStreamType::FromValue(0x10));
+}
+
+TEST_F(MoqtMessageSpecificTest, EmptyStreamTypeParserToDataStream) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  MoqtParserTestVisitor data_visitor;
+  MoqtDataParser data_parser(std::move(type_parser), &data_visitor);
+  stream.Receive("\x10", false);
+  data_parser.ReadStreamType();
+  EXPECT_EQ(data_parser.stream_type(), MoqtDataStreamType::FromValue(0x10));
+}
+
+TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinBeforeType) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  stream.Receive("", true);
+  absl::StatusOr<uint64_t> type = type_parser.ReadStreamType();
+  EXPECT_THAT(
+      type,
+      StatusIs(
+          absl::StatusCode::kInvalidArgument,
+          HasSubstr(
+              "FIN received before or immediately after the stream type")));
+}
+
+TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinInTheMiddleOfType) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  stream.Receive("\xff", true);
+  absl::StatusOr<uint64_t> type = type_parser.ReadStreamType();
+  EXPECT_THAT(
+      type,
+      StatusIs(
+          absl::StatusCode::kInvalidArgument,
+          HasSubstr(
+              "FIN received before or immediately after the stream type")));
+}
+
+TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinAfterType) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  stream.Receive("\x03", true);
+  absl::StatusOr<uint64_t> type = type_parser.ReadStreamType();
+  EXPECT_THAT(
+      type,
+      StatusIs(
+          absl::StatusCode::kInvalidArgument,
+          HasSubstr(
+              "FIN received before or immediately after the stream type")));
+}
+
+TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinForPadding) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  stream.Receive("\xa6\xd3", true);
+  absl::StatusOr<uint64_t> type = type_parser.ReadStreamType();
+  EXPECT_THAT(
+      type, IsOkAndHolds(static_cast<uint64_t>(MoqtDataStreamType::kPadding)));
+}
+
+TEST_F(MoqtMessageSpecificTest, StreamTypeParserMovedFrom) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtStreamTypeParser type_parser(&stream);
+  MoqtControlStreamParser control_parser(std::move(type_parser));
+  EXPECT_THAT(
+      type_parser.ReadStreamType(),  // NOLINT(bugprone-use-after-move)
+      StatusIs(absl::StatusCode::kInternal, HasSubstr("moved-from parser")));
+}
+
 }  // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index c9b54f5..909d49d 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -194,8 +194,8 @@
 void MoqtSession::OnIncomingUnidirectionalStreamAvailable() {
   while (webtransport::Stream* stream =
              session_->AcceptIncomingUnidirectionalStream()) {
-    stream->SetVisitor(
-        std::make_unique<IncomingDataStream>(stream, this, callbacks_.clock));
+    stream->SetVisitor(std::make_unique<IncomingDataStream>(
+        MoqtStreamTypeParser(stream), this, callbacks_.clock));
     stream->visitor()->OnCanRead();
   }
 }
@@ -886,24 +886,23 @@
 }
 
 void MoqtSession::UnknownBidiStream::OnCanRead() {
-  absl::StatusOr<MoqtMessageType> message_type =
-      parser_->ReadFirstMessageType();
-  if (absl::IsUnavailable(message_type.status())) {
+  absl::StatusOr<uint64_t> type = parser_.ReadStreamType();
+  if (absl::IsUnavailable(type.status())) {
     return;
   }
-  if (absl::IsInvalidArgument(message_type.status())) {
+  if (absl::IsInvalidArgument(type.status())) {
     // Received a FIN before any type has been available, which is malformed.
-    session_->Error(MoqtError::kProtocolViolation,
-                    message_type.status().message());
+    session_->Error(MoqtError::kProtocolViolation, type.status().message());
     return;
   }
-  if (!message_type.ok()) {
+  if (!type.ok()) {
     // The result is neither of "OK", "no type available", or "parse error".
     // This is unexpected; treat it as an internal error, and reset the stream.
     stream_->ResetWithUserCode(kResetCodeInternalError);
     return;
   }
-  switch (*message_type) {
+  MoqtMessageType message_type = static_cast<MoqtMessageType>(*type);
+  switch (message_type) {
     case MoqtMessageType::kSetup: {
       if (session_->control_stream_.GetIfAvailable() != nullptr) {
         session_->Error(MoqtError::kProtocolViolation,
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index b031dc2..21ff0a6 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -234,9 +234,7 @@
     // responsible for calling stream->SetVisitor().
     UnknownBidiStream(MoqtSession* session,
                       webtransport::Stream* absl_nonnull stream)
-        : session_(session),
-          stream_(stream),
-          parser_(std::make_unique<MoqtControlStreamParser>(stream)) {}
+        : session_(session), stream_(stream), parser_(stream) {}
     ~UnknownBidiStream() {}
 
     // webtransport::StreamVisitor overrides.
@@ -249,7 +247,7 @@
    private:
     MoqtSession* session_;
     webtransport::Stream* stream_;
-    std::unique_ptr<MoqtControlStreamParser> parser_;
+    MoqtStreamTypeParser parser_;
   };
 
   class QUICHE_EXPORT ControlStream : public MoqtBidiStreamBase {
diff --git a/quiche/quic/moqt/moqt_uni_stream.h b/quiche/quic/moqt/moqt_uni_stream.h
index b681125..e1234fb 100644
--- a/quiche/quic/moqt/moqt_uni_stream.h
+++ b/quiche/quic/moqt/moqt_uni_stream.h
@@ -210,13 +210,17 @@
 class QUICHE_EXPORT IncomingDataStream : public webtransport::StreamVisitor,
                                          public MoqtDataParserVisitor {
  public:
+  IncomingDataStream(MoqtStreamTypeParser type_parser,
+                     SessionToUniStreamInterface* absl_nonnull session,
+                     const quic::QuicClock* absl_nonnull clock)
+      : stream_(type_parser.stream()),
+        parser_(std::move(type_parser), this),
+        session_(session),
+        clock_(clock) {}
   IncomingDataStream(webtransport::Stream* absl_nonnull stream,
                      SessionToUniStreamInterface* absl_nonnull session,
                      const quic::QuicClock* absl_nonnull clock)
-      : stream_(stream),
-        parser_(stream, this),
-        session_(session),
-        clock_(clock) {}
+      : IncomingDataStream(MoqtStreamTypeParser(stream), session, clock) {}
   ~IncomingDataStream();
 
   // webtransport::StreamVisitor implementation.