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

net/http/bidirectional_stream_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/http/bidirectional_stream.h"
+
+#include "base/macros.h"
+#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/base/test_data_directory.h"
+#include "net/http/http_network_session.h"
+#include "net/http/http_request_info.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 int kBodyDataSize = arraysize(kBodyData);
+const base::StringPiece kBodyDataStringPiece(kBodyData, kBodyDataSize);
+// Size of the buffer to be allocated for each read.
+const size_t kReadBufferSize = 4096;
+
+// Delegate that automatically reads data from the stream, but does not send
+// any data unless it is told to.
+class TestBidirectionalStreamDelegate : public BidirectionalStream::Delegate {
+ public:
+  TestBidirectionalStreamDelegate(IOBuffer* read_buf, int read_buf_len)
+      : bytes_read_(0),
+        error_(OK),
+        read_buf_(read_buf),
+        read_buf_len_(read_buf_len) {}
+
+  ~TestBidirectionalStreamDelegate() override {}
+
+  void OnRequestHeadersSent() override {}
+
+  void OnHeaders(const SpdyHeaderBlock& response_headers) override {
+    response_headers_ = response_headers;
+    StartOrContinueReading();
+  }
+
+  void OnReadCompleted(int bytes_read) override {

[mmenke, 2015/12/08 22:58:34]

Check we haven't received trailers yet?

[xunjieli, 2015/12/10 23:25:50]

Actually we aren't delaying OnTrailers event until read completes, so it could
be called before OnReadCompleted is called (for instance, if the embedder hasn't
called ReadData)

[mmenke, 2015/12/11 16:42:53]

That seems a bit weird.  Document it?  Or change the interface.  What does the
GRPC interface do?

[xunjieli, 2015/12/11 23:48:39]

gRPC does not require the underlying implemenation to delay OnTrailers after
read completes. I have added a comment.

+    CHECK_GT(bytes_read, OK);
+    bytes_read_ += bytes_read;
+    data_received_.append(read_buf_->data(), bytes_read_);
+    StartOrContinueReading();
+  }
+
+  void OnDataSent() override {}

[mmenke, 2015/12/08 22:58:34]

No tests at this layer send any data.  Add some?

[xunjieli, 2015/12/10 23:25:51]

There's one test, SendPostRequest, that sends data. I have more in
bidirectional_stream_spdy_job_unittest.cc

+
+  void OnTrailers(const SpdyHeaderBlock& trailers) override {

[mmenke, 2015/12/08 22:58:35]

Should we check for re-entrancy in the callbacks?

[xunjieli, 2015/12/10 23:25:51]

That sounds like something we should do. How should I check re-entrancy?

[mmenke, 2015/12/11 16:42:53]

I may have had something more clever in mind yesterday (I'm sure I did! 
Something awe-inspiringly brilliant, honest!), but you can just set a class
variable before calling into stream_, and clear it on return, and DCHECK on
that.

[xunjieli, 2015/12/11 23:48:39]

Done.

+    trailers_ = trailers;
+    loop_.Quit();
+  }
+
+  void OnFailed(int error) override {
+    CHECK_NE(OK, error);
+    error_ = error;
+    loop_.Quit();
+  }
+
+  void CreateBidirectionalStream(
+      const BidirectionalStream::RequestInfo& request_info,
+      RequestPriority priority,
+      HttpNetworkSession* session) {
+    stream_.reset(
+        new BidirectionalStream(request_info, priority, session, this));
+    loop_.Run();

[mmenke, 2015/12/08 22:58:35]

"CreateBidirectionalStream" does not indicate "Enters a message loop and spins
it until the response is complete".  Maybe call it RunTest or somesuch?  Or have
a separate method to start spinning the message loop.

[xunjieli, 2015/12/10 23:25:51]

Done.

+  }
+
+  void SendData(IOBuffer* data, int length, bool end_of_stream) {
+    stream_->SendData(data, length, end_of_stream);
+  }
+
+  // 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_; }
+
+ private:
+  // Starts or continues read data from |stream_| until there is no more byte
+  // can be read synchronously.

[mmenke, 2015/12/08 22:58:34]

nit:  Grammar.

[xunjieli, 2015/12/10 23:25:51]

Done.

+  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_);
+    }
+    if (rv == OK)

[mmenke, 2015/12/08 22:58:35]

No tests simulate async reads / writes, or multiple writes...Should they?

[xunjieli, 2015/12/10 23:25:51]

Those are in bidirectional_stream_spdy_job_unittest.cc, I don't know if I should
duplicate the test cases, I just want to make sure the wrapper works.

