Added a net::BidirectionalStream to expose a bidirectional streaming interface

net/http/bidirectional_stream_unittest.cc

Index: net/http/bidirectional_stream_unittest.cc
diff --git a/net/http/bidirectional_stream_unittest.cc b/net/http/bidirectional_stream_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..54f275f30ca17f44bf3a66f58918519a669e6a17
--- /dev/null
+++ b/net/http/bidirectional_stream_unittest.cc
@@ -0,0 +1,779 @@
+// Copyright 2015 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 "net/http/bidirectional_stream.h"
+
+#include "base/macros.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/run_loop.h"
+#include "base/strings/string_number_conversions.h"
+#include "base/strings/string_piece.h"
+#include "base/time/time.h"
+#include "base/timer/mock_timer.h"
+#include "net/base/net_errors.h"
+#include "net/base/test_data_directory.h"
+#include "net/http/bidirectional_stream_request_info.h"
+#include "net/http/http_network_session.h"
+#include "net/http/http_response_headers.h"
+#include "net/log/net_log.h"
+#include "net/socket/socket_test_util.h"
+#include "net/spdy/spdy_session.h"
+#include "net/spdy/spdy_test_util_common.h"
+#include "net/spdy/spdy_test_util_common.h"
+#include "net/test/cert_test_util.h"
+#include "net/url_request/url_request_test_util.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace net {
+
+namespace {
+
+const char kBodyData[] = "Body data";
+const size_t kBodyDataSize = arraysize(kBodyData);
+// Size of the buffer to be allocated for each read.
+const size_t kReadBufferSize = 4096;
+
+// Delegate that reads data but does not send any data.
+class TestDelegateBase : public BidirectionalStream::Delegate {
+ public:
+  TestDelegateBase(IOBuffer* read_buf, int read_buf_len)
+      : TestDelegateBase(read_buf,
+                         read_buf_len,
+                         make_scoped_ptr(new base::Timer(false, false))) {}
+
+  TestDelegateBase(IOBuffer* read_buf,
+                   int read_buf_len,
+                   scoped_ptr<base::Timer> timer)
+      : read_buf_(read_buf),
+        read_buf_len_(read_buf_len),
+        timer_(timer.release()),
+        loop_(nullptr),
+        error_(OK),
+        on_read_complete_count_(0),
+        on_data_sent_count_(0),
+        do_not_start_read_(false),
+        run_until_completion_(false),
+        not_expect_callback_(false) {}
+
+  ~TestDelegateBase() override {}
+
+  void OnHeadersSent() override { CHECK(!not_expect_callback_); }
+
+  void OnHeadersReceived(const SpdyHeaderBlock& response_headers) override {
+    CHECK(!not_expect_callback_);
+
+    response_headers_ = response_headers;
+    if (!do_not_start_read_)
+      StartOrContinueReading();
+  }
+
+  void OnDataRead(int bytes_read) override {
+    CHECK(!not_expect_callback_);
+
+    ++on_read_complete_count_;
+    CHECK_GE(bytes_read, OK);
+    data_received_.append(read_buf_->data(), bytes_read);
+    if (!do_not_start_read_)
+      StartOrContinueReading();
+  }
+
+  void OnDataSent() override {
+    CHECK(!not_expect_callback_);
+
+    ++on_data_sent_count_;
+  }
+
+  void OnTrailersReceived(const SpdyHeaderBlock& trailers) override {
+    CHECK(!not_expect_callback_);
+
+    trailers_ = trailers;
+    if (run_until_completion_)
+      loop_->Quit();
+  }
+
+  void OnFailed(int error) override {
+    CHECK(!not_expect_callback_);
+    CHECK_EQ(OK, error_);
+    CHECK_NE(OK, error);
+
+    error_ = error;
+    if (run_until_completion_)
+      loop_->Quit();
+  }
+
+  void Start(const BidirectionalStreamRequestInfo& request_info,
+             RequestPriority priority,
+             HttpNetworkSession* session) {
+    stream_.reset(new BidirectionalStream(request_info, priority, session, this,
+                                          timer_.Pass()));
+    if (run_until_completion_)
+      loop_->Run();
+  }
+
+  void SendData(IOBuffer* data, int length, bool end_of_stream) {
+    not_expect_callback_ = true;
+    stream_->SendData(data, length, end_of_stream);
+    not_expect_callback_ = false;
+  }
+
+  // Starts or continues reading data from |stream_| until no more bytes
+  // can be read synchronously.
+  void StartOrContinueReading() {
+    int rv = ReadData();
+    while (rv > 0) {
+      rv = ReadData();
+    }
+    if (run_until_completion_ && rv == 0)
+      loop_->Quit();
+  }
+
+  // Calls ReadData on the |stream_| and updates internal states.
+  int ReadData() {
+    not_expect_callback_ = true;
+    int rv = stream_->ReadData(read_buf_.get(), read_buf_len_);
+    not_expect_callback_ = false;
+    if (rv > 0) {
+      data_received_.append(read_buf_->data(), rv);
+    }
+    return rv;
+  }
+
+  NextProto GetProtocol() const { return stream_->GetProtocol(); }
+
+  int64_t GetTotalReceivedBytes() const {
+    return stream_->GetTotalReceivedBytes();
+  }
+
+  int64_t GetTotalSentBytes() const { return stream_->GetTotalSentBytes(); }
+
+  // Const getters for internal states.
+  const std::string& data_received() const { return data_received_; }
+  int error() const { return error_; }
+  const SpdyHeaderBlock& response_headers() const { return response_headers_; }
+  const SpdyHeaderBlock& trailers() const { return trailers_; }
+  int on_read_complete_count() const { return on_read_complete_count_; }
+  int on_data_sent_count() const { return on_data_sent_count_; }
+
+  // Sets whether the delegate should automatically start reading.
+  void set_do_not_start_read(bool do_not_start_read) {
+    do_not_start_read_ = do_not_start_read;
+  }
+  // Sets whether the delegate should wait until the completion of the stream.
+  void SetRunUntilCompletion(bool run_until_completion) {
+    run_until_completion_ = run_until_completion;
+    loop_.reset(new base::RunLoop);
+  }
+
+ protected:
+  // Cancels |stream_|.
+  void CancelStream() { stream_->Cancel(); }
+  // Quits |loop_|.
+  void QuitLoop() { loop_->Quit(); }
+
+ private:
+  scoped_ptr<BidirectionalStream> stream_;
+  scoped_refptr<IOBuffer> read_buf_;
+  int read_buf_len_;
+  scoped_ptr<base::Timer> timer_;
+  std::string data_received_;
+  scoped_ptr<base::RunLoop> loop_;
+  SpdyHeaderBlock response_headers_;
+  SpdyHeaderBlock trailers_;
+  int error_;
+  int on_read_complete_count_;
+  int on_data_sent_count_;
+  bool do_not_start_read_;
+  bool run_until_completion_;
+  // This is to ensure that delegate callback is not invoked synchronously when
+  // calling into |stream_|.
+  bool not_expect_callback_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestDelegateBase);
+};
+
+// A delegate that cancels the request after response headers are received.
+class CancelStreamDelegate : public TestDelegateBase {
+ public:
+  CancelStreamDelegate(IOBuffer* buf, int buf_len)
+      : TestDelegateBase(buf, buf_len) {}
+  ~CancelStreamDelegate() override {}
+
+  void OnHeadersReceived(const SpdyHeaderBlock& response_headers) override {
+    TestDelegateBase::OnHeadersReceived(response_headers);
+    CancelStream();
+    QuitLoop();
+  }
+
+  void OnDataSent() override { NOTREACHED(); }
+
+  void OnDataRead(int bytes_read) override { NOTREACHED(); }
+
+  void OnTrailersReceived(const SpdyHeaderBlock& trailers) override {
+    NOTREACHED();
+  }
+
+  void OnFailed(int error) override { NOTREACHED(); }
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(CancelStreamDelegate);
+};
+
+// A Timer that does not start a delayed task unless the timer is fired.
+class MockTimer : public base::MockTimer {
+ public:
+  MockTimer() : base::MockTimer(false, false) {}
+  ~MockTimer() override {}
+
+  void Start(const tracked_objects::Location& posted_from,
+             base::TimeDelta delay,
+             const base::Closure& user_task) override {
+    // Sets a maximum delay, so the timer does not fire unless it is told to.
+    base::TimeDelta infinite_delay = base::TimeDelta::Max();
+    base::MockTimer::Start(posted_from, infinite_delay, user_task);
+  }
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(MockTimer);
+};
+
+}  // namespace
+
+class BidirectionalStreamTest : public testing::Test {
+ public:
+  BidirectionalStreamTest()
+      : spdy_util_(kProtoHTTP2, false),
+        ssl_data_(SSLSocketDataProvider(ASYNC, OK)),
+        session_deps_(kProtoHTTP2) {
+    ssl_data_.SetNextProto(kProtoHTTP2);
+    ssl_data_.cert = ImportCertFromFile(GetTestCertsDirectory(), "ok_cert.pem");
+  }
+
+ protected:
+  void TearDown() override {
+    if (sequenced_data_) {
+      EXPECT_TRUE(sequenced_data_->AllReadDataConsumed());
+      EXPECT_TRUE(sequenced_data_->AllWriteDataConsumed());
+    }
+    if (deterministic_data_) {
+      EXPECT_TRUE(deterministic_data_->AllReadDataConsumed());
+      EXPECT_TRUE(deterministic_data_->AllWriteDataConsumed());
+    }
+  }
+
+  // Initializes the session using SequencedSocketData.
+  void InitSession(MockRead* reads,
+                   size_t reads_count,
+                   MockWrite* writes,
+                   size_t writes_count,
+                   const SpdySessionKey& key) {
+    ASSERT_TRUE(ssl_data_.cert.get());
+    session_deps_.socket_factory->AddSSLSocketDataProvider(&ssl_data_);
+    sequenced_data_.reset(
+        new SequencedSocketData(reads, reads_count, writes, writes_count));
+    session_deps_.socket_factory->AddSocketDataProvider(sequenced_data_.get());
+    http_session_ = SpdySessionDependencies::SpdyCreateSession(&session_deps_);
+    session_ = CreateSecureSpdySession(http_session_.get(), key, BoundNetLog());
+    CHECK(!deterministic_data_);
+  }
+
+  // Initializes the session using DeterministicSocketData.
+  void InitSessionDeterministic(MockRead* reads,
+                                size_t reads_count,
+                                MockWrite* writes,
+                                size_t writes_count,
+                                const SpdySessionKey& key) {
+    ASSERT_TRUE(ssl_data_.cert.get());
+    session_deps_.deterministic_socket_factory->AddSSLSocketDataProvider(
+        &ssl_data_);
+    MockConnect connect_data(SYNCHRONOUS, OK);
+    deterministic_data_.reset(
+        new DeterministicSocketData(reads, reads_count, writes, writes_count));
+    deterministic_data_->set_connect_data(connect_data);
+    session_deps_.deterministic_socket_factory->AddSocketDataProvider(
+        deterministic_data_.get());
+    http_session_ =
+        SpdySessionDependencies::SpdyCreateSessionDeterministic(&session_deps_);
+    session_ = CreateSecureSpdySession(http_session_.get(), key, BoundNetLog());
+    CHECK(!sequenced_data_);
+  }
+
+  SpdyTestUtil spdy_util_;
+  SSLSocketDataProvider ssl_data_;
+  SpdySessionDependencies session_deps_;
+  scoped_ptr<SequencedSocketData> sequenced_data_;
+  scoped_ptr<DeterministicSocketData> deterministic_data_;
+  scoped_ptr<HttpNetworkSession> http_session_;
+  base::WeakPtr<SpdySession> session_;
+};
+
+TEST_F(BidirectionalStreamTest, CreateInsecureStream) {
+  BidirectionalStreamRequestInfo request_info;
+  request_info.method = "GET";
+  request_info.url = GURL("http://www.example.org/");
+
+  TestDelegateBase delegate(nullptr, 0);
+  HttpNetworkSession::Params params =
+      SpdySessionDependencies::CreateSessionParams(&session_deps_);
+  scoped_ptr<HttpNetworkSession> session(new HttpNetworkSession(params));
+  delegate.SetRunUntilCompletion(true);
+  delegate.Start(request_info, LOWEST, session.get());
+
+  EXPECT_EQ(ERR_DISALLOWED_URL_SCHEME, delegate.error());
+}
+
+// Simulates user calling ReadData after END_STREAM has been received in
+// BidirectionalStreamSpdyJob.
+TEST_F(BidirectionalStreamTest, TestReadDataAfterClose) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet("https://www.example.org", false, 1, LOWEST));
+  // Empty DATA frame with an END_STREAM flag.
+  scoped_ptr<SpdyFrame> end_stream(
+      spdy_util_.ConstructSpdyBodyFrame(1, nullptr, 0, true));
+  MockWrite writes[] = {
+      CreateMockWrite(*req.get(), 0),
+  };
+
+  const char* const kExtraResponseHeaders[] = {"header-name", "header-value"};
+
+  scoped_ptr<SpdyFrame> resp(
+      spdy_util_.ConstructSpdyGetSynReply(kExtraResponseHeaders, 1, 1));
+
+  scoped_ptr<SpdyFrame> body_frame(spdy_util_.ConstructSpdyBodyFrame(1, false));
+  // Last body frame has END_STREAM flag set.
+  scoped_ptr<SpdyFrame> last_body_frame(
+      spdy_util_.ConstructSpdyBodyFrame(1, true));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 1),       CreateMockRead(*body_frame, 2),
+      CreateMockRead(*body_frame, 3), CreateMockRead(*last_body_frame, 4),
+      MockRead(SYNCHRONOUS, 0, 5),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 443);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  BidirectionalStreamRequestInfo request_info;
+  request_info.method = "GET";
+  request_info.url = GURL("https://www.example.org/");
+  request_info.end_stream_on_headers = true;
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  // Create a MockTimer. Retain a raw pointer since the underlying
+  // BidirectionalStreamJob owns it.
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  delegate->set_do_not_start_read(true);
+
+  delegate->Start(request_info, DEFAULT_PRIORITY, http_session_.get());
+
+  // Write request, and deliver response headers.
+  deterministic_data_->RunFor(2);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(timer->IsRunning());
+  // ReadData returns asynchronously because no data is buffered.
+  int rv = delegate->ReadData();
+  EXPECT_EQ(ERR_IO_PENDING, rv);
+  // Deliver a DATA frame.
+  deterministic_data_->RunFor(1);
+  base::RunLoop().RunUntilIdle();
+  timer->Fire();
+  // Asynchronous completion callback is invoke.
+  EXPECT_EQ(1, delegate->on_read_complete_count());
+  EXPECT_EQ(kUploadDataSize * 1,
+            static_cast<int>(delegate->data_received().size()));
+
+  // Deliver the rest. Note that user has not called a second ReadData.
+  deterministic_data_->RunFor(3);
+  base::RunLoop().RunUntilIdle();
+  // ReadData now. Read should complete synchronously.
+  rv = delegate->ReadData();
+  EXPECT_EQ(kUploadDataSize * 2, rv);
+  rv = delegate->ReadData();
+  EXPECT_EQ(OK, rv);  // EOF.
+  EXPECT_EQ(1, delegate->on_read_complete_count());
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+  EXPECT_EQ(CountWriteBytes(writes, arraysize(writes)),
+            delegate->GetTotalSentBytes());
+  EXPECT_EQ(CountReadBytes(reads, arraysize(reads)),
+            delegate->GetTotalReceivedBytes());
+}
+
+TEST_F(BidirectionalStreamTest, TestInterleaveReadDataAndSendData) {
+  BufferedSpdyFramer framer(spdy_util_.spdy_version(), false);
+
+  scoped_ptr<SpdyFrame> req(spdy_util_.ConstructSpdyPost(
+      "https://www.example.org", 1, kBodyDataSize * 3, LOWEST, NULL, 0));
+  scoped_ptr<SpdyFrame> data_frame1(
+      framer.CreateDataFrame(1, kBodyData, kBodyDataSize, DATA_FLAG_NONE));
+  scoped_ptr<SpdyFrame> data_frame2(
+      framer.CreateDataFrame(1, kBodyData, kBodyDataSize, DATA_FLAG_NONE));
+  scoped_ptr<SpdyFrame> data_frame3(
+      framer.CreateDataFrame(1, kBodyData, kBodyDataSize, DATA_FLAG_FIN));
+  MockWrite writes[] = {
+      CreateMockWrite(*req, 0),  // request
+      CreateMockWrite(*data_frame1, 2), CreateMockWrite(*data_frame2, 4),
+      CreateMockWrite(*data_frame3, 6),
+  };
+
+  const char* const kExtraResponseHeaders[] = {"header-name", "header-value"};
+
+  scoped_ptr<SpdyFrame> resp(
+      spdy_util_.ConstructSpdyGetSynReply(kExtraResponseHeaders, 1, 1));
+
+  scoped_ptr<SpdyFrame> response_body_frame1(
+      spdy_util_.ConstructSpdyBodyFrame(1, false));
+  scoped_ptr<SpdyFrame> response_body_frame2(
+      spdy_util_.ConstructSpdyBodyFrame(1, true));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 1), CreateMockRead(*response_body_frame1, 3),
+      CreateMockRead(*response_body_frame2, 5), MockRead(ASYNC, 0, 7),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 443);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  BidirectionalStreamRequestInfo request_info;
+  request_info.method = "POST";
+  request_info.url = GURL("https://www.example.org/");
+  request_info.extra_headers.SetHeader(net::HttpRequestHeaders::kContentLength,
+                                       base::SizeTToString(kBodyDataSize * 3));
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  delegate->set_do_not_start_read(true);
+  delegate->Start(request_info, DEFAULT_PRIORITY, http_session_.get());
+  // Send the request and receive response headers.
+  deterministic_data_->RunFor(2);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(timer->IsRunning());
+
+  // Send a DATA frame.
+  scoped_refptr<StringIOBuffer> buf(
+      new StringIOBuffer(std::string(kBodyData, kBodyDataSize)));
+  delegate->SendData(buf.get(), buf->size(), false);
+  deterministic_data_->RunFor(1);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(1, delegate->on_data_sent_count());
+
+  // ReadData and it should return asynchronously because no data is buffered.
+  int rv = delegate->ReadData();
+  EXPECT_EQ(ERR_IO_PENDING, rv);
+  // Deliver a DATA frame, and fire the timer.
+  deterministic_data_->RunFor(1);
+  base::RunLoop().RunUntilIdle();
+  timer->Fire();
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(1, delegate->on_read_complete_count());
+
+  // Send a DATA frame.
+  delegate->SendData(buf.get(), buf->size(), false);
+  deterministic_data_->RunFor(1);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(2, delegate->on_data_sent_count());
+
+  // ReadData and it should return asynchronously because no data is buffered.
+  rv = delegate->ReadData();
+  EXPECT_EQ(ERR_IO_PENDING, rv);
+  // Deliver a DATA frame, and fire the timer.
+  deterministic_data_->RunFor(1);
+  timer->Fire();
+  base::RunLoop().RunUntilIdle();
+  // Last DATA frame is read. Server half closes.
+  EXPECT_EQ(2, delegate->on_read_complete_count());
+
+  // Send last body frame. Client half closes.
+  delegate->SendData(buf.get(), buf->size(), true);
+  deterministic_data_->RunFor(1);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(3, delegate->on_data_sent_count());
+
+  // OnClose is invoked since both sides are closed.
+  rv = delegate->ReadData();
+  EXPECT_EQ(OK, rv);
+  deterministic_data_->RunFor(1);
+  EXPECT_EQ(2, delegate->on_read_complete_count());
+  EXPECT_EQ(3, delegate->on_data_sent_count());
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+  EXPECT_EQ(CountWriteBytes(writes, arraysize(writes)),
+            delegate->GetTotalSentBytes());
+  EXPECT_EQ(CountReadBytes(reads, arraysize(reads)),
+            delegate->GetTotalReceivedBytes());
+}
+
+// Tests that BidirectionalStreamSpdyJob::OnClose will complete any remaining
+// read even if the read queue is empty.
+TEST_F(BidirectionalStreamTest, TestCompleteAsyncRead) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet("https://www.example.org", false, 1, LOWEST));
+  // Empty DATA frame with an END_STREAM flag.
+  scoped_ptr<SpdyFrame> end_stream(
+      spdy_util_.ConstructSpdyBodyFrame(1, nullptr, 0, true));
+
+  MockWrite writes[] = {CreateMockWrite(*req.get(), 0)};
+
+  scoped_ptr<SpdyFrame> resp(
+      spdy_util_.ConstructSpdyGetSynReply(nullptr, 0, 1));
+
+  scoped_ptr<SpdyFrame> response_body_frame(
+      spdy_util_.ConstructSpdyBodyFrame(1, nullptr, 0, true));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 1), CreateMockRead(*response_body_frame, 2),
+      MockRead(SYNCHRONOUS, 0, 3),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 443);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  BidirectionalStreamRequestInfo request_info;
+  request_info.method = "GET";
+  request_info.url = GURL("https://www.example.org/");
+  request_info.end_stream_on_headers = true;
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  delegate->set_do_not_start_read(true);
+  delegate->Start(request_info, DEFAULT_PRIORITY, http_session_.get());
+  // Write request, and deliver response headers.
+  deterministic_data_->RunFor(2);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(timer->IsRunning());
+
+  // Let ReadData return asynchronously.
+  int rv = delegate->ReadData();
+  EXPECT_EQ(ERR_IO_PENDING, rv);
+  // Deliver END_STREAM.
+  // OnClose should trigger completion of the remaining read.
+  deterministic_data_->RunFor(3);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(1, delegate->on_read_complete_count());
+  EXPECT_EQ(0u, delegate->data_received().size());
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+  EXPECT_EQ(CountWriteBytes(writes, arraysize(writes)),
+            delegate->GetTotalSentBytes());
+  EXPECT_EQ(CountReadBytes(reads, arraysize(reads)),
+            delegate->GetTotalReceivedBytes());
+}
+
+TEST_F(BidirectionalStreamTest, TestBuffering) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet("https://www.example.org", false, 1, LOWEST));
+  // Empty DATA frame with an END_STREAM flag.
+  scoped_ptr<SpdyFrame> end_stream(
+      spdy_util_.ConstructSpdyBodyFrame(1, nullptr, 0, true));
+
+  MockWrite writes[] = {CreateMockWrite(*req.get(), 0)};
+
+  const char* const kExtraResponseHeaders[] = {"header-name", "header-value"};
+
+  scoped_ptr<SpdyFrame> resp(
+      spdy_util_.ConstructSpdyGetSynReply(kExtraResponseHeaders, 1, 1));
+
+  scoped_ptr<SpdyFrame> body_frame(spdy_util_.ConstructSpdyBodyFrame(1, false));
+  // Last body frame has END_STREAM flag set.
+  scoped_ptr<SpdyFrame> last_body_frame(
+      spdy_util_.ConstructSpdyBodyFrame(1, true));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 1),       CreateMockRead(*body_frame, 2),
+      CreateMockRead(*body_frame, 3), CreateMockRead(*last_body_frame, 4),
+      MockRead(SYNCHRONOUS, 0, 5),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 443);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  BidirectionalStreamRequestInfo request_info;
+  request_info.method = "GET";
+  request_info.url = GURL("https://www.example.org/");
+  request_info.end_stream_on_headers = true;
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  delegate->Start(request_info, DEFAULT_PRIORITY, http_session_.get());
+  // Deliver two DATA frames together.
+  deterministic_data_->RunFor(4);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_TRUE(timer->IsRunning());
+  timer->Fire();
+  base::RunLoop().RunUntilIdle();
+  // This should trigger |more_read_data_pending_| to execute the task at a
+  // later time, and Delegate::OnReadComplete should not have been called.
+  EXPECT_TRUE(timer->IsRunning());
+  EXPECT_EQ(0, delegate->on_read_complete_count());
+
+  // Fire the timer now, the two DATA frame should be combined into one
+  // single Delegate::OnReadComplete callback.
+  timer->Fire();
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(1, delegate->on_read_complete_count());
+  EXPECT_EQ(kUploadDataSize * 2,
+            static_cast<int>(delegate->data_received().size()));
+
+  // Deliver last DATA frame and EOF. There will be an additional
+  // Delegate::OnReadComplete callback.
+  deterministic_data_->RunFor(2);
+  EXPECT_EQ(2, delegate->on_read_complete_count());
+  const SpdyHeaderBlock response_headers = delegate->response_headers();
+  EXPECT_EQ("200", response_headers.find(":status")->second);
+  EXPECT_EQ("header-value", response_headers.find("header-name")->second);
+
+  EXPECT_EQ(kUploadDataSize * 3,
+            static_cast<int>(delegate->data_received().size()));
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+  EXPECT_EQ(CountWriteBytes(writes, arraysize(writes)),
+            delegate->GetTotalSentBytes());
+  EXPECT_EQ(CountReadBytes(reads, arraysize(reads)),
+            delegate->GetTotalReceivedBytes());
+}
+
+TEST_F(BidirectionalStreamTest, TestBufferingWithTrailers) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet("https://www.example.org", false, 1, LOWEST));
+  // Empty DATA frame with an END_STREAM flag.
+  scoped_ptr<SpdyFrame> end_stream(
+      spdy_util_.ConstructSpdyBodyFrame(1, nullptr, 0, true));
+
+  MockWrite writes[] = {
+      CreateMockWrite(*req.get(), 0),
+  };
+
+  const char* const kExtraResponseHeaders[] = {"header-name", "header-value"};
+
+  scoped_ptr<SpdyFrame> resp(
+      spdy_util_.ConstructSpdyGetSynReply(kExtraResponseHeaders, 1, 1));
+
+  scoped_ptr<SpdyFrame> body_frame(spdy_util_.ConstructSpdyBodyFrame(1, false));
+
+  const char* const kTrailers[] = {"foo", "bar"};
+  scoped_ptr<SpdyFrame> trailers(
+      spdy_util_.ConstructSpdyHeaderFrame(1, kTrailers, 1, true));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 1),       CreateMockRead(*body_frame, 2),
+      CreateMockRead(*body_frame, 3), CreateMockRead(*body_frame, 4),
+      CreateMockRead(*trailers, 5),   MockRead(SYNCHRONOUS, 0, 6),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 443);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+
+  BidirectionalStreamRequestInfo request_info;
+  request_info.method = "GET";
+  request_info.url = GURL("https://www.example.org/");
+  request_info.end_stream_on_headers = true;
+
+  delegate->Start(request_info, DEFAULT_PRIORITY, http_session_.get());
+  // Deliver all three DATA frames together.
+  deterministic_data_->RunFor(5);

