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

net/spdy/bidirectional_stream_spdy_job_unittest.cc

+// 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/spdy/bidirectional_stream_spdy_job.h"
+
+#include <string>
+
+#include "base/macros.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/run_loop.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/http/http_request_info.h"
+#include "net/http/http_response_headers.h"
+#include "net/http/http_response_info.h"
+#include "net/socket/socket_test_util.h"
+#include "net/spdy/spdy_read_queue.h"
+#include "net/spdy/spdy_session.h"
+#include "net/spdy/spdy_test_util_common.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 BidirectionalStreamJob::Delegate {
+ public:
+  TestDelegateBase(base::WeakPtr<SpdySession> session,
+                   IOBuffer* read_buf,
+                   int read_buf_len)
+      : TestDelegateBase(session,
+                         read_buf,
+                         read_buf_len,
+                         make_scoped_ptr(new base::Timer(false, false))) {}
+
+  TestDelegateBase(base::WeakPtr<SpdySession> session,
+                   IOBuffer* read_buf,
+                   int read_buf_len,
+                   scoped_ptr<base::Timer> timer)
+      : stream_job_(new BidirectionalStreamSpdyJob(session, timer.Pass())),
+        read_buf_(read_buf),
+        read_buf_len_(read_buf_len),
+        loop_(nullptr),
+        error_(OK),
+        bytes_read_(0),
+        on_read_complete_count_(0),
+        on_data_sent_count_(0),
+        do_not_start_read_(false),
+        run_until_completion_(false) {}
+
+  ~TestDelegateBase() override {}
+
+  void OnRequestHeadersSent() override {}
+
+  void OnHeaders(const SpdyHeaderBlock& response_headers) override {
+    response_headers_ = response_headers;
+    if (!do_not_start_read_)
+      StartOrContinueReading();
+  }
+
+  void OnReadCompleted(int bytes_read) override {
+    on_read_complete_count_++;
+    CHECK_GE(bytes_read, OK);
+    bytes_read_ += bytes_read;
+    data_received_.append(read_buf_->data(), bytes_read);
+    if (!do_not_start_read_)
+      StartOrContinueReading();
+  }
+
+  void OnDataSent() override { on_data_sent_count_++; }
+
+  void OnTrailers(const SpdyHeaderBlock& trailers) override {
+    trailers_ = trailers;
+    if (run_until_completion_)
+      loop_->Quit();
+  }
+
+  void OnFailed(int error) override {
+    CHECK_NE(OK, error);
+    error_ = error;
+    if (run_until_completion_)
+      loop_->Quit();
+  }
+
+  void Start(const HttpRequestInfo& request,
+             RequestPriority priority,
+             const BoundNetLog& net_log) {
+    stream_job_->Start(request, priority, net_log, this);
+    if (run_until_completion_)
+      loop_->Run();
+  }
+
+  void SendData(IOBuffer* data, int length, bool end_of_stream) {
+    stream_job_->SendData(data, length, end_of_stream);
+  }
+
+  // Starts or continues read data from |stream_job_| until there is no more
+  // byte 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_job_| and updates internal states.
+  int ReadData() {
+    int rv = stream_job_->ReadData(read_buf_.get(), read_buf_len_);
+    if (rv > 0) {
+      data_received_.append(read_buf_->data(), rv);
+      bytes_read_ += rv;
+    }
+    return rv;
+  }
+
+  NextProto GetProtocol() const { return stream_job_->GetProtocol(); }
+
+  int64_t GetTotalReceivedBytes() const {
+    return stream_job_->GetTotalReceivedBytes();
+  }
+
+  int64_t GetTotalSentBytes() const { return stream_job_->GetTotalSentBytes(); }
+
+  // Const getters for internal states.
+  const std::string& data_received() const { return data_received_; }
+  int bytes_read() const { return bytes_read_; }
+  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_job_|.
+  void CancelStream() { stream_job_->Cancel(); }
+  // Quits |loop_|.
+  void QuitLoop() { loop_->Quit(); }
+
+ private:
+  scoped_ptr<BidirectionalStreamSpdyJob> stream_job_;
+  scoped_refptr<IOBuffer> read_buf_;
+  int read_buf_len_;
+  std::string data_received_;
+  scoped_ptr<base::RunLoop> loop_;
+  SpdyHeaderBlock response_headers_;
+  SpdyHeaderBlock trailers_;
+  int error_;
+  int bytes_read_;
+  int on_read_complete_count_;
+  int on_data_sent_count_;
+  bool do_not_start_read_;
+  bool run_until_completion_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestDelegateBase);
+};
+
+// A delegate that cancels the request after response headers are received.
+class CancelStreamDelegate : public TestDelegateBase {
+ public:
+  CancelStreamDelegate(base::WeakPtr<SpdySession> session,
+                       IOBuffer* buf,
+                       int buf_len)
+      : TestDelegateBase(session, buf, buf_len) {}
+
+  ~CancelStreamDelegate() override {}
+
+  void OnHeaders(const SpdyHeaderBlock& response_headers) override {
+    TestDelegateBase::OnHeaders(response_headers);
+    CancelStream();
+    QuitLoop();
+  }
+
+  void OnDataSent() override { NOTREACHED(); }
+
+  void OnReadCompleted(int bytes_read) override { NOTREACHED(); }
+
+  void OnTrailers(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 BidirectionalStreamSpdyJobTest : public testing::Test {
+ public:
+  BidirectionalStreamSpdyJobTest()
+      : spdy_util_(kProtoHTTP2, false), session_deps_(kProtoHTTP2) {}
+
+ 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) {
+    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_ =
+        CreateInsecureSpdySession(http_session_.get(), key, BoundNetLog());
+  }
+
+  // Initializes the session using DeterministicSocketData.
+  void InitSessionDeterministic(MockRead* reads,
+                                size_t reads_count,
+                                MockWrite* writes,
+                                size_t writes_count,
+                                const SpdySessionKey& key) {
+    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_ =
+        CreateInsecureSpdySession(http_session_.get(), key, BoundNetLog());
+  }
+
+  SpdyTestUtil spdy_util_;
+  SpdySessionDependencies session_deps_;
+  scoped_ptr<SequencedSocketData> sequenced_data_;
+  scoped_ptr<DeterministicSocketData> deterministic_data_;
+  scoped_ptr<HttpNetworkSession> http_session_;
+  base::WeakPtr<SpdySession> session_;
+};
+
+// Simulates user calling ReadData after END_STREAM has been received in
+// BidirectionalStreamSpdyJob.
+TEST_F(BidirectionalStreamSpdyJobTest, TestReadDataAfterClose) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(nullptr, 0, false, 1, LOWEST, true, false));
+  // 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), CreateMockWrite(*end_stream, 1),
+  };
+
+  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, 2),       CreateMockRead(*body_frame, 3),
+      CreateMockRead(*body_frame, 4), CreateMockRead(*last_body_frame, 5),
+      MockRead(SYNCHRONOUS, 0, 6),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 80);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  HttpRequestInfo request;
+  request.method = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      session_, read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  delegate->set_do_not_start_read(true);
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  // Write request.
+  deterministic_data_->RunFor(1);
+  // Write END_STREAM.
+  scoped_refptr<net::IOBuffer> empty_request_buffer = new net::IOBuffer(0);
+  delegate->SendData(empty_request_buffer.get(), 0, true);
+  // 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, delegate->bytes_read());
+
+  // Deliver the rest. Note that user has not called a second ReadData.
+  deterministic_data_->RunFor(4);
+  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(BidirectionalStreamSpdyJobTest, TestInterleaveReadDataAndSendData) {
+  BufferedSpdyFramer framer(spdy_util_.spdy_version(), false);
+
+  scoped_ptr<SpdyFrame> req(spdy_util_.ConstructChunkedSpdyPost(nullptr, 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", 80);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  HttpRequestInfo request;
+  request.method = "POST";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      session_, read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  delegate->set_do_not_start_read(true);
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  // 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(BidirectionalStreamSpdyJobTest, TestCompleteAsyncRead) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(nullptr, 0, false, 1, LOWEST, true, false));
+  // 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), CreateMockWrite(*end_stream, 1),
+  };
+
+  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, 2), CreateMockRead(*response_body_frame, 3),
+      MockRead(SYNCHRONOUS, 0, 4),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 80);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  HttpRequestInfo request;
+  request.method = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      session_, read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  delegate->set_do_not_start_read(true);
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  // Write request.
+  deterministic_data_->RunFor(1);
+  // Write END_STREAM.
+  scoped_refptr<net::IOBuffer> empty_request_buffer = new net::IOBuffer(0);
+  delegate->SendData(empty_request_buffer.get(), 0, true);
+  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(0, delegate->bytes_read());
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+  EXPECT_EQ(CountWriteBytes(writes, arraysize(writes)),
+            delegate->GetTotalSentBytes());
+  EXPECT_EQ(CountReadBytes(reads, arraysize(reads)),
+            delegate->GetTotalReceivedBytes());
+}
+
+TEST_F(BidirectionalStreamSpdyJobTest, TestBuffering) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(nullptr, 0, false, 1, LOWEST, true, false));
+  // 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), CreateMockWrite(*end_stream, 1),
+  };
+
+  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, 2),       CreateMockRead(*body_frame, 3),
+      CreateMockRead(*body_frame, 4), CreateMockRead(*last_body_frame, 5),
+      MockRead(SYNCHRONOUS, 0, 6),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 80);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  HttpRequestInfo request;
+  request.method = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      session_, read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  // Write request.
+  deterministic_data_->RunFor(1);
+  // Write END_STREAM.
+  scoped_refptr<net::IOBuffer> empty_request_buffer = new net::IOBuffer(0);
+  delegate->SendData(empty_request_buffer.get(), 0, true);
+  // 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, delegate->bytes_read());
+
+  // 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, delegate->bytes_read());
+
+  EXPECT_EQ(kProtoHTTP2, delegate->GetProtocol());
+  EXPECT_EQ(CountWriteBytes(writes, arraysize(writes)),
+            delegate->GetTotalSentBytes());
+  EXPECT_EQ(CountReadBytes(reads, arraysize(reads)),
+            delegate->GetTotalReceivedBytes());
+}
+
+TEST_F(BidirectionalStreamSpdyJobTest, TestBufferingWithTrailers) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(nullptr, 0, false, 1, LOWEST, true, false));
+  // 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), CreateMockWrite(*end_stream, 1),
+  };
+
+  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, 2),       CreateMockRead(*body_frame, 3),
+      CreateMockRead(*body_frame, 4), CreateMockRead(*body_frame, 5),
+      CreateMockRead(*trailers, 6),   MockRead(SYNCHRONOUS, 0, 7),
+  };
+
+  HostPortPair host_port_pair("www.example.org", 80);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSessionDeterministic(reads, arraysize(reads), writes, arraysize(writes),
+                           key);
+
+  HttpRequestInfo request;
+  request.method = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  MockTimer* timer = new MockTimer();
+  scoped_ptr<TestDelegateBase> delegate(new TestDelegateBase(
+      session_, read_buffer.get(), kReadBufferSize, make_scoped_ptr(timer)));
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  // Write request.
+  deterministic_data_->RunFor(1);
+  // Write END_STREAM.
+  scoped_refptr<net::IOBuffer> empty_request_buffer = new net::IOBuffer(0);
+  delegate->SendData(empty_request_buffer.get(), 0, true);
+  // Deliver all three DATA frames together.
+  deterministic_data_->RunFor(5);
+  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());
+
+  // Deliver trailers. Remaining read should be completed, since OnClose is
+  // called right after OnTrailers. 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, delegate->bytes_read());
+  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(BidirectionalStreamSpdyJobTest, CancelStream) {
+  scoped_ptr<SpdyFrame> req(spdy_util_.ConstructChunkedSpdyPost(nullptr, 0));
+
+  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", 80);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+  InitSession(reads, arraysize(reads), writes, arraysize(writes), key);
+
+  HttpRequestInfo request;
+  request.method = "POST";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  scoped_ptr<CancelStreamDelegate> delegate(
+      new CancelStreamDelegate(session_, read_buffer.get(), kReadBufferSize));
+  delegate->SetRunUntilCompletion(true);
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  // 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(0, delegate->bytes_read());
+
+  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