+      loop_.Quit();
+  }
+
+  SpdyHeaderBlock response_headers_;
+  SpdyHeaderBlock trailers_;
+  int bytes_read_;

[mmenke, 2015/12/08 22:58:35]

This is just data_received_.size(), right?  Can we just get rid of it?

[xunjieli, 2015/12/10 23:25:51]

Done.

+  int error_;
+  scoped_refptr<IOBuffer> read_buf_;
+  int read_buf_len_;
+  std::string data_received_;
+  base::RunLoop loop_;
+  scoped_ptr<BidirectionalStream> stream_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestBidirectionalStreamDelegate);
+};
+
+// A delegate that sends data after request headers are sent.
+class SendDataDelegate : public TestBidirectionalStreamDelegate {
+ public:
+  SendDataDelegate(IOBuffer* buf, int buf_len, base::StringPiece data)
+      : TestBidirectionalStreamDelegate(buf, buf_len), data_(data) {}
+
+  ~SendDataDelegate() override {}
+
+  void OnRequestHeadersSent() override {
+    TestBidirectionalStreamDelegate::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_;
+
+  DISALLOW_COPY_AND_ASSIGN(SendDataDelegate);
+};
+
+class BidirectionalStreamTest : public testing::Test {
+ public:
+  BidirectionalStreamTest()
+      : spdy_util_(kProtoHTTP2, false), session_deps_(kProtoHTTP2) {}
+
+ protected:
+  // 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_ = CreateSecureSpdySession(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_;
+};
+
+}  // namespace
+
+TEST_F(BidirectionalStreamTest, CreateInsecureStream) {
+  BidirectionalStream::RequestInfo request;
+  request.method = "GET";
+  request.url = GURL("http://www.example.org/");
+
+  TestBidirectionalStreamDelegate delegate(nullptr, 0);
+  HttpNetworkSession::Params params =
+      SpdySessionDependencies::CreateSessionParams(&session_deps_);
+  scoped_ptr<HttpNetworkSession> session(new HttpNetworkSession(params));
+  delegate.CreateBidirectionalStream(request, LOWEST, session.get());
+  EXPECT_EQ(ERR_DISALLOWED_URL_SCHEME, delegate.error());
+}
+
+TEST_F(BidirectionalStreamTest, SendPostRequest) {
+  spdy_util_.set_default_url(GURL("https://www.example.org"));
+  BufferedSpdyFramer framer(spdy_util_.spdy_version(), false);
+
+  scoped_ptr<SpdyFrame> req(spdy_util_.ConstructChunkedSpdyPost(nullptr, 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

[mmenke, 2015/12/08 22:58:34]

Should we check async reads?  Requests without trailers?  (Including sync and
async connection closes).

[xunjieli, 2015/12/10 23:25:51]

Those are in bidirectional_stream_spdy_job_unittest.cc. Not sure if I should
duplicate the test cases.

[mmenke, 2015/12/11 16:42:53]

Others may disagree, but I actually think if we're going to only do one or the
other (And I agree that doing both here is probably overkill), it's better to
test the interface used by external consumers directly, rather than the code
that backs it.

[xunjieli, 2015/12/11 23:48:39]

Done. I think your idea is better. If we add QUIC support, we can make the test
run a second time with QUIC, and we won't duplicate test code.

+  };
+
+  HostPortPair host_port_pair("www.example.org", 443);
+  SpdySessionKey key(host_port_pair, ProxyServer::Direct(),
+                     PRIVACY_MODE_DISABLED);
+
+  SSLSocketDataProvider ssl_data(ASYNC, OK);
+  ssl_data.SetNextProto(kProtoHTTP2);
+  ssl_data.cert = ImportCertFromFile(GetTestCertsDirectory(), "ok_cert.pem");
+  ASSERT_TRUE(ssl_data.cert.get());
+  session_deps_.socket_factory->AddSSLSocketDataProvider(&ssl_data);
+
+  InitSession(reads, arraysize(reads), writes, arraysize(writes), key);
+
+  BidirectionalStream::RequestInfo request;
+  request.method = "POST";
+  request.url = GURL("https://www.example.org/");
+
+  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
+  scoped_ptr<SendDataDelegate> delegate(new SendDataDelegate(
+      read_buffer.get(), kReadBufferSize, kBodyDataStringPiece));
+  delegate->CreateBidirectionalStream(request, DEFAULT_PRIORITY,
+                                      http_session_.get());
+
+  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(kBodyDataSize, delegate->bytes_read());
+  EXPECT_EQ(std::string(kBodyData, kBodyDataSize), delegate->data_received());
+  EXPECT_EQ("bar", delegate->trailers().find("foo")->second);
+}
+
+}  // namespace net