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

net/http/http_stream_factory_impl_job.cc

 // Copyright (c) 2012 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/http_stream_factory_impl_job.h"
 
 #include <algorithm>
 #include <string>
 
 #include "base/bind.h"
@@ skipping 24 matching lines @@
 #include "net/http/http_stream_factory.h"
 #include "net/http/http_stream_factory_impl_request.h"
 #include "net/log/net_log.h"
 #include "net/quic/quic_http_stream.h"
 #include "net/socket/client_socket_handle.h"
 #include "net/socket/client_socket_pool.h"
 #include "net/socket/client_socket_pool_manager.h"
 #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"
 #include "net/spdy/spdy_session_pool.h"
 #include "net/ssl/ssl_cert_request_info.h"
 #include "net/ssl/ssl_connection_status_flags.h"
 #include "net/ssl/ssl_failure_state.h"
 
 namespace net {
 
@@ skipping 75 matching lines @@
       quic_request_(session_->quic_stream_factory()),
       using_existing_quic_session_(false),
       spdy_certificate_error_(OK),
       establishing_tunnel_(false),
       was_npn_negotiated_(false),
       protocol_negotiated_(kProtoUnknown),
       num_streams_(0),
       spdy_session_direct_(false),
       job_status_(STATUS_RUNNING),
       other_job_status_(STATUS_RUNNING),
+      for_bidirectional_(false),
       ptr_factory_(this) {
   DCHECK(stream_factory);
   DCHECK(session);
   if (IsQuicAlternative()) {
     DCHECK(session_->params().enable_quic);
     using_quic_ = true;
   }
 }
 
 HttpStreamFactoryImpl::Job::~Job() {
@@ skipping 11 matching lines @@
     session_->proxy_service()->CancelPacRequest(pac_request_);
 
   // The stream could be in a partial state.  It is not reusable.
   if (stream_.get() && next_state_ != STATE_DONE)
     stream_->Close(true /* not reusable */);
 }
 
 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();
 }
 
 int HttpStreamFactoryImpl::Job::Preconnect(int num_streams) {
   DCHECK_GT(num_streams, 0);
   base::WeakPtr<HttpServerProperties> http_server_properties =
       session_->http_server_properties();
   if (http_server_properties &&
       http_server_properties->SupportsRequestPriority(
           HostPortPair::FromURL(request_info_.url))) {
@@ skipping 170 matching lines @@
   MaybeCopyConnectionAttemptsFromSocketOrHandle();
 
   request_->Complete(was_npn_negotiated(), protocol_negotiated(), using_spdy());
   request_->OnWebSocketHandshakeStreamReady(this,
                                             server_ssl_config_,
                                             proxy_info_,
                                             websocket_stream_.release());
   // |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
   // NULL at this point if the SpdySession closed immediately after creation.
   base::WeakPtr<SpdySession> spdy_session = new_spdy_session_;
   new_spdy_session_.reset();
 
   MaybeCopyConnectionAttemptsFromSocketOrHandle();
 
   // TODO(jgraettinger): Notify the factory, and let that notify |request_|,
   // rather than notifying |request_| directly.
   if (IsOrphaned()) {
     if (spdy_session) {
       stream_factory_->OnNewSpdySessionReady(
           spdy_session, spdy_session_direct_, server_ssl_config_, proxy_info_,
           was_npn_negotiated(), protocol_negotiated(), using_spdy(), net_log_);
     }
     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.
 }
 
 void HttpStreamFactoryImpl::Job::OnStreamFailedCallback(int result) {
   DCHECK(!IsPreconnecting());
 
   MaybeCopyConnectionAttemptsFromSocketOrHandle();
 
   if (IsOrphaned()) {
@@ skipping 166 matching lines @@
       next_state_ = STATE_DONE;
       if (new_spdy_session_.get()) {
         base::ThreadTaskRunnerHandle::Get()->PostTask(
             FROM_HERE, base::Bind(&Job::OnNewSpdySessionReadyCallback,
                                   ptr_factory_.GetWeakPtr()));
       } else if (stream_factory_->for_websockets_) {
         DCHECK(websocket_stream_);
         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(
             FROM_HERE,
             base::Bind(&Job::OnStreamReadyCallback, ptr_factory_.GetWeakPtr()));
       }
       return ERR_IO_PENDING;
 
     default:
       DCHECK(result != ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN ||
@@ skipping 66 matching lines @@
         NOTREACHED() << "bad state";
         rv = ERR_FAILED;
         break;
     }
   } while (rv != ERR_IO_PENDING && next_state_ != STATE_NONE);
   return rv;
 }
 
 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);
   return rv;
 }
 
 int HttpStreamFactoryImpl::Job::DoStart() {
   if (IsSpdyAlternative() || IsQuicAlternative()) {
     server_ = alternative_service_.host_port_pair();
   } else {
@@ skipping 472 matching lines @@
 
 int HttpStreamFactoryImpl::Job::DoWaitingUserAction(int result) {
   // This state indicates that the stream request is in a partially
   // completed state, and we've called back to the delegate for more
   // information.
 
   // We're always waiting here for the delegate to call us back.
   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
   // will know when SpdySessions become available.
 
   bool use_relative_url = direct || request_info_.url.SchemeIs("https");
   stream_.reset(new SpdyHttpStream(session, use_relative_url));
   return OK;
 }
 
@@ skipping 33 matching lines @@
 
   CHECK(!stream_.get());
 
   bool direct = !IsHttpsProxyAndHttpUrl();
   if (existing_spdy_session_.get()) {
     // We picked up an existing session, so we don't need our socket.
     if (connection_->socket())
       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;
   }
 
   SpdySessionKey spdy_session_key = GetSpdySessionKey();
   base::WeakPtr<SpdySession> spdy_session;
   int result = valid_spdy_session_pool_->FindAvailableSession(
       spdy_session_key, net_log_, &spdy_session);
   if (result != OK) {
     return result;
   }
   if (spdy_session) {
-    return SetSpdyHttpStream(spdy_session, direct);
+    return SetSpdyHttpStreamOrBidirectionalStream(spdy_session, direct);
   }
 
   result = valid_spdy_session_pool_->CreateAvailableSessionFromSocket(
       spdy_session_key, connection_.Pass(), net_log_, spdy_certificate_error_,
       using_ssl_, &spdy_session);
   if (result != OK) {
     return result;
   }
 
   if (!spdy_session->HasAcceptableTransportSecurity()) {
@@ skipping 14 matching lines @@
   }
 
   new_spdy_session_ = spdy_session;
   spdy_session_direct_ = direct;
   const HostPortPair& host_port_pair = spdy_session_key.host_port_pair();
   base::WeakPtr<HttpServerProperties> http_server_properties =
       session_->http_server_properties();
   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) {
   if (result < 0)
     return result;
 
   session_->proxy_service()->ReportSuccess(proxy_info_,
                                            session_->network_delegate());
   next_state_ = STATE_NONE;
   return OK;
@@ skipping 345 matching lines @@
   if (connection_->socket()) {
     ConnectionAttempts socket_attempts;
     connection_->socket()->GetConnectionAttempts(&socket_attempts);
     request_->AddConnectionAttempts(socket_attempts);
   } else {
     request_->AddConnectionAttempts(connection_->connection_attempts());
   }
 }
 
 }  // namespace net