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 23 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_session.h"
 #include "net/spdy/spdy_session_pool.h"
 #include "net/ssl/ssl_cert_request_info.h"
 #include "net/ssl/ssl_failure_state.h"
 
 namespace net {
 
 // Returns parameters associated with the start of a HTTP stream job.
 scoped_ptr<base::Value> NetLogHttpStreamJobCallback(
@@ skipping 73 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_);
+  DCHECK(!IsPreconnecting());
+  DCHECK(!stream_factory_->for_websockets_);

[mmenke, 2015/10/08 19:31:53]

Think the latter two of these make more sense as DCHECKs in StartInternal (Know
you're copying from OnWebSocketHandshakeStreamReadyCallback, but think that gets
things wrong there, too)

[xunjieli, 2015/10/19 21:07:46]

Hmm.. A few unit tests fail those two DCHECKs if I move them to StartInternal.

+
+  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,
+                                      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_) {

[mmenke, 2015/10/08 19:31:53]

Hrm...Should we fail earlier for non-HTTPS/HTTP2 requests?

[xunjieli, 2015/10/19 21:07:46]

I added a check in the wrapper to fail early for non-HTTPS requests. Is there a
way to tell from the response info that this is a HTTP2 request?

+        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(
             FROM_HERE,
             base::Bind(&Job::OnStreamReadyCallback, ptr_factory_.GetWeakPtr()));
       }
       return ERR_IO_PENDING;
 
     default:
       DCHECK(result != ERR_ALTERNATIVE_CERT_NOT_VALID_FOR_ORIGIN ||
@@ skipping 558 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(

[mmenke, 2015/10/08 19:31:53]

This method needs to be renamed.

[xunjieli, 2015/10/19 21:07:46]

Done.

     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 441 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