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 58 matching lines @@
                                 const HttpRequestInfo& request_info,
                                 RequestPriority priority,
                                 const SSLConfig& server_ssl_config,
                                 const SSLConfig& proxy_ssl_config,
                                 NetLog* net_log)
     : request_(NULL),
       request_info_(request_info),
       priority_(priority),
       server_ssl_config_(server_ssl_config),
       proxy_ssl_config_(proxy_ssl_config),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_HTTP_STREAM_JOB)),

[mmenke, 2013/06/10 20:26:03]

Suggestion for another CL:  If this is for a WebSocketStream, we could use
different NetLog source type.  In the future, this could make it easier to debug
WebSocket issues.

Suggest another CL because net-internals code would need to have a little extra
code, too, and think it's easier just to worry about it in another CL.

[yhirano, 2013/06/11 11:13:16]

Filed a bug.

       io_callback_(base::Bind(&Job::OnIOComplete, base::Unretained(this))),
       connection_(new ClientSocketHandle),
       session_(session),
       stream_factory_(stream_factory),
       next_state_(STATE_NONE),
       pac_request_(NULL),
       blocking_job_(NULL),
       waiting_job_(NULL),
       using_ssl_(false),
       using_spdy_(false),
@@ skipping 22 matching lines @@
   if (next_state_ == STATE_WAITING_USER_ACTION) {
     connection_->socket()->Disconnect();
     connection_.reset();
   }
 
   if (pac_request_)
     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 */);

[mmenke, 2013/06/10 20:26:03]

I'm assuming that WebSocketStreams do not need similar treatment, since they'll
never return anything to the socket pool.

[yhirano, 2013/06/11 11:13:16]

I think so.

 }
 
 void HttpStreamFactoryImpl::Job::Start(Request* request) {
   DCHECK(request);
   request_ = request;
   StartInternal();
 }
 
 int HttpStreamFactoryImpl::Job::Preconnect(int num_streams) {
   DCHECK_GT(num_streams, 0);
@@ skipping 128 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()) {
+    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()) {
+      if (new_spdy_session_) {
         base::MessageLoop::current()->PostTask(
             FROM_HERE,
-            base::Bind(&HttpStreamFactoryImpl::Job::OnSpdySessionReadyCallback,
-                       ptr_factory_.GetWeakPtr()));
+            base::Bind(
+                &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 {

[mmenke, 2013/06/10 20:26:03]

optional nit:  Just to keep this structure parallel with the above code, may
want to add "DCHECK(stream_.get());" here.

[yhirano, 2013/06/11 11:13:16]

Done.

         base::MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(
-                &HttpStreamFactoryImpl::Job::OnStreamReadyCallback,
+                &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_ && request_->websocket_stream_factory());

[mmenke, 2013/06/10 20:26:03]

If we're orphaned before this point, request_ will be NULL, and we're in trouble
here.  Seems like there are a couple:

1)  Have the HttpStreamFactory have a single stream factory pointer, and just
use that for all jobs.
2)  Have the Jobs keep track of their HttpStreamFactory pointers themselves.
3)  Fail WebSocket jobs once they're orphaned.  This may be the way to go.  I'm
not sure what the expected lifetime of a WebSocketStream is going to be, or what
the odds are of destroying one and then re-creating one for the same server. 
May be other ways to bypass the orphan logic for WebSocket jobs.

I think we should also have a unit test for this case.

[yhirano, 2013/06/11 11:13:16]

I chose the third way.
I tried, but I could not create the situation.

[mmenke, 2013/06/12 02:27:05]

Just deleting the waiter in the unit tests before we get a stream, and then
waiting until idle and making sure the pool has no active/idle sockets doesn't
work?

I'd really like a test for this case if we can manage one - think it'll be an
obscure enough case that it will easily be overlooked in future refactors.

I'll investigate myself tomorrow, if you're still unable to trigger the
codepath.  Think we can land the CL after that, though I want to do a final once
over.

[mmenke, 2013/06/12 02:31:13]

You may need to create mock socket data that connects asynchronously, though
since we call Job::OnStreamReadyCallback asynchronously, I don't think this is
necessary.

[mmenke, 2013/06/12 02:54:25]

Actually, I think we should make sure there is a pending connection, then delete
the request, then make sure there are no pending connections, then run until
idle, and check again.

[yhirano, 2013/06/12 06:44:56]

Neither deleting waiter nor request call HttpStreamFactoryImpl::OrphanJob nor
HttpStreamFactoryImpl::Job::Orphan (I believe you meant request, not waiter).
I think there should be an OrphanJobs call in
HttpStreamFactoryImpl::Request::~Request, but I'm not confident of that.

[mmenke, 2013/06/12 18:21:33]

You're right - all calls to OrphanJobs go through OrphanJobsExcept, so we need
to have multiple jobs for a request to reach that code.  Requests are added in
HttpStreamFactoryImpl::RequestStream, when we get an alternate scheme.  Deleting
is also not enough to trigger orphaning other jobs.

I still feel we should probably have a test for this, but I'm going to have to
do some digging to figure out how to do that.

[yhirano, 2013/06/13 14:20:38]

Done.
To control the timing of socket connection, I added
net::AsyncSocket::OnConnectComplete and changed the implementation of
MockTcpClientSocket::Connect a bit.
I am examining if the change breaks other tests by using trybots.

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