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 eabbed0bba902d2475b00a21b1aff6706dadd966..06a73380c32d424fe7ca4b07b15ff1c41bbe1842 100644
--- a/net/http/http_stream_factory_impl_job.cc
+++ b/net/http/http_stream_factory_impl_job.cc
@@ -26,6 +26,7 @@
 #include "net/base/net_util.h"
 #include "net/base/port_util.h"
 #include "net/cert/cert_verifier.h"
+#include "net/http/bidirectional_stream_job.h"

[mef, 2015/12/21 17:31:43]

Move into conditional include?

[xunjieli, 2015/12/21 22:01:22]

Done.

 #include "net/http/http_basic_stream.h"
 #include "net/http/http_network_session.h"
 #include "net/http/http_proxy_client_socket.h"
@@ -50,6 +51,10 @@
 #include "net/ssl/ssl_connection_status_flags.h"
 #include "net/ssl/ssl_failure_state.h"
 
+#if defined(ENABLE_BIDIRECTIONAL_STREAM)
+#include "net/spdy/bidirectional_stream_spdy_job.h"
+#endif
+
 namespace net {
 
 // Returns parameters associated with the start of a HTTP stream job.
@@ -137,6 +142,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 +174,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 +366,25 @@ void HttpStreamFactoryImpl::Job::OnWebSocketHandshakeStreamReadyCallback() {
   // |this| may be deleted after this call.
 }
 
+void HttpStreamFactoryImpl::Job::OnBidirectionalStreamJobReadyCallback() {
+  DCHECK(bidirectional_stream_job_);

[mef, 2015/12/21 17:31:42]

Should this and other sections be under #if
defined(ENABLE_BIDIRECTIONAL_STREAM)?

[xunjieli, 2015/12/21 22:01:22]

Done.

+
+  MaybeCopyConnectionAttemptsFromSocketOrHandle();
+
+  if (IsOrphaned()) {
+    stream_factory_->OnOrphanedJobComplete(this);
+  } else {
+    request_->Complete(was_npn_negotiated(), protocol_negotiated(),
+                       using_spdy());
+    request_->OnBidirectionalStreamJobReady(
+        this, server_ssl_config_, proxy_info_,
+        bidirectional_stream_job_.release());
+  }
+  // |this| may be deleted after this call.
+}
+
 void HttpStreamFactoryImpl::Job::OnNewSpdySessionReadyCallback() {
-  DCHECK(stream_.get());
+  DCHECK(stream_.get() || bidirectional_stream_job_.get());
   DCHECK(!IsPreconnecting());
   DCHECK(using_spdy());
   // Note: an event loop iteration has passed, so |new_spdy_session_| may be
@@ -379,8 +404,18 @@ void HttpStreamFactoryImpl::Job::OnNewSpdySessionReadyCallback() {
     }
     stream_factory_->OnOrphanedJobComplete(this);
   } else {
-    request_->OnNewSpdySessionReady(
-        this, stream_.Pass(), spdy_session, spdy_session_direct_);
+    if (for_bidirectional_) {
+      DCHECK(bidirectional_stream_job_);
+      request_->OnNewSpdySessionReady(this, /*spdy_http_stream=*/nullptr,
+                                      std::move(bidirectional_stream_job_),
+                                      spdy_session, spdy_session_direct_);
+
+    } else {
+      DCHECK(stream_);
+      request_->OnNewSpdySessionReady(this, std::move(stream_),
+                                      /** bidirectional_stream_job=*/nullptr,
+                                      spdy_session, spdy_session_direct_);
+    }
   }
   // |this| may be deleted after this call.
 }
@@ -567,6 +602,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_job_) {
+          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::OnBidirectionalStreamJobReadyCallback,
+                                    ptr_factory_.GetWeakPtr()));
+        }
       } else {
         DCHECK(stream_.get());
         base::ThreadTaskRunnerHandle::Get()->PostTask(
@@ -1144,12 +1189,23 @@ int HttpStreamFactoryImpl::Job::DoWaitingUserAction(int result) {
   return ERR_IO_PENDING;
 }
 
-int HttpStreamFactoryImpl::Job::SetSpdyHttpStream(
-    base::WeakPtr<SpdySession> session, bool direct) {
+int HttpStreamFactoryImpl::Job::SetSpdyHttpStreamOrBidirectionalStreamJob(
+    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_) {
+#if defined(ENABLE_BIDIRECTIONAL_STREAM)
+    // TODO(xunjieli): Create QUIC's version of BidirectionalStreamJob.
+    bidirectional_stream_job_.reset(new BidirectionalStreamSpdyJob(session));
+    return OK;
+#else
+    DCHECK(false);
+    return ERR_FAILED;
+#endif
+  }
 
   // TODO(willchan): Delete this code, because eventually, the
   // HttpStreamFactoryImpl will be creating all the SpdyHttpStreams, since it
@@ -1203,7 +1259,8 @@ int HttpStreamFactoryImpl::Job::DoCreateStream() {
       connection_->socket()->Disconnect();
     connection_->Reset();
 
-    int set_result = SetSpdyHttpStream(existing_spdy_session_, direct);
+    int set_result = SetSpdyHttpStreamOrBidirectionalStreamJob(
+        existing_spdy_session_, direct);
     existing_spdy_session_.reset();
     return set_result;
   }
@@ -1216,7 +1273,7 @@ int HttpStreamFactoryImpl::Job::DoCreateStream() {
     return result;
   }
   if (spdy_session) {
-    return SetSpdyHttpStream(spdy_session, direct);
+    return SetSpdyHttpStreamOrBidirectionalStreamJob(spdy_session, direct);
   }
 
   result = valid_spdy_session_pool_->CreateAvailableSessionFromSocket(
@@ -1251,12 +1308,13 @@ int HttpStreamFactoryImpl::Job::DoCreateStream() {
   if (http_server_properties)
     http_server_properties->SetSupportsSpdy(host_port_pair, true);
 
-  // Create a SpdyHttpStream attached to the session;
-  // OnNewSpdySessionReadyCallback is not called until an event loop
+  // Create a SpdyHttpStream or a BidirectionalStreamJob 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 SetSpdyHttpStreamOrBidirectionalStreamJob(new_spdy_session_,
+                                                   spdy_session_direct_);
 }
 
 int HttpStreamFactoryImpl::Job::DoCreateStreamComplete(int result) {