[mmenke, 2015/12/14 19:48:37]

Suggest using SequencedSocketData instead.  We want to get rid of
DeterministicSocketData, as SequencedSocketData is easier to use.  While RunFor
can let you probe objects at intermediate steps, which SequencedSocketData does
not currently support, the fact is that no one takes advantage of that.

[xunjieli, 2015/12/14 21:03:05]

I am aware of the deprecation plan for DeterministicSocketData. But I find it
very convenient to control how many frames to receive/send. Since we consolidate
several DATA frames into one async callback, if using Quit/Run alone, we can't
really test the buffering logic (e.g firing the timer when a particular data
frame is received).

[mmenke, 2015/12/14 21:49:38]

Hrm...  If we're going to need to be able to test this sort of thing, we may
need to add this feature to sequenced socket data.  It shouldn't be too hard. 
The reason for preferring it over DeterministicSocket data is the code is much
saner, failures are much easier to debug.  SequencedSocketData also forces be in
sequential order, with each number only occurring once, which
DeterministicSocket does not, so some very ...creative... tests can still manage
to pass.

[Ryan Hamilton, 2015/12/14 22:33:18]

I believe that SequencedSocketData already has a mechanism to do this. It has a
PAUSED state that it enters when, say, and ASYNC MockRead returns
ERR_IO_PENDING. This causes the state machine to basically stop until
CompleteRead() is called. I think it should be possible to use this mechanism in
place of DeterministicSocketData::RunFor().

