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 c3a6073336d53675c50728799df21ba2aa3f9656..de505fca00d9d710796988d4ba2763ec96f1bfed 100644
--- a/net/http/http_stream_factory_impl_job.cc
+++ b/net/http/http_stream_factory_impl_job.cc
@@ -25,6 +25,7 @@
 #include "net/base/net_util.h"
 #include "net/base/port_util.h"
 #include "net/cert/cert_verifier.h"
+#include "net/http/bidirectional_stream.h"
 #include "net/http/http_basic_stream.h"
 #include "net/http/http_network_session.h"
 #include "net/http/http_proxy_client_socket.h"
@@ -134,6 +135,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);
@@ -165,6 +167,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();
 }
 
@@ -355,8 +359,26 @@ void HttpStreamFactoryImpl::Job::OnWebSocketHandshakeStreamReadyCallback() {
   // |this| may be deleted after this call.
 }
 
+void HttpStreamFactoryImpl::Job::OnBidirectionalStreamReadyCallback() {
+  DCHECK(bidirectional_stream_);
+  DCHECK(!IsPreconnecting());
+  DCHECK(!stream_factory_->for_websockets_);
+
+  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
@@ -376,8 +398,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, nullptr /** spdy_http_stream */,

[Bence, 2015/10/01 05:29:28]

I think the syntax most often used is
/*spdy_http_stream=*/ nullptr

[xunjieli, 2015/10/01 18:41:16]

Done.

+                                      bidirectional_stream_.Pass(),
+                                      spdy_session, spdy_session_direct_);
+
+    } else {
+      request_->OnNewSpdySessionReady(this, stream_.Pass(),
+                                      nullptr /** bidirectional_stream */,

[Bence, 2015/10/01 05:29:28]

/*bidirectional_stream=*/ nullptr

[xunjieli, 2015/10/01 18:41:16]

Done.

+                                      spdy_session, spdy_session_direct_);
+    }
   }
   // |this| may be deleted after this call.
 }
@@ -564,6 +594,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) {
+          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(
@@ -1148,6 +1188,10 @@ int HttpStreamFactoryImpl::Job::SetSpdyHttpStream(
   // implemented.
   if (stream_factory_->for_websockets_)
     return ERR_NOT_IMPLEMENTED;
+  if (for_bidirectional_) {
+    bidirectional_stream_.reset(new BidirectionalStream(session));
+    return OK;
+  }
 
   // TODO(willchan): Delete this code, because eventually, the
   // HttpStreamFactoryImpl will be creating all the SpdyHttpStreams, since it