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/memory/scoped_ptr.h"
+#include "base/run_loop.h"
+#include "base/strings/string_piece.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);
+const base::StringPiece kBodyDataStringPiece(kBodyData, kBodyDataSize);
+// Size of the buffer to be allocated for each read.
+const size_t kReadBufferSize = 4096;
+const size_t kSmallReadBufferSize = 4;
+
+class TestBidirectionalStreamSpdyJobDelegate
+    : public BidirectionalStreamJob::Delegate {
+ public:
+  TestBidirectionalStreamSpdyJobDelegate(base::WeakPtr<SpdySession> session,
+                                         IOBuffer* read_buf,
+                                         int read_buf_len)
+      : bytes_read_(0),
+        close_status_(ERR_FAILED),
+        stream_(new BidirectionalStreamSpdyJob(session)),
+        loop_(new base::RunLoop),
+        read_buf_(read_buf),
+        read_buf_len_(read_buf_len) {}
+
+  ~TestBidirectionalStreamSpdyJobDelegate() override {}
+
+  void OnFailed(Error rv) override { loop_->Quit(); }
+
+  void OnRequestHeadersSent() override {}
+
+  void OnHeaders(const SpdyHeaderBlock& response_headers) override {
+    response_headers_ = response_headers;
+    StartOrContinueReading();
+  }
+
+  void OnReadCompleted(int bytes_read) override {
+    CHECK_GT(bytes_read, OK);
+    bytes_read_ += bytes_read;
+    data_received_.append(read_buf_->data(), bytes_read_);
+    StartOrContinueReading();
+  }
+
+  void OnDataSent() override {}
+
+  void OnTrailers(const SpdyHeaderBlock& trailers) override {
+    trailers_ = trailers;
+  }
+
+  void OnClose(int status) override {
+    close_status_ = status;
+    loop_->Quit();
+  }
+
+  void Start(const HttpRequestInfo& request,
+             RequestPriority priority,
+             const BoundNetLog& net_log) {
+    stream_->Start(request, priority, net_log, this);
+    loop_->Run();
+  }
+
+  void SendData(IOBuffer* data, int length, bool end_of_stream) {
+    stream_->SendData(data, length, end_of_stream);
+  }
+
+  const std::string& data_received() const { return data_received_; }
+
+  SpdyHeaderBlock response_headers_;
+  SpdyHeaderBlock trailers_;
+  int bytes_read_;
+  int close_status_;
+
+ protected:
+  scoped_ptr<BidirectionalStreamSpdyJob> stream_;
+  scoped_ptr<base::RunLoop> loop_;
+
+ private:
+  void StartOrContinueReading() {
+    int rv = stream_->ReadData(read_buf_.get(), read_buf_len_);
+    while (rv > 0) {
+      bytes_read_ += rv;
+      data_received_.append(read_buf_->data(), rv);
+      rv = stream_->ReadData(read_buf_.get(), read_buf_len_);
+    }
+  }
+
+  scoped_refptr<IOBuffer> read_buf_;
+  int read_buf_len_;
+  std::string data_received_;
+};
+
+// A delegate that sends data after request headers are sent.
+class SendDataDelegate : public TestBidirectionalStreamSpdyJobDelegate {
+ public:
+  SendDataDelegate(base::WeakPtr<SpdySession> session,
+                   IOBuffer* buf,
+                   int buf_len,
+                   base::StringPiece data)
+      : TestBidirectionalStreamSpdyJobDelegate(session, buf, buf_len),
+        data_(data) {}
+
+  ~SendDataDelegate() override {}
+
+  void OnRequestHeadersSent() override {
+    TestBidirectionalStreamSpdyJobDelegate::OnRequestHeadersSent();
+    if (data_.data()) {
+      scoped_refptr<StringIOBuffer> buf(new StringIOBuffer(data_.as_string()));
+      SendData(buf.get(), buf->size(), NO_MORE_DATA_TO_SEND);
+    }
+  }
+
+ private:
+  base::StringPiece data_;
+};
+
+// A delegate that cancels the request after response headers are received.
+class CancelStreamDelegate : public TestBidirectionalStreamSpdyJobDelegate {
+ public:
+  CancelStreamDelegate(base::WeakPtr<SpdySession> session,
+                       IOBuffer* buf,
+                       int buf_len)
+      : TestBidirectionalStreamSpdyJobDelegate(session, buf, buf_len) {}
+
+  ~CancelStreamDelegate() override {}
+
+  void OnHeaders(const SpdyHeaderBlock& response_headers) override {
+    TestBidirectionalStreamSpdyJobDelegate::OnHeaders(response_headers);
+    stream_->Cancel();
+    loop_->Quit();
+  }
+
+  void OnDataSent() override { ASSERT_TRUE(false); }
+
+  void OnReadCompleted(int bytes_read) override { ASSERT_TRUE(false); }
+
+  void OnTrailers(const SpdyHeaderBlock& trailers) override {
+    ASSERT_TRUE(false);
+  }
+
+  void OnClose(int error) override { ASSERT_TRUE(false); }
+};
+
+}  // namespace
+
+class BidirectionalStreamSpdyJobTest : public testing::Test {
+ public:
+  BidirectionalStreamSpdyJobTest()
+      : spdy_util_(kProtoHTTP2), session_deps_(kProtoHTTP2) {}
+
+ protected:
+  void TearDown() override {
+    EXPECT_TRUE(sequenced_data_->AllReadDataConsumed());
+    EXPECT_TRUE(sequenced_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());
+  }
+
+  SpdyTestUtil spdy_util_;
+  SpdySessionDependencies session_deps_;
+  scoped_ptr<SequencedSocketData> sequenced_data_;
+  scoped_ptr<HttpNetworkSession> http_session_;
+  base::WeakPtr<SpdySession> session_;
+};
+
+TEST_F(BidirectionalStreamSpdyJobTest, SendGetRequest) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(NULL, 0, false, 1, LOWEST, 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(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, 2),
+      CreateMockRead(*trailers, 3), MockRead(SYNCHRONOUS, 0, 4),
+  };
+
+  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 = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  scoped_ptr<TestBidirectionalStreamSpdyJobDelegate> delegate(
+      new TestBidirectionalStreamSpdyJobDelegate(session_, read_buffer.get(),
+                                                 kReadBufferSize));
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  base::StringPiece status = delegate->response_headers_[":status"];
+  EXPECT_EQ("200", status);
+  base::StringPiece extra_header = delegate->response_headers_["header-name"];
+  EXPECT_EQ("header-value", extra_header);
+  EXPECT_EQ(kUploadDataSize, delegate->bytes_read_);
+  EXPECT_EQ(kUploadData, delegate->data_received());
+  base::StringPiece trailer = delegate->trailers_["foo"];
+  EXPECT_EQ("bar", trailer);
+  EXPECT_EQ(OK, delegate->close_status_);
+}
+
+TEST_F(BidirectionalStreamSpdyJobTest, SendGetRequestSmallReadBuffer) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(NULL, 0, false, 1, LOWEST, 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(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, 2),
+      CreateMockRead(*trailers, 3), MockRead(SYNCHRONOUS, 0, 4),
+  };
+
+  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 = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kSmallReadBufferSize));
+  scoped_ptr<TestBidirectionalStreamSpdyJobDelegate> delegate(
+      new TestBidirectionalStreamSpdyJobDelegate(session_, read_buffer.get(),
+                                                 kSmallReadBufferSize));
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  base::StringPiece status = delegate->response_headers_[":status"];
+  EXPECT_EQ("200", status);
+  base::StringPiece extra_header = delegate->response_headers_["header-name"];
+  EXPECT_EQ("header-value", extra_header);
+  EXPECT_EQ(kUploadDataSize, delegate->bytes_read_);
+  EXPECT_EQ(kUploadData, delegate->data_received());
+  base::StringPiece trailer = delegate->trailers_["foo"];
+  EXPECT_EQ("bar", trailer);
+  EXPECT_EQ(OK, delegate->close_status_);
+}
+
+TEST_F(BidirectionalStreamSpdyJobTest, SendGetRequestNoTrailers) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(NULL, 0, false, 1, LOWEST, 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(spdy_util_.ConstructSpdyBodyFrame(1, true));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 1), CreateMockRead(*body, 2),
+      MockRead(SYNCHRONOUS, 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 = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  scoped_ptr<TestBidirectionalStreamSpdyJobDelegate> delegate(
+      new TestBidirectionalStreamSpdyJobDelegate(session_, read_buffer.get(),
+                                                 kReadBufferSize));
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  base::StringPiece status = delegate->response_headers_[":status"];
+  EXPECT_EQ("200", status);
+  base::StringPiece extra_header = delegate->response_headers_["header-name"];
+  EXPECT_EQ("header-value", extra_header);
+  EXPECT_EQ(kUploadDataSize, delegate->bytes_read_);
+  EXPECT_EQ(kUploadData, delegate->data_received());
+  EXPECT_EQ(OK, delegate->close_status_);
+}
+
+TEST_F(BidirectionalStreamSpdyJobTest, CancelStream) {
+  scoped_ptr<SpdyFrame> req(
+      spdy_util_.ConstructSpdyGet(NULL, 0, false, 1, LOWEST, true));
+  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 = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  scoped_ptr<TestBidirectionalStreamSpdyJobDelegate> delegate(
+      new CancelStreamDelegate(session_, read_buffer.get(), kReadBufferSize));
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+  // Makes sure delegate does not get called.
+  base::RunLoop().RunUntilIdle();
+  base::StringPiece status = delegate->response_headers_[":status"];
+  EXPECT_EQ("200", status);
+  base::StringPiece extra_header = delegate->response_headers_["header-name"];
+  EXPECT_EQ("header-value", extra_header);
+  EXPECT_EQ(0, delegate->bytes_read_);
+}
+
+TEST_F(BidirectionalStreamSpdyJobTest, SendPostRequest) {
+  BufferedSpdyFramer framer(spdy_util_.spdy_version(), false);
+
+  scoped_ptr<SpdyFrame> req(spdy_util_.ConstructChunkedSpdyPost(NULL, 0));
+  scoped_ptr<SpdyFrame> body(
+      framer.CreateDataFrame(1, kBodyData, kBodyDataSize, DATA_FLAG_FIN));
+  MockWrite writes[] = {
+      CreateMockWrite(*req, 0),   // request
+      CreateMockWrite(*body, 1),  // POST upload frame
+  };
+  const char* const kExtraResponseHeaders[] = {"header-name", "header-value"};
+  scoped_ptr<SpdyFrame> resp(
+      spdy_util_.ConstructSpdyPostSynReply(kExtraResponseHeaders, 1));
+
+  scoped_ptr<SpdyFrame> resp_data(
+      framer.CreateDataFrame(1, kBodyData, kBodyDataSize, DATA_FLAG_NONE));
+
+  const char* const kExtraHeaders[] = {"foo", "bar"};
+  scoped_ptr<SpdyFrame> trailers(
+      spdy_util_.ConstructSpdyHeaderFrame(1, kExtraHeaders, 1, true));
+
+  MockRead reads[] = {
+      CreateMockRead(*resp, 2), CreateMockRead(*resp_data, 3),
+      CreateMockRead(*trailers, 4), MockRead(SYNCHRONOUS, 0, 5)  // EOF
+  };
+
+  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<TestBidirectionalStreamSpdyJobDelegate> delegate(
+      new SendDataDelegate(session_, read_buffer.get(), kReadBufferSize,
+                           kBodyDataStringPiece));
+  BoundNetLog net_log;
+  delegate->Start(request, DEFAULT_PRIORITY, net_log);
+
+  base::StringPiece header = delegate->response_headers_[":status"];
+  EXPECT_EQ("200", header);
+  base::StringPiece extra_header = delegate->response_headers_["header-name"];
+  EXPECT_EQ("header-value", extra_header);
+  EXPECT_EQ((int)kBodyDataSize, delegate->bytes_read_);
+  EXPECT_EQ(std::string(kBodyData, kBodyDataSize), delegate->data_received());
+  base::StringPiece trailer = delegate->trailers_["foo"];
+  EXPECT_EQ("bar", trailer);
+  EXPECT_EQ(OK, delegate->close_status_);
+}
+
+}  // namespace net