With RunFor() the stopping point is decided by the argument to RunFor(). With
SequencedSocketData, the stopping point is determined by the
MockRead/MockWrites.

Does that sounds plausible?

+  base::RunLoop().RunUntilIdle();

[mmenke, 2015/12/14 19:48:37]

In my experience, RunUntilIdle makes for flaky, regression prone tests.  Ideally
you should be using RunLoop::Run() and then RunLoop::Quit when the even you care
about has happened.

[xunjieli, 2015/12/14 21:03:05]

I will try adjusting it accordingly. But I still have a question on
DeterministicSocketData, see above.

+  EXPECT_TRUE(timer->IsRunning());
+  timer->Fire();
+  base::RunLoop().RunUntilIdle();
+  // This should trigger |more_read_data_pending_| to execute the task at a
+  // later time, and Delegate::OnReadComplete should not have been called.
+  EXPECT_TRUE(timer->IsRunning());
+  EXPECT_EQ(0, delegate->on_read_complete_count());
+
+  // Deliver trailers. Remaining read should be completed, since OnClose is
+  // called right after OnTrailersReceived. The three DATA frames should be
+  // delivered in a single OnReadCompleted callback.
+  deterministic_data_->RunFor(2);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(timer->IsRunning());
+  EXPECT_EQ(1, delegate->on_read_complete_count());
+  EXPECT_EQ(kUploadDataSize * 3,
+            static_cast<int>(delegate->data_received().size()));
+  const SpdyHeaderBlock response_headers = delegate->response_headers();
+  EXPECT_EQ("200", response_headers.find(":status")->second);
+  EXPECT_EQ("header-value", response_headers.find("header-name")->second);
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+  EXPECT_EQ(CountWriteBytes(writes, arraysize(writes)),
+            delegate->GetTotalSentBytes());
+  EXPECT_EQ(CountReadBytes(reads, arraysize(reads)),
+            delegate->GetTotalReceivedBytes());
+}
+
+TEST_F(BidirectionalStreamTest, CancelStream) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet("https://www.example.org", false, 1, LOWEST));
+
+  scoped_ptr<SpdyFrame> rst(
+      spdy_util_.ConstructSpdyRstStream(1, RST_STREAM_CANCEL));
+  MockWrite writes[] = {
+      CreateMockWrite(*req, 0), CreateMockWrite(*rst, 2),
+  };
+
+  const char* const kExtraResponseHeaders[] = {"header-name", "header-value"};
+
+  scoped_ptr<SpdyFrame> resp(
+      spdy_util_.ConstructSpdyGetSynReply(kExtraResponseHeaders, 1, 1));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 1), MockRead(ASYNC, 0, 3),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 443);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSession(reads, arraysize(reads), writes, arraysize(writes), key);
+
+  BidirectionalStreamRequestInfo request_info;
+  request_info.method = "GET";
+  request_info.url = GURL("https://www.example.org/");
+  request_info.end_stream_on_headers = true;
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  scoped_ptr<CancelStreamDelegate> delegate(
+      new CancelStreamDelegate(read_buffer.get(), kReadBufferSize));
+  delegate->SetRunUntilCompletion(true);
+  delegate->Start(request_info, DEFAULT_PRIORITY, http_session_.get());
+  // Makes sure delegate does not get called.
+  base::RunLoop().RunUntilIdle();
+  const SpdyHeaderBlock response_headers = delegate->response_headers();
+  EXPECT_EQ("200", response_headers.find(":status")->second);
+  EXPECT_EQ("header-value", response_headers.find("header-name")->second);
+  EXPECT_EQ(0u, delegate->data_received().size());
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+
+  // Stream is canceled. The total received and sent bytes should be 0.
+  EXPECT_EQ(0, delegate->GetTotalSentBytes());
+  EXPECT_EQ(0, delegate->GetTotalReceivedBytes());
+}
+
+}  // namespace net