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/string_util.h"
 #include "base/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 252 matching lines @@
   // 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") ||
          proxy_info_.proxy_server().is_https() ||
          force_spdy_always_;
 }
 
 void HttpStreamFactoryImpl::Job::OnStreamReadyCallback() {
+  DCHECK(!request_->for_websocket());
   DCHECK(stream_.get());
   DCHECK(!IsPreconnecting());
   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::OnSocketReadyForWebSocketCallback() {
+  DCHECK(request_->for_websocket());
+  DCHECK(connection_.get() && connection_->is_initialized());
+  DCHECK(!IsPreconnecting());

[mmenke, 2013/05/23 20:26:20]

Looks like when we're preconnected, request_ is NULL, so this should probably go
before we dereference request_.

[yhirano, 2013/05/24 14:11:16]

Done.

+  if (IsOrphaned()) {
+    stream_factory_->OnOrphanedJobComplete(this);
+  } else {
+    request_->Complete(was_npn_negotiated(),
+                       protocol_negotiated(),
+                       using_spdy(),
+                       net_log_);
+    request_->OnSocketReadyForWebSocket(this, server_ssl_config_, proxy_info_,
+                                        connection_.release());
+  }
+  // |this| may be deleted after this call.
+}
+
+void HttpStreamFactoryImpl::Job::OnSpdySessionReadyForWebSocketCallback() {
+  DCHECK(request_->for_websocket());
+  DCHECK(spdy_session_to_pass_);
+  DCHECK(!IsPreconnecting());
+  scoped_refptr<SpdySession> session = spdy_session_to_pass_;
+  spdy_session_to_pass_ = NULL;
+  if (IsOrphaned()) {
+    stream_factory_->OnOrphanedJobComplete(this);

[mmenke, 2013/05/23 20:26:20]

This is what was confusing me before - we don't call OnNewSpdySessionReady here.
 So if we're orphaned, and there's another request out for the same SpdySession,
when we're here, don't make the session available for the other request.  This
seems like a problem.

[yhirano, 2013/05/24 14:11:16]

In PS 13 (or newer), OnNewSpdySessionReady is called when a new spdy session is
generated regardless of for_websocket value.

+  } else {
+    request_->Complete(was_npn_negotiated(),
+                       protocol_negotiated(),
+                       using_spdy(),
+                       net_log_);
+    request_->OnSpdySessionReadyForWebSocket(this,
+                                             server_ssl_config_,
+                                             proxy_info_,
+                                             session);
+  }
+  // |this| may be deleted after this call.
+}
+
+void HttpStreamFactoryImpl::Job::OnNewSpdySessionReadyCallback() {
   DCHECK(!stream_.get());
   DCHECK(!IsPreconnecting());
+  DCHECK(!request_->for_websocket());
   DCHECK(using_spdy());
   DCHECK(new_spdy_session_);
   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 40 matching lines @@
     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_) {
-    stream_factory_->OnSpdySessionReady(
+    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(
@@ skipping 55 matching lines @@
 
         ProxyClientSocket* proxy_socket =
             static_cast<ProxyClientSocket*>(connection_->socket());
         const HttpResponseInfo* tunnel_auth_response =
             proxy_socket->GetConnectResponseInfo();
 
         next_state_ = STATE_WAITING_USER_ACTION;
         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:
       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());
         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_) {
+      if (request_->for_websocket()) {
+        if (spdy_session_to_pass_) {
+          MessageLoop::current()->PostTask(
+              FROM_HERE,
+              base::Bind(
+                  &Job::OnSpdySessionReadyForWebSocketCallback,
+                  ptr_factory_.GetWeakPtr()));
+        } else if (connection_ && connection_->is_initialized()) {
+          MessageLoop::current()->PostTask(
+              FROM_HERE,
+              base::Bind(
+                  &Job::OnSocketReadyForWebSocketCallback,
+                  ptr_factory_.GetWeakPtr()));
+        } else {
+          return ERR_FAILED;

[mmenke, 2013/05/23 20:26:20]

We should fail asynchronously in this case, just like in the default case. 
Notice that HttpStreamFactoryImpl::Job::OnIOComplete() doesn't actually do
anything with the result, so this function always has to return ERR_IO_PENDING
(Which, admittedly, makes having a return value a little weird, but that's
beyond the scope of this CL).

[yhirano, 2013/05/24 14:11:16]

Done.

+        }
+      } else if (new_spdy_session_) {
         MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(
-                &HttpStreamFactoryImpl::Job::OnSpdySessionReadyCallback,
+                &Job::OnNewSpdySessionReadyCallback,
                 ptr_factory_.GetWeakPtr()));
       } else {
         MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(
-                &HttpStreamFactoryImpl::Job::OnStreamReadyCallback,
+                &Job::OnStreamReadyCallback,
                 ptr_factory_.GetWeakPtr()));
       }
       return ERR_IO_PENDING;
 
     default:
       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 337 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 93 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(!request_->for_websocket());
       stream_.reset(stream_factory_->http_pipelined_host_pool_.
                     CreateStreamOnExistingPipeline(
                         *http_pipelining_key_.get()));
       CHECK(stream_.get());
+    } else if (request_->for_websocket()) {
+      // Do nothing.
+      // connection_ will be passed to the delegate.

[mmenke, 2013/05/23 20:26:20]

Not creating a stream in DoCreateStream in this case seems a bit odd.

[yhirano, 2013/05/24 14:11:16]

Done.

     } 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 34 matching lines @@
           pair, connection_.release(), net_log_, spdy_certificate_error_,
           &new_spdy_session_, using_ssl_);
       if (error != OK)
         return error;
       const HostPortPair& host_port_pair = pair.first;
       HttpServerProperties* http_server_properties =
           session_->http_server_properties();
       if (http_server_properties)
         http_server_properties->SetSupportsSpdy(host_port_pair, true);
       spdy_session_direct_ = direct;
+      if (request_->for_websocket()) {
+        DCHECK(!spdy_session_to_pass_);
+        spdy_session_to_pass_.swap(new_spdy_session_);
+      }
       return OK;
     }
   }
 
   if (spdy_session->IsClosed())
     return ERR_CONNECTION_CLOSED;
 
+  if (request_->for_websocket()) {
+    DCHECK(!spdy_session_to_pass_);
+    spdy_session_to_pass_.swap(spdy_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(spdy_session, use_relative_url));
   return OK;
 }
 
 int HttpStreamFactoryImpl::Job::DoCreateStreamComplete(int result) {
@@ skipping 282 matching lines @@
 }
 
 bool HttpStreamFactoryImpl::Job::IsOrphaned() const {
   return !IsPreconnecting() && !request_;
 }
 
 bool HttpStreamFactoryImpl::Job::IsRequestEligibleForPipelining() {
   if (IsPreconnecting() || !request_) {
     return false;
   }
+  if (request_->for_websocket()) {
+    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