Implement RTCQuicStream.readInto()

Bug: 874296
Tbr: hbos@chromium.org
Change-Id: Id2905574237fe9221d49a1e772c9ba0f3e7957e1
Reviewed-on: https://chromium-review.googlesource.com/c/1289436
Commit-Queue: Steve Anton <steveanton@chromium.org>
Reviewed-by: Steve Anton <steveanton@chromium.org>
Cr-Commit-Position: refs/heads/master@{#608878}

third_party/WebKit/LayoutTests/external/wpt/webrtc/RTCQuicStream.https.html

@@ -55,10 +55,8 @@ promise_test(async t => {
 promise_test(async t => {
   const [ localQuicTransport, remoteQuicTransport ] =
       await makeTwoConnectedQuicTransports(t);
-  const firstLocalStream = localQuicTransport.createStream();
-  firstLocalStream.finish();
-  const secondLocalStream = localQuicTransport.createStream();
-  secondLocalStream.finish();
+  localQuicTransport.createStream().finish();
+  localQuicTransport.createStream().finish();
   const remoteWatcher =
       new EventWatcher(t, remoteQuicTransport, [ 'quicstream', 'statechange' ]);
   const { stream: firstRemoteStream } =
@@ -105,20 +103,6 @@ promise_test(async t => {
   assert_equals(localStream.state, 'closed');
 }, `reset() following finish() changes state to 'closed'.`);
 
-promise_test(async t => {
-  const [ localQuicTransport, remoteQuicTransport ] =
-      await makeTwoConnectedQuicTransports(t);
-  const localStream = localQuicTransport.createStream();
-  localStream.finish();
-  const remoteWatcher = new EventWatcher(t, remoteQuicTransport, 'quicstream');
-  const { stream: remoteStream } = await remoteWatcher.wait_for('quicstream');
-  assert_equals(remoteStream.state, 'open');
-  const remoteStreamWatcher = new EventWatcher(t, remoteStream, 'statechange');
-  await remoteStreamWatcher.wait_for('statechange');
-  assert_equals(remoteStream.state, 'closing');
-}, 'createStream() followed by finish() fires a quicstream event followed by ' +
-    `a statechange event to 'closing' on the remote side.`);
-
 promise_test(async t => {
   const [ localQuicTransport, remoteQuicTransport ] =
       await makeTwoConnectedQuicTransports(t);
@@ -133,33 +117,6 @@ promise_test(async t => {
 }, 'createStream() followed by reset() fires a quicstream event followed ' +
     `by a statechange event to 'closed' on the remote side.`);
 
-promise_test(async t => {
-  const [ localQuicTransport, remoteQuicTransport ] =
-      await makeTwoConnectedQuicTransports(t);
-  remoteQuicTransport.onquicstream = ({ stream }) => stream.reset();
-  const localStream = localQuicTransport.createStream();
-  localStream.finish();
-  const localWatcher = new EventWatcher(t, localStream, 'statechange');
-  await localWatcher.wait_for('statechange');
-  assert_equals(localStream.state, 'closed');
-}, 'finish() on a remote stream that has already finished fires a ' +
-    `statechange event to 'closed' on the remote side.`);
-
-promise_test(async t => {
-  const [ localQuicTransport, remoteQuicTransport ] =
-      await makeTwoConnectedQuicTransports(t);
-  const localStream = localQuicTransport.createStream();
-  localStream.finish();
-  localStream.reset();
-  const remoteWatcher = new EventWatcher(t, remoteQuicTransport, 'quicstream');
-  const { stream: remoteStream } = await remoteWatcher.wait_for('quicstream');
-  const remoteStreamWatcher = new EventWatcher(t, remoteStream, 'statechange');
-  await remoteStreamWatcher.wait_for('statechange');
-  assert_equals(remoteStream.state, 'closing');
-  await remoteStreamWatcher.wait_for('statechange');
-  assert_equals(remoteStream.state, 'closed');
-}, 'finish() then reset() fires two statechange events on the remote side.');
-
 promise_test(async t => {
   const [ localQuicTransport, remoteQuicTransport ] =
     await makeTwoConnectedQuicTransports(t);
@@ -213,6 +170,24 @@ promise_test(async t => {
 }, 'write() throws if total write buffered amount would be greater than ' +
     'maxWriteBufferedAmount.');
 
+promise_test(async t => {
+  const [ localQuicTransport, remoteQuicTransport ] =
+      await makeTwoConnectedQuicTransports(t);
+  const localStream = localQuicTransport.createStream();
+  localStream.write(new Uint8Array(10));
+  const remoteWatcher = new EventWatcher(t, remoteQuicTransport, 'quicstream');
+  await remoteWatcher.wait_for('quicstream');
+}, 'write() causes quicstream event to fire on the remote transport.');
+
+promise_test(async t => {
+  const [ localQuicTransport, remoteQuicTransport ] =
+      await makeTwoConnectedQuicTransports(t);
+  const localStream = localQuicTransport.createStream();
+  localStream.finish();
+  const remoteWatcher = new EventWatcher(t, remoteQuicTransport, 'quicstream');
+  await remoteWatcher.wait_for('quicstream');
+}, 'finish() causes quicstream event to fire on the remote transport.');
+
 promise_test(async t => {
   const [ localQuicTransport, remoteQuicTransport ] =
       await makeTwoConnectedQuicTransports(t);
@@ -302,4 +277,17 @@ closed_stream_test(async (t, stream) => {
       stream.waitForWriteBufferedAmountBelow(0));
 }, 'waitForWriteBufferedBelow() rejects with InvalidStateError.');
 
+promise_test(async t => {
+  const [ localQuicTransport, remoteQuicTransport ] =
+      await makeTwoConnectedQuicTransports(t);
+  const localStream = localQuicTransport.createStream();
+  assert_object_equals(
+      localStream.readInto(new Uint8Array(10)),
+      { amount: 0, finished: false });
+}, 'readInto() on new local stream returns amount 0.');
+
+closed_stream_test(async (t, stream) => {
+  assert_throws('InvalidStateError', () => stream.readInto(new Uint8Array(1)));
+}, 'readInto() throws InvalidStateError.');
+
 </script>

third_party/WebKit/LayoutTests/webexposed/global-interface-listing-expected.txt

@@ -5591,6 +5591,7 @@ interface RTCPeerConnectionIceEvent : Event
     method constructor
 interface RTCQuicStream : EventTarget
     attribute @@toStringTag
+    getter maxReadBufferedAmount
     getter maxWriteBufferedAmount
     getter onstatechange
     getter readBufferedAmount
@@ -5599,6 +5600,7 @@ interface RTCQuicStream : EventTarget
     getter writeBufferedAmount
     method constructor
     method finish
+    method readInto
     method reset
     method waitForWriteBufferedAmountBelow
     method write

third_party/blink/renderer/modules/modules_idl_files.gni

@@ -623,6 +623,7 @@ modules_dictionary_idl_files =
           "peerconnection/rtc_peer_connection_ice_event_init.idl",
           "peerconnection/rtc_quic_parameters.idl",
           "peerconnection/rtc_quic_stream_event_init.idl",
+          "peerconnection/rtc_quic_stream_read_result.idl",
           "peerconnection/rtc_rtcp_parameters.idl",
           "peerconnection/rtc_rtp_capabilities.idl",
           "peerconnection/rtc_rtp_codec_capability.idl",

third_party/blink/renderer/modules/peerconnection/rtc_quic_stream.cc

@@ -3,6 +3,7 @@
 // found in the LICENSE file.
 #include "third_party/blink/renderer/modules/peerconnection/rtc_quic_stream.h"
 
+#include "base/containers/span.h"
 #include "third_party/blink/renderer/bindings/core/v8/script_promise_resolver.h"
 #include "third_party/blink/renderer/core/dom/events/event.h"
 #include "third_party/blink/renderer/platform/bindings/exception_state.h"
@@ -10,6 +11,7 @@
 namespace blink {
 
 const uint32_t RTCQuicStream::kWriteBufferSize = 4 * 1024;
+const uint32_t RTCQuicStream::kReadBufferSize = 4 * 1024;
 
 class RTCQuicStream::PendingWriteBufferedAmountPromise
     : public GarbageCollected<PendingWriteBufferedAmountPromise> {
@@ -59,7 +61,11 @@ String RTCQuicStream::state() const {
 }
 
 uint32_t RTCQuicStream::readBufferedAmount() const {
-  return read_buffered_amount_;
+  return receive_buffer_.size();
+}
+
+uint32_t RTCQuicStream::maxReadBufferedAmount() const {
+  return kReadBufferSize;
 }
 
 uint32_t RTCQuicStream::writeBufferedAmount() const {
@@ -70,6 +76,33 @@ uint32_t RTCQuicStream::maxWriteBufferedAmount() const {
   return kWriteBufferSize;
 }
 
+RTCQuicStreamReadResult* RTCQuicStream::readInto(
+    NotShared<DOMUint8Array> data,
+    ExceptionState& exception_state) {
+  if (RaiseIfNotReadable(exception_state)) {
+    return 0;
+  }
+  uint32_t read_amount = static_cast<uint32_t>(receive_buffer_.ReadInto(
+      base::make_span(data.View()->Data(), data.View()->length())));
+  if (!received_fin_ && read_amount > 0) {
+    proxy_->MarkReceivedDataConsumed(read_amount);
+  }
+  if (receive_buffer_.empty() && received_fin_) {
+    read_fin_ = true;
+    if (wrote_fin_) {
+      DCHECK_EQ(state_, RTCQuicStreamState::kClosing);
+      Close(CloseReason::kReadWriteFinished);
+    } else {
+      DCHECK_EQ(state_, RTCQuicStreamState::kOpen);
+      state_ = RTCQuicStreamState::kClosing;
+    }
+  }
+  auto* result = RTCQuicStreamReadResult::Create();
+  result->setAmount(read_amount);
+  result->setFinished(read_fin_);
+  return result;
+}
+
 void RTCQuicStream::write(NotShared<DOMUint8Array> data,
                           ExceptionState& exception_state) {
   if (RaiseIfNotWritable(exception_state)) {
@@ -103,7 +136,7 @@ void RTCQuicStream::finish() {
   }
   proxy_->WriteData({}, /*fin=*/true);
   wrote_fin_ = true;
-  if (readable_) {
+  if (!read_fin_) {
     DCHECK_EQ(state_, RTCQuicStreamState::kOpen);
     state_ = RTCQuicStreamState::kClosing;
     RejectPendingWaitForWriteBufferedAmountBelowPromises();
@@ -139,6 +172,22 @@ ScriptPromise RTCQuicStream::waitForWriteBufferedAmountBelow(
   return promise;
 }
 
+bool RTCQuicStream::RaiseIfNotReadable(ExceptionState& exception_state) {
+  if (read_fin_) {
+    exception_state.ThrowDOMException(
+        DOMExceptionCode::kInvalidStateError,
+        "The stream is not readable: The end of the stream has been read.");
+    return true;
+  }
+  if (IsClosed()) {
+    exception_state.ThrowDOMException(
+        DOMExceptionCode::kInvalidStateError,
+        "The stream is not readable: The stream is closed.");
+    return true;
+  }
+  return false;
+}
+
 bool RTCQuicStream::RaiseIfNotWritable(ExceptionState& exception_state) {
   if (wrote_fin_) {
     exception_state.ThrowDOMException(
@@ -176,20 +225,10 @@ void RTCQuicStream::OnRemoteReset() {
 }
 
 void RTCQuicStream::OnDataReceived(Vector<uint8_t> data, bool fin) {
-  if (!fin) {
-    return;
-  }
-  DCHECK_NE(state_, RTCQuicStreamState::kClosed);
-  DCHECK(readable_);
-  readable_ = false;
-  if (!wrote_fin_) {
-    DCHECK_EQ(state_, RTCQuicStreamState::kOpen);
-    state_ = RTCQuicStreamState::kClosing;
-  } else {
-    DCHECK_EQ(state_, RTCQuicStreamState::kClosing);
-    Close(CloseReason::kReadWriteFinished);
-  }
-  DispatchEvent(*Event::Create(event_type_names::kStatechange));
+  DCHECK(!received_fin_);
+  DCHECK_LE(data.size(), kReadBufferSize - receive_buffer_.size());
+  received_fin_ = fin;
+  receive_buffer_.Append(std::move(data));
 }
 
 void RTCQuicStream::OnWriteDataConsumed(uint32_t amount) {
@@ -250,7 +289,7 @@ void RTCQuicStream::Close(CloseReason reason) {
   }
 
   // Clear observable state.
-  readable_ = false;
+  receive_buffer_.Clear();
   write_buffered_amount_ = 0;
 
   // It's illegal to resolve or reject promises when the ExecutionContext is

third_party/blink/renderer/modules/peerconnection/rtc_quic_stream.h

@@ -11,6 +11,8 @@
 #include "third_party/blink/renderer/modules/event_target_modules.h"
 #include "third_party/blink/renderer/modules/modules_export.h"
 #include "third_party/blink/renderer/modules/peerconnection/adapters/quic_stream_proxy.h"
+#include "third_party/blink/renderer/modules/peerconnection/byte_buffer_queue.h"
+#include "third_party/blink/renderer/modules/peerconnection/rtc_quic_stream_read_result.h"
 #include "third_party/blink/renderer/modules/peerconnection/rtc_quic_transport.h"
 
 namespace blink {
@@ -30,6 +32,7 @@ class MODULES_EXPORT RTCQuicStream final : public EventTargetWithInlineData,
  public:
   // TODO(steveanton): These maybe should be adjustable.
   static const uint32_t kWriteBufferSize;
+  static const uint32_t kReadBufferSize;
 
   enum class CloseReason {
     // Both read and write sides have been finished.
@@ -56,8 +59,11 @@ class MODULES_EXPORT RTCQuicStream final : public EventTargetWithInlineData,
   RTCQuicTransport* transport() const;
   String state() const;
   uint32_t readBufferedAmount() const;
+  uint32_t maxReadBufferedAmount() const;
   uint32_t writeBufferedAmount() const;
   uint32_t maxWriteBufferedAmount() const;
+  RTCQuicStreamReadResult* readInto(NotShared<DOMUint8Array> data,
+                                    ExceptionState& exception_state);
   void write(NotShared<DOMUint8Array> data, ExceptionState& exception_state);
   void finish();
   void reset();
@@ -82,6 +88,7 @@ class MODULES_EXPORT RTCQuicStream final : public EventTargetWithInlineData,
   void OnDataReceived(Vector<uint8_t> data, bool fin) override;
   void OnWriteDataConsumed(uint32_t amount) override;
 
+  bool RaiseIfNotReadable(ExceptionState&);
   bool RaiseIfNotWritable(ExceptionState&);
 
   // Permenantly closes the RTCQuicStream with the given reason.
@@ -95,8 +102,16 @@ class MODULES_EXPORT RTCQuicStream final : public EventTargetWithInlineData,
 
   Member<RTCQuicTransport> transport_;
   RTCQuicStreamState state_ = RTCQuicStreamState::kOpen;
-  bool readable_ = true;
-  uint32_t read_buffered_amount_ = 0;
+
+  // Data that has been received but not read.
+  // OnDataReceived() appends to the read buffer.
+  // readInto() will read out from the front of the buffer.
+  ByteBufferQueue receive_buffer_;
+  // True if the fin has been received from the network.
+  bool received_fin_ = false;
+  // True if the fin has been read out via readInto(). This signifies that the
+  // RTCQuicStream is closed for reading.
+  bool read_fin_ = false;
 
   // Amount of bytes written but may not yet have been sent by the underlying
   // P2PQuicStream.

third_party/blink/renderer/modules/peerconnection/rtc_quic_stream.idl

@@ -19,8 +19,10 @@ enum RTCQuicStreamState {
     readonly attribute RTCQuicTransport transport;
     readonly attribute RTCQuicStreamState state;
     readonly attribute unsigned long readBufferedAmount;
+    readonly attribute unsigned long maxReadBufferedAmount;
     readonly attribute unsigned long writeBufferedAmount;
     readonly attribute unsigned long maxWriteBufferedAmount;
+    [RaisesException] RTCQuicStreamReadResult readInto(Uint8Array data);
     [RaisesException] void write(Uint8Array data);
     void finish();
     void reset();

third_party/blink/renderer/modules/peerconnection/rtc_quic_stream_read_result.idl

+// Copyright 2018 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.
+
+// https://w3c.github.io/webrtc-quic/#dom-rtcquicstreamreadresult
+dictionary RTCQuicStreamReadResult {
+    unsigned long amount;
+    boolean finished;
+};

third_party/blink/renderer/modules/peerconnection/rtc_quic_stream_test.cc

@@ -510,4 +510,383 @@ TEST_F(
   RunUntilIdle();
 }
 
+// The following group tests readInto(), readBufferedAmount(), and
+// maxReadBufferedAmount().
+
+static base::span<const uint8_t> GetSpan(NotShared<DOMUint8Array> data) {
+  return base::make_span(data.View()->Data(), data.View()->length());
+}
+
+// Test that readInto() with an empty data buffer succeeds but does not post
+// MarkReceivedDataConsumed() to the underlying P2PQuicStream.
+TEST_F(RTCQuicStreamTest, ReadIntoEmptyDoesNotPostMarkReceivedDataConsumed) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+  EXPECT_CALL(*p2p_quic_stream, MarkReceivedDataConsumed(_)).Times(0);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  EXPECT_EQ(0u, stream->readBufferedAmount());
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(0));
+  RTCQuicStreamReadResult* result =
+      stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+  EXPECT_EQ(0u, result->amount());
+  EXPECT_FALSE(result->finished());
+  EXPECT_EQ(0u, stream->readBufferedAmount());
+
+  RunUntilIdle();
+}
+
+// Test that data delived via OnDataReceived() increases readBufferedAmount.
+TEST_F(RTCQuicStreamTest, OnDataReceivedIncreasesReadBufferedAmount) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({1, 2, 3}, /*fin=*/false);
+
+  RunUntilIdle();
+
+  EXPECT_EQ(3u, stream->readBufferedAmount());
+}
+
+// Test that readInto() reads out data received from OnDataReceived() and
+// decreases readBufferedAmount.
+TEST_F(RTCQuicStreamTest, ReadIntoReadsDataFromOnDataReceived) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({1, 2, 3}, /*fin=*/false);
+
+  RunUntilIdle();
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(5));
+  RTCQuicStreamReadResult* result =
+      stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+  EXPECT_EQ(3u, result->amount());
+  EXPECT_FALSE(result->finished());
+  EXPECT_THAT(GetSpan(read_buffer), ElementsAre(1, 2, 3, 0, 0));
+  EXPECT_EQ(0u, stream->readBufferedAmount());
+
+  RunUntilIdle();
+}
+
+// Test that readInto() posts MarkReceivedDataConsumed() with the amount of
+// bytes read to the underlying P2PQuicStream.
+TEST_F(RTCQuicStreamTest, ReadIntoPostsMarkReceivedDataConsumed) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+  EXPECT_CALL(*p2p_quic_stream, MarkReceivedDataConsumed(3)).Times(1);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({1, 2, 3}, /*fin=*/false);
+
+  RunUntilIdle();
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(5));
+  RTCQuicStreamReadResult* result =
+      stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+  EXPECT_EQ(3u, result->amount());
+  EXPECT_FALSE(result->finished());
+
+  RunUntilIdle();
+}
+
+// Test that readInto() returns {finished: true} if OnDataReceived() has
+// indicated the stream is finished without receiving any data.
+TEST_F(RTCQuicStreamTest, ReadIntoReadsBareFin) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({}, /*fin=*/true);
+
+  RunUntilIdle();
+
+  NotShared<DOMUint8Array> empty_read_buffer(DOMUint8Array::Create(0));
+  RTCQuicStreamReadResult* result =
+      stream->readInto(empty_read_buffer, ASSERT_NO_EXCEPTION);
+  EXPECT_EQ(0u, result->amount());
+  EXPECT_TRUE(result->finished());
+
+  RunUntilIdle();
+}
+
+// Test that readInto() indicates finished once all buffered received data has
+// been read.
+TEST_F(RTCQuicStreamTest, ReadIntoReadsBufferedDataAndFinish) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({1, 2, 3}, /*fin=*/true);
+
+  RunUntilIdle();
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(3));
+  RTCQuicStreamReadResult* result =
+      stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+  EXPECT_EQ(3u, result->amount());
+  EXPECT_TRUE(result->finished());
+  EXPECT_THAT(GetSpan(read_buffer), ElementsAre(1, 2, 3));
+
+  RunUntilIdle();
+}
+
+// Test that readInto() does not indicate finished until all buffered data has
+// been read out, even if the fin has already been received.
+TEST_F(RTCQuicStreamTest, ReadIntoReadsPartialDataBeforeFin) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({1, 2, 3}, /*fin=*/true);
+
+  RunUntilIdle();
+
+  {
+    NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(2));
+    RTCQuicStreamReadResult* result =
+        stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+    EXPECT_EQ(2u, result->amount());
+    EXPECT_FALSE(result->finished());
+    EXPECT_THAT(GetSpan(read_buffer), ElementsAre(1, 2));
+  }
+
+  {
+    NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(2));
+    RTCQuicStreamReadResult* result =
+        stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+    EXPECT_EQ(1u, result->amount());
+    EXPECT_TRUE(result->finished());
+    EXPECT_THAT(GetSpan(read_buffer), ElementsAre(3, 0));
+  }
+
+  RunUntilIdle();
+}
+
+// Test that readInto() does not post MarkReceivedDataConsumed() to the
+// underlying P2PQuicStream once the finish flag has been delivered in
+// OnDataReceived().
+TEST_F(RTCQuicStreamTest,
+       ReadIntoDoesNotPostMarkReceivedDataConsumedOnceFinReceived) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+  EXPECT_CALL(*p2p_quic_stream, MarkReceivedDataConsumed(_)).Times(0);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({1, 2, 3}, /*fin=*/true);
+
+  RunUntilIdle();
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(2));
+
+  RTCQuicStreamReadResult* result =
+      stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+  EXPECT_EQ(2u, result->amount());
+  EXPECT_FALSE(result->finished());
+
+  result = stream->readInto(read_buffer, ASSERT_NO_EXCEPTION);
+  EXPECT_EQ(1u, result->amount());
+  EXPECT_TRUE(result->finished());
+
+  RunUntilIdle();
+}
+
+// Test that readInto() throws an InvalidStateError if the finish flag has
+// already been read out.
+TEST_F(RTCQuicStreamTest, ReadIntoThrowsIfFinishAlreadyRead) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({}, /*fin=*/true);
+
+  RunUntilIdle();
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(2));
+  EXPECT_TRUE(stream->readInto(read_buffer, ASSERT_NO_EXCEPTION)->finished());
+
+  stream->readInto(read_buffer, scope.GetExceptionState());
+  EXPECT_EQ(DOMExceptionCode::kInvalidStateError,
+            scope.GetExceptionState().CodeAs<DOMExceptionCode>());
+}
+
+// Test that readInto() throws an InvalidStateError if the stream is closed.
+TEST_F(RTCQuicStreamTest, ReadIntoThrowsIfClosed) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+  stream->reset();
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(2));
+  stream->readInto(read_buffer, scope.GetExceptionState());
+  EXPECT_EQ(DOMExceptionCode::kInvalidStateError,
+            scope.GetExceptionState().CodeAs<DOMExceptionCode>());
+
+  RunUntilIdle();
+}
+
+// The following group tests state transitions with reset(), finish(), remote
+// reset() and remote finish()
+
+// Test that a OnRemoteReset() immediately transitions the state to 'closed'
+// and clears any buffered data.
+TEST_F(RTCQuicStreamTest, OnRemoteResetTransitionsToClosedAndClearsBuffers) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  stream->write(CreateUint8Array({1, 2}), ASSERT_NO_EXCEPTION);
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({5, 6, 7, 8}, /*fin=*/false);
+
+  RunUntilIdle();
+
+  EXPECT_EQ("open", stream->state());
+  EXPECT_EQ(2u, stream->writeBufferedAmount());
+  EXPECT_EQ(4u, stream->readBufferedAmount());
+
+  stream_delegate->OnRemoteReset();
+
+  RunUntilIdle();
+
+  EXPECT_EQ("closed", stream->state());
+  EXPECT_EQ(0u, stream->writeBufferedAmount());
+  EXPECT_EQ(0u, stream->readBufferedAmount());
+
+  RunUntilIdle();
+}
+
+// Test that calling finish() after reading out a remote finish transitions
+// state to 'closed'.
+TEST_F(RTCQuicStreamTest, FinishAfterReadingRemoteFinishTransitionsToClosed) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+  EXPECT_CALL(*p2p_quic_stream, WriteData(_, true)).Times(1);
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({}, /*fin=*/true);
+
+  RunUntilIdle();
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(10));
+  EXPECT_TRUE(stream->readInto(read_buffer, ASSERT_NO_EXCEPTION)->finished());
+
+  EXPECT_EQ("closing", stream->state());
+
+  stream->finish();
+
+  EXPECT_EQ("closed", stream->state());
+
+  RunUntilIdle();
+}
+
+// Test that reading out a remote finish after calling finish() transitions
+// state to 'closed'.
+TEST_F(RTCQuicStreamTest, ReadingRemoteFinishAfterFinishTransitionsToClosed) {
+  V8TestingScope scope;
+
+  P2PQuicStream::Delegate* stream_delegate = nullptr;
+  auto p2p_quic_stream = std::make_unique<MockP2PQuicStream>(&stream_delegate);
+  EXPECT_CALL(*p2p_quic_stream, WriteData(_, true)).Times(1);
+  Persistent<RTCQuicStream> stream =
+      CreateQuicStream(scope, p2p_quic_stream.get());
+
+  stream->finish();
+
+  EXPECT_EQ("closing", stream->state());
+
+  RunUntilIdle();
+
+  ASSERT_TRUE(stream_delegate);
+  stream_delegate->OnDataReceived({}, /*fin=*/true);
+
+  RunUntilIdle();
+
+  EXPECT_EQ("closing", stream->state());
+
+  NotShared<DOMUint8Array> read_buffer(DOMUint8Array::Create(10));
+  EXPECT_TRUE(stream->readInto(read_buffer, ASSERT_NO_EXCEPTION)->finished());
+
+  EXPECT_EQ("closed", stream->state());
+}
+
 }  // namespace blink