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

net/http/http_stream_factory_impl_job.cc

Index: net/http/http_stream_factory_impl_job.cc
diff --git a/net/http/http_stream_factory_impl_job.cc b/net/http/http_stream_factory_impl_job.cc
index 3625e4a4ef9ec69d6ef2fa518835fc3d91c1b243..a075765a23e88f8d7de39f78c35bd98ddd4c11de 100644
--- a/net/http/http_stream_factory_impl_job.cc
+++ b/net/http/http_stream_factory_impl_job.cc
@@ -42,6 +42,7 @@
 #include "net/socket/socks_client_socket_pool.h"
 #include "net/socket/ssl_client_socket.h"
 #include "net/socket/ssl_client_socket_pool.h"
+#include "net/spdy/bidirectional_spdy_stream.h"
 #include "net/spdy/spdy_http_stream.h"
 #include "net/spdy/spdy_protocol.h"
 #include "net/spdy/spdy_session.h"
@@ -137,6 +138,7 @@ HttpStreamFactoryImpl::Job::Job(HttpStreamFactoryImpl* stream_factory,
       spdy_session_direct_(false),
       job_status_(STATUS_RUNNING),
       other_job_status_(STATUS_RUNNING),
+      for_bidirectional_(false),
       ptr_factory_(this) {
   DCHECK(stream_factory);
   DCHECK(session);
@@ -168,6 +170,8 @@ HttpStreamFactoryImpl::Job::~Job() {
 void HttpStreamFactoryImpl::Job::Start(Request* request) {
   DCHECK(request);
   request_ = request;
+  // Saves |for_bidirectional_|, since request is nulled when job is orphaned.
+  for_bidirectional_ = request_->for_bidirectional();
   StartInternal();
 }
 
@@ -358,8 +362,24 @@ void HttpStreamFactoryImpl::Job::OnWebSocketHandshakeStreamReadyCallback() {
   // |this| may be deleted after this call.
 }
 
+void HttpStreamFactoryImpl::Job::OnBidirectionalStreamReadyCallback() {
+  DCHECK(bidirectional_stream_);
+
+  MaybeCopyConnectionAttemptsFromSocketOrHandle();
+
+  if (IsOrphaned()) {
+    stream_factory_->OnOrphanedJobComplete(this);
+  } else {
+    request_->Complete(was_npn_negotiated(), protocol_negotiated(),
+                       using_spdy());
+    request_->OnBidirectionalStreamReady(this, server_ssl_config_, proxy_info_,
+                                         bidirectional_stream_.release());
+  }
+  // |this| may be deleted after this call.
+}
+
 void HttpStreamFactoryImpl::Job::OnNewSpdySessionReadyCallback() {
-  DCHECK(stream_.get());
+  DCHECK(stream_.get() || bidirectional_stream_.get());
   DCHECK(!IsPreconnecting());
   DCHECK(using_spdy());
   // Note: an event loop iteration has passed, so |new_spdy_session_| may be
@@ -379,8 +399,16 @@ void HttpStreamFactoryImpl::Job::OnNewSpdySessionReadyCallback() {
     }
     stream_factory_->OnOrphanedJobComplete(this);
   } else {
-    request_->OnNewSpdySessionReady(
-        this, stream_.Pass(), spdy_session, spdy_session_direct_);
+    if (for_bidirectional_) {
+      request_->OnNewSpdySessionReady(this, /*spdy_http_stream=*/nullptr,
+                                      bidirectional_stream_.Pass(),
+                                      spdy_session, spdy_session_direct_);
+
+    } else {
+      request_->OnNewSpdySessionReady(this, stream_.Pass(),
+                                      /** bidirectional_stream=*/nullptr,

[mef, 2015/10/20 21:56:35]

I wonder what would happen if you call Pass() on scoped_ptr that holds null?

[xunjieli, 2015/10/21 19:35:36]

I am not sure. I guess that will give nulls? I will add DCHECKs here.

+                                      spdy_session, spdy_session_direct_);
+    }
   }
   // |this| may be deleted after this call.
 }
@@ -567,6 +595,16 @@ int HttpStreamFactoryImpl::Job::RunLoop(int result) {
         base::ThreadTaskRunnerHandle::Get()->PostTask(
             FROM_HERE, base::Bind(&Job::OnWebSocketHandshakeStreamReadyCallback,
                                   ptr_factory_.GetWeakPtr()));
+      } else if (for_bidirectional_) {
+        if (bidirectional_stream_ == nullptr) {

[mef, 2015/10/20 21:56:35]

why do you need this guard? Shouldn't it go to default: section for errors?

[xunjieli, 2015/10/21 19:35:36]

If H2 is not negotiated, we will still enter the OK case, so we need an extra
check here to make sure we fail if H2 is not negotiated and bidi stream is
requested.

+          base::ThreadTaskRunnerHandle::Get()->PostTask(
+              FROM_HERE, base::Bind(&Job::OnStreamFailedCallback,
+                                    ptr_factory_.GetWeakPtr(), ERR_FAILED));
+        } else {
+          base::ThreadTaskRunnerHandle::Get()->PostTask(
+              FROM_HERE, base::Bind(&Job::OnBidirectionalStreamReadyCallback,
+                                    ptr_factory_.GetWeakPtr()));
+        }
       } else {
         DCHECK(stream_.get());
         base::ThreadTaskRunnerHandle::Get()->PostTask(
@@ -653,6 +691,7 @@ int HttpStreamFactoryImpl::Job::DoLoop(int result) {
 
 int HttpStreamFactoryImpl::Job::StartInternal() {
   CHECK_EQ(STATE_NONE, next_state_);
+

[mef, 2015/10/20 21:56:35]

spurious nl?

[xunjieli, 2015/10/21 19:35:36]

Done.

   next_state_ = STATE_START;
   int rv = RunLoop(OK);
   DCHECK_EQ(ERR_IO_PENDING, rv);
@@ -1145,12 +1184,18 @@ int HttpStreamFactoryImpl::Job::DoWaitingUserAction(int result) {
   return ERR_IO_PENDING;
 }
 
-int HttpStreamFactoryImpl::Job::SetSpdyHttpStream(
-    base::WeakPtr<SpdySession> session, bool direct) {
+int HttpStreamFactoryImpl::Job::SetSpdyHttpStreamOrBidirectionalStream(
+    base::WeakPtr<SpdySession> session,
+    bool direct) {
   // TODO(ricea): Restore the code for WebSockets over SPDY once it's
   // implemented.
   if (stream_factory_->for_websockets_)
     return ERR_NOT_IMPLEMENTED;
+  if (for_bidirectional_) {
+    // TODO(xunjieli): Enable creation of QUIC's version of BidirectionalStream.
+    bidirectional_stream_.reset(new BidirectionalSpdyStream(session));
+    return OK;
+  }
 
   // TODO(willchan): Delete this code, because eventually, the
   // HttpStreamFactoryImpl will be creating all the SpdyHttpStreams, since it
@@ -1204,7 +1249,8 @@ int HttpStreamFactoryImpl::Job::DoCreateStream() {
       connection_->socket()->Disconnect();
     connection_->Reset();
 
-    int set_result = SetSpdyHttpStream(existing_spdy_session_, direct);
+    int set_result =
+        SetSpdyHttpStreamOrBidirectionalStream(existing_spdy_session_, direct);
     existing_spdy_session_.reset();
     return set_result;
   }
@@ -1217,7 +1263,7 @@ int HttpStreamFactoryImpl::Job::DoCreateStream() {
     return result;
   }
   if (spdy_session) {
-    return SetSpdyHttpStream(spdy_session, direct);
+    return SetSpdyHttpStreamOrBidirectionalStream(spdy_session, direct);
   }
 
   result = valid_spdy_session_pool_->CreateAvailableSessionFromSocket(
@@ -1252,12 +1298,13 @@ int HttpStreamFactoryImpl::Job::DoCreateStream() {
   if (http_server_properties)
     http_server_properties->SetSupportsSpdy(host_port_pair, true);
 
-  // Create a SpdyHttpStream attached to the session;
+  // Create a SpdyHttpStream or a BidirectionalStream attached to the session;
   // OnNewSpdySessionReadyCallback is not called until an event loop
   // iteration later, so if the SpdySession is closed between then, allow
   // reuse state from the underlying socket, sampled by SpdyHttpStream,
   // bubble up to the request.
-  return SetSpdyHttpStream(new_spdy_session_, spdy_session_direct_);
+  return SetSpdyHttpStreamOrBidirectionalStream(new_spdy_session_,
+                                                spdy_session_direct_);
 }
 
 int HttpStreamFactoryImpl::Job::DoCreateStreamComplete(int result) {