Implement QUIC-based net::BidirectionalStream

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 <utility>
 
@@ skipping 198 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),
+      stream_type_(HttpStreamRequest::BIDIRECTIONAL_STREAM),
       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();
+  // Saves |stream_type_|, since request is nulled when job is orphaned.
+  stream_type_ = request_->stream_type();
   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 232 matching lines @@
   // 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 {
-    if (for_bidirectional_) {
+    if (stream_type_ == HttpStreamRequest::BIDIRECTIONAL_STREAM) {
 #if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
       DCHECK(bidirectional_stream_job_);
       request_->OnNewSpdySessionReady(this, /*spdy_http_stream=*/nullptr,
                                       std::move(bidirectional_stream_job_),
                                       spdy_session, spdy_session_direct_);
 #else
       DCHECK(false);
 #endif
-
     } 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.
 }
 
@@ skipping 172 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_) {
+      } else if (stream_type_ == HttpStreamRequest::BIDIRECTIONAL_STREAM) {
 #if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
         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()));
         }
@@ skipping 292 matching lines @@
       if (session_->quic_stream_factory()->IsQuicDisabled(destination.port()))
         return ERR_QUIC_PROTOCOL_ERROR;
     } else {
       DCHECK(using_ssl_);
       // The certificate of a QUIC alternative server is expected to be valid
       // for the origin of the request (in addition to being valid for the
       // server itself).
       destination = server_;
       ssl_config = &server_ssl_config_;
     }
-
     int rv =
         quic_request_.Request(destination, request_info_.privacy_mode,
                               ssl_config->GetCertVerifyFlags(), url,
                               request_info_.method, net_log_, io_callback_);
     if (rv == OK) {
       using_existing_quic_session_ = true;
     } else {
       // OK, there's no available QUIC session. Let |waiting_job_| resume
       // if it's paused.
       if (waiting_job_) {
@@ skipping 221 matching lines @@
     MaybeMarkAlternativeServiceBroken();
     return result;
   }
 
   if (using_quic_) {
     if (result < 0) {
       job_status_ = STATUS_BROKEN;
       MaybeMarkAlternativeServiceBroken();
       return result;
     }
-    stream_ = quic_request_.ReleaseStream();
+    if (stream_type_ == HttpStreamRequest::BIDIRECTIONAL_STREAM) {
+#if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
+      bidirectional_stream_job_ = quic_request_.CreateBidirectionalStreamJob();
+      if (!bidirectional_stream_job_) {
+        // Quic session is closed before stream can be created.
+        return ERR_CONNECTION_CLOSED;
+      }
+#else
+      NOTREACHED();
+#endif
+    } else {
+      stream_ = quic_request_.CreateStream();
+      if (!stream_) {
+        // Quic session is closed before stream can be created.
+        return ERR_CONNECTION_CLOSED;
+      }
+    }
     next_state_ = STATE_NONE;
     return OK;
   }
 
   if (result < 0 && !ssl_started)
     return ReconsiderProxyAfterError(result);
   establishing_tunnel_ = false;
 
   if (connection_->socket()) {
     // We officially have a new connection.  Record the type.
@@ skipping 36 matching lines @@
   return ERR_IO_PENDING;
 }
 
 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 (stream_type_ == HttpStreamRequest::BIDIRECTIONAL_STREAM) {
 #if BUILDFLAG(ENABLE_BIDIRECTIONAL_STREAM)
-    // TODO(xunjieli): Create QUIC's version of BidirectionalStreamJob.
     bidirectional_stream_job_.reset(new BidirectionalStreamSpdyJob(session));
     return OK;
 #else
-    DCHECK(false);
+    NOTREACHED();
     return ERR_FAILED;
 #endif
   }
 
   // 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));
@@ skipping 467 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