Introduce RequestWebSocketStream into HttpStreamFactory

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 <string>
+
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/logging.h"
 #include "base/stl_util.h"
 #include "base/strings/string_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/values.h"
 #include "build/build_config.h"
 #include "net/base/connection_type_histograms.h"
 #include "net/base/net_log.h"
@@ skipping 183 matching lines @@
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
         base::Bind(&HttpStreamFactoryImpl::Job::OnIOComplete,
                    ptr_factory_.GetWeakPtr(), OK));
   }
 }
 
 void HttpStreamFactoryImpl::Job::Orphan(const Request* request) {
   DCHECK_EQ(request_, request);
   request_ = NULL;
-  // We've been orphaned, but there's a job we're blocked on. Don't bother
-  // racing, just cancel ourself.
   if (blocking_job_) {
+    // We've been orphaned, but there's a job we're blocked on. Don't bother
+    // racing, just cancel ourself.
     DCHECK(blocking_job_->waiting_job_);
     blocking_job_->waiting_job_ = NULL;
     blocking_job_ = NULL;
+    if (stream_factory_->for_websockets_ &&
+        connection_ && connection_->socket())
+      connection_->socket()->Disconnect();
+    stream_factory_->OnOrphanedJobComplete(this);
+  } else if (stream_factory_->for_websockets_) {
+    // We cancel this job because WebSocketStream can't be created
+    // without a WebSocketStreamBase::Factory which is stored in Request class
+    // and isn't accessible from this job.
+    if (connection_ && connection_->socket())
+      connection_->socket()->Disconnect();
     stream_factory_->OnOrphanedJobComplete(this);
   }
 }
 
 bool HttpStreamFactoryImpl::Job::was_npn_negotiated() const {
   return was_npn_negotiated_;
 }
 
 NextProto HttpStreamFactoryImpl::Job::protocol_negotiated()
     const {
@@ skipping 41 matching lines @@
   }
 }
 
 bool HttpStreamFactoryImpl::Job::CanUseExistingSpdySession() const {
   // We need to make sure that if a spdy session was created for
   // https://somehost/ that we don't use that session for http://somehost:443/.
   // The only time we can use an existing session is if the request URL is
   // https (the normal case) or if we're connection to a SPDY proxy, or
   // if we're running with force_spdy_always_.  crbug.com/133176
   return request_info_.url.SchemeIs("https") ||
+         request_info_.url.SchemeIs("wss") ||
          proxy_info_.proxy_server().is_https() ||
          force_spdy_always_;
 }
 
 void HttpStreamFactoryImpl::Job::OnStreamReadyCallback() {
   DCHECK(stream_.get());
   DCHECK(!IsPreconnecting());
+  DCHECK(!stream_factory_->for_websockets_);
   if (IsOrphaned()) {
     stream_factory_->OnOrphanedJobComplete(this);
   } else {
     request_->Complete(was_npn_negotiated(),
                        protocol_negotiated(),
                        using_spdy(),
                        net_log_);
     request_->OnStreamReady(this, server_ssl_config_, proxy_info_,
                             stream_.release());
   }
   // |this| may be deleted after this call.
 }
 
-void HttpStreamFactoryImpl::Job::OnSpdySessionReadyCallback() {
+void HttpStreamFactoryImpl::Job::OnWebSocketStreamReadyCallback() {
+  DCHECK(websocket_stream_);
+  DCHECK(!IsPreconnecting());
+  DCHECK(stream_factory_->for_websockets_);
+  if (IsOrphaned()) {

[mmenke, 2013/06/19 14:30:28]

This should be:

"DCHECK(!IsOrphaned());", without the branch, right, since we close orphaned
jobs?

[yhirano, 2013/06/20 05:01:44]

Done.

+    stream_factory_->OnOrphanedJobComplete(this);
+  } else {
+    request_->Complete(was_npn_negotiated(),
+                       protocol_negotiated(),
+                       using_spdy(),
+                       net_log_);
+    request_->OnWebSocketStreamReady(this,
+                                     server_ssl_config_,
+                                     proxy_info_,
+                                     websocket_stream_.release());
+  }
+  // |this| may be deleted after this call.
+}
+
+void HttpStreamFactoryImpl::Job::OnNewSpdySessionReadyCallback() {
   DCHECK(!stream_.get());
   DCHECK(!IsPreconnecting());
   DCHECK(using_spdy());
   DCHECK(new_spdy_session_.get());
   scoped_refptr<SpdySession> spdy_session = new_spdy_session_;
   new_spdy_session_ = NULL;
   if (IsOrphaned()) {
-    stream_factory_->OnSpdySessionReady(
+    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_->OnSpdySessionReady(this, spdy_session, spdy_session_direct_);
+    request_->OnNewSpdySessionReady(this, spdy_session, spdy_session_direct_);
   }
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnStreamFailedCallback(int result) {
   DCHECK(!IsPreconnecting());
   if (IsOrphaned())
     stream_factory_->OnOrphanedJobComplete(this);
   else
     request_->OnStreamFailed(this, result, server_ssl_config_);
@@ skipping 39 matching lines @@
   if (IsOrphaned())
     stream_factory_->OnOrphanedJobComplete(this);
   else
     request_->OnHttpsProxyTunnelResponse(
         this, response_info, server_ssl_config_, proxy_info_, stream);
   // |this| may be deleted after this call.
 }
 
 void HttpStreamFactoryImpl::Job::OnPreconnectsComplete() {
   DCHECK(!request_);
-  if (new_spdy_session_.get()) {
-    stream_factory_->OnSpdySessionReady(new_spdy_session_,
-                                        spdy_session_direct_,
-                                        server_ssl_config_,
-                                        proxy_info_,
-                                        was_npn_negotiated(),
-                                        protocol_negotiated(),
-                                        using_spdy(),
-                                        net_log_);
+  if (new_spdy_session_) {
+    stream_factory_->OnNewSpdySessionReady(
+        new_spdy_session_, spdy_session_direct_, server_ssl_config_,
+        proxy_info_, was_npn_negotiated(), protocol_negotiated(), using_spdy(),
+        net_log_);
   }
   stream_factory_->OnPreconnectsComplete(this);
   // |this| may be deleted after this call.
 }
 
 // static
 int HttpStreamFactoryImpl::Job::OnHostResolution(
     SpdySessionPool* spdy_session_pool,
     const SpdySessionKey& spdy_session_key,
     const AddressList& addresses,
@@ skipping 52 matching lines @@
 
         ProxyClientSocket* proxy_socket =
             static_cast<ProxyClientSocket*>(connection_->socket());
         const HttpResponseInfo* tunnel_auth_response =
             proxy_socket->GetConnectResponseInfo();
 
         next_state_ = STATE_WAITING_USER_ACTION;
         base::MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(
-                &HttpStreamFactoryImpl::Job::OnNeedsProxyAuthCallback,
+                &Job::OnNeedsProxyAuthCallback,
                 ptr_factory_.GetWeakPtr(),
                 *tunnel_auth_response,
                 proxy_socket->GetAuthController()));
       }
       return ERR_IO_PENDING;
 
     case ERR_SSL_CLIENT_AUTH_CERT_NEEDED:
       base::MessageLoop::current()->PostTask(
           FROM_HERE,
           base::Bind(
-              &HttpStreamFactoryImpl::Job::OnNeedsClientAuthCallback,
+              &Job::OnNeedsClientAuthCallback,
               ptr_factory_.GetWeakPtr(),
               connection_->ssl_error_response_info().cert_request_info));
       return ERR_IO_PENDING;
 
     case ERR_HTTPS_PROXY_TUNNEL_RESPONSE:
       {
         DCHECK(connection_.get());
         DCHECK(connection_->socket());
         DCHECK(establishing_tunnel_);
 
         ProxyClientSocket* proxy_socket =
             static_cast<ProxyClientSocket*>(connection_->socket());
         base::MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(
-                &HttpStreamFactoryImpl::Job::OnHttpsProxyTunnelResponseCallback,
+                &Job::OnHttpsProxyTunnelResponseCallback,
                 ptr_factory_.GetWeakPtr(),
                 *proxy_socket->GetConnectResponseInfo(),
                 proxy_socket->CreateConnectResponseStream()));
         return ERR_IO_PENDING;
       }
 
     case OK:
       next_state_ = STATE_DONE;
-      if (new_spdy_session_.get()) {
-        base::MessageLoop::current()->PostTask(
-            FROM_HERE,
-            base::Bind(&HttpStreamFactoryImpl::Job::OnSpdySessionReadyCallback,
-                       ptr_factory_.GetWeakPtr()));
-      } else {
+      if (new_spdy_session_) {
         base::MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(
-                &HttpStreamFactoryImpl::Job::OnStreamReadyCallback,
+                &Job::OnNewSpdySessionReadyCallback,
+                ptr_factory_.GetWeakPtr()));
+      } else if (stream_factory_->for_websockets_) {
+        DCHECK(websocket_stream_);
+        base::MessageLoop::current()->PostTask(
+            FROM_HERE,
+            base::Bind(
+                &Job::OnWebSocketStreamReadyCallback,
+                ptr_factory_.GetWeakPtr()));
+      } else {
+        DCHECK(stream_.get());
+        base::MessageLoop::current()->PostTask(
+            FROM_HERE,
+            base::Bind(
+                &Job::OnStreamReadyCallback,
                 ptr_factory_.GetWeakPtr()));
       }
       return ERR_IO_PENDING;
 
     default:
       base::MessageLoop::current()->PostTask(
           FROM_HERE,
           base::Bind(
-              &HttpStreamFactoryImpl::Job::OnStreamFailedCallback,
+              &Job::OnStreamFailedCallback,
               ptr_factory_.GetWeakPtr(),
               result));
       return ERR_IO_PENDING;
   }
   return result;
 }
 
 int HttpStreamFactoryImpl::Job::DoLoop(int result) {
   DCHECK_NE(next_state_, STATE_NONE);
   int rv = result;
@@ skipping 166 matching lines @@
   next_state_ = STATE_INIT_CONNECTION;
   return OK;
 }
 
 int HttpStreamFactoryImpl::Job::DoInitConnection() {
   DCHECK(!blocking_job_);
   DCHECK(!connection_->is_initialized());
   DCHECK(proxy_info_.proxy_server().is_valid());
   next_state_ = STATE_INIT_CONNECTION_COMPLETE;
 
-  using_ssl_ = request_info_.url.SchemeIs("https") || ShouldForceSpdySSL();
+  using_ssl_ = request_info_.url.SchemeIs("https") ||
+      request_info_.url.SchemeIs("wss") || ShouldForceSpdySSL();
   using_spdy_ = false;
 
   if (ShouldForceQuic())
     using_quic_ = true;
 
   if (using_quic_) {
     DCHECK(session_->params().enable_quic);
     if (!proxy_info_.is_direct()) {
       NOTREACHED();
       // TODO(rch): support QUIC proxies.
@@ skipping 64 matching lines @@
     // Disable revocation checking for HTTPS proxies since the revocation
     // requests are probably going to need to go through the proxy too.
     proxy_ssl_config_.rev_checking_enabled = false;
   }
   if (using_ssl_) {
     InitSSLConfig(origin_, &server_ssl_config_,
                   false /* not a proxy server */);
   }
 
   if (IsPreconnecting()) {
+    DCHECK(!stream_factory_->for_websockets_);
     return PreconnectSocketsForHttpRequest(
         origin_url_,
         request_info_.extra_headers,
         request_info_.load_flags,
         priority_,
         session_,
         proxy_info_,
         ShouldForceSpdySSL(),
         want_spdy_over_npn,
         server_ssl_config_,
         proxy_ssl_config_,
         request_info_.privacy_mode,
         net_log_,
         num_streams_);
   } else {
     // If we can't use a SPDY session, don't both checking for one after
     // the hostname is resolved.
     OnHostResolutionCallback resolution_callback = CanUseExistingSpdySession() ?
         base::Bind(&Job::OnHostResolution, session_->spdy_session_pool(),
                    GetSpdySessionKey()) :
         OnHostResolutionCallback();
+    if (stream_factory_->for_websockets_) {
+      return InitSocketHandleForWebSocketRequest(
+          origin_url_, request_info_.extra_headers, request_info_.load_flags,
+          priority_, session_, proxy_info_, ShouldForceSpdySSL(),
+          want_spdy_over_npn, server_ssl_config_, proxy_ssl_config_,
+          request_info_.privacy_mode, net_log_,
+          connection_.get(), resolution_callback, io_callback_);
+    }
     return InitSocketHandleForHttpRequest(
         origin_url_, request_info_.extra_headers, request_info_.load_flags,
         priority_, session_, proxy_info_, ShouldForceSpdySSL(),
         want_spdy_over_npn, server_ssl_config_, proxy_ssl_config_,
         request_info_.privacy_mode, net_log_,
         connection_.get(), resolution_callback, io_callback_);
   }
 }
 
 int HttpStreamFactoryImpl::Job::DoInitConnectionComplete(int result) {
@@ skipping 49 matching lines @@
     if (ssl_socket->WasNpnNegotiated()) {
       was_npn_negotiated_ = true;
       std::string proto;
       std::string server_protos;
       SSLClientSocket::NextProtoStatus status =
           ssl_socket->GetNextProto(&proto, &server_protos);
       NextProto protocol_negotiated =
           SSLClientSocket::NextProtoFromString(proto);
       protocol_negotiated_ = protocol_negotiated;
       net_log_.AddEvent(
-           NetLog::TYPE_HTTP_STREAM_REQUEST_PROTO,
-           base::Bind(&NetLogHttpStreamProtoCallback,
-                      status, &proto, &server_protos));
+          NetLog::TYPE_HTTP_STREAM_REQUEST_PROTO,
+          base::Bind(&NetLogHttpStreamProtoCallback,
+                     status, &proto, &server_protos));
       if (ssl_socket->was_spdy_negotiated())
         SwitchToSpdyMode();
     }
     if (ShouldForceSpdySSL())
       SwitchToSpdyMode();
   } else if (proxy_info_.is_https() && connection_->socket() &&
         result == OK) {
     ProxyClientSocket* proxy_socket =
       static_cast<ProxyClientSocket*>(connection_->socket());
     if (proxy_socket->IsUsingSpdy()) {
@@ skipping 95 matching lines @@
   if (connection_->socket() && !connection_->is_reused())
     SetSocketMotivation();
 
   if (!using_spdy_) {
     // We may get ftp scheme when fetching ftp resources through proxy.
     bool using_proxy = (proxy_info_.is_http() || proxy_info_.is_https()) &&
                        (request_info_.url.SchemeIs("http") ||
                         request_info_.url.SchemeIs("ftp"));
     if (stream_factory_->http_pipelined_host_pool_.
             IsExistingPipelineAvailableForKey(*http_pipelining_key_.get())) {
+      DCHECK(!stream_factory_->for_websockets_);
       stream_.reset(stream_factory_->http_pipelined_host_pool_.
                     CreateStreamOnExistingPipeline(
                         *http_pipelining_key_.get()));
       CHECK(stream_.get());
+    } else if (stream_factory_->for_websockets_) {
+      DCHECK(request_);
+      DCHECK(request_->websocket_stream_factory());
+      websocket_stream_.reset(
+          request_->websocket_stream_factory()->CreateBasicStream(
+              connection_.release(), using_proxy));
     } else if (!using_proxy && IsRequestEligibleForPipelining()) {
       // TODO(simonjam): Support proxies.
       stream_.reset(
           stream_factory_->http_pipelined_host_pool_.CreateStreamOnNewPipeline(
               *http_pipelining_key_.get(),
               connection_.release(),
               server_ssl_config_,
               proxy_info_,
               net_log_,
               was_npn_negotiated_,
@@ skipping 51 matching lines @@
     }
   }
 
   if (spdy_session->IsClosed())
     return ERR_CONNECTION_CLOSED;
 
   // 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(spdy_session.get(), use_relative_url));
+  if (stream_factory_->for_websockets_) {
+    DCHECK(request_);
+    DCHECK(request_->websocket_stream_factory());
+    bool use_relative_url = direct || request_info_.url.SchemeIs("wss");
+    websocket_stream_.reset(
+        request_->websocket_stream_factory()->CreateSpdyStream(
+            spdy_session, use_relative_url));
+  } else {
+    bool use_relative_url = direct || request_info_.url.SchemeIs("https");
+    stream_.reset(new SpdyHttpStream(spdy_session, use_relative_url));
+  }
   return OK;
 }
 
 int HttpStreamFactoryImpl::Job::DoCreateStreamComplete(int result) {
   if (result < 0)
     return result;
 
   session_->proxy_service()->ReportSuccess(proxy_info_);
   next_state_ = STATE_NONE;
   return OK;
@@ skipping 280 matching lines @@
 }
 
 bool HttpStreamFactoryImpl::Job::IsOrphaned() const {
   return !IsPreconnecting() && !request_;
 }
 
 bool HttpStreamFactoryImpl::Job::IsRequestEligibleForPipelining() {
   if (IsPreconnecting() || !request_) {
     return false;
   }
+  if (stream_factory_->for_websockets_) {
+    return false;
+  }
   if (session_->force_http_pipelining()) {
     return true;
   }
   if (!session_->params().http_pipelining_enabled) {
     return false;
   }
   if (using_ssl_) {
     return false;
   }
   if (request_info_.method != "GET" && request_info_.method != "HEAD") {
     return false;
   }
   if (request_info_.load_flags &
       (net::LOAD_MAIN_FRAME | net::LOAD_SUB_FRAME | net::LOAD_PREFETCH |
        net::LOAD_IS_DOWNLOAD)) {
     // Avoid pipelining resources that may be streamed for a long time.
     return false;
   }
   return stream_factory_->http_pipelined_host_pool_.IsKeyEligibleForPipelining(
       *http_pipelining_key_.get());
 }
 
 }  // namespace net