WebSocket over SPDY: WebSocketJob handling SpdyWebSocketStream

net/websockets/websocket_job.cc

 // Copyright (c) 2011 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/websockets/websocket_job.h"
 
 #include <algorithm>
 
 #include "base/lazy_instance.h"
 #include "base/string_tokenizer.h"
 #include "googleurl/src/gurl.h"
 #include "net/base/net_errors.h"
 #include "net/base/net_log.h"
 #include "net/base/cookie_store.h"
 #include "net/base/io_buffer.h"
 #include "net/http/http_network_session.h"
 #include "net/http/http_transaction_factory.h"
 #include "net/http/http_util.h"
 #include "net/spdy/spdy_session.h"
 #include "net/spdy/spdy_session_pool.h"
 #include "net/url_request/url_request_context.h"
 #include "net/websockets/websocket_frame_handler.h"
 #include "net/websockets/websocket_handshake_handler.h"
 #include "net/websockets/websocket_net_log_params.h"
 #include "net/websockets/websocket_throttle.h"
 
+static const int kMaxPendingSendAllowed = 32768;  // 32 kilobytes.
+
 namespace {
 
 // lower-case header names.
 const char* const kCookieHeaders[] = {
   "cookie", "cookie2"
 };
 const char* const kSetCookieHeaders[] = {
   "set-cookie", "set-cookie2"
 };
 
@@ skipping 134 matching lines @@
     Release();  // Balanced with OnStartOpenConnection().
   }
 }
 
 int WebSocketJob::OnStartOpenConnection(
     SocketStream* socket, CompletionCallback* callback) {
   DCHECK(!callback_);
   state_ = CONNECTING;
   addresses_ = socket->address_list();
   WebSocketThrottle::GetInstance()->PutInQueue(this);
-  if (!waiting_) {
-    int result = TrySpdyStream();
-    if (result != ERR_IO_PENDING)
-      return result;
+  if (waiting_) {
+    // PutInQueue() may set |waiting_| true for throttling. In this case,
+    // Wakeup() will be called later.
+    callback_ = callback;
+    AddRef();  // Balanced when callback_ becomes NULL.
+    return ERR_IO_PENDING;
   }
-  callback_ = callback;
-  AddRef();  // Balanced when callback_ becomes NULL.
-  return ERR_IO_PENDING;  // Wakeup will be called later.
+  return TrySpdyStream();
 }
 
 void WebSocketJob::OnConnected(
     SocketStream* socket, int max_pending_send_allowed) {
   if (state_ == CLOSED)
     return;
   DCHECK_EQ(CONNECTING, state_);
   if (delegate_)
     delegate_->OnConnected(socket, max_pending_send_allowed);
 }
@@ skipping 70 matching lines @@
     delegate->OnClose(socket);
 }
 
 void WebSocketJob::OnAuthRequired(
     SocketStream* socket, AuthChallengeInfo* auth_info) {
   if (delegate_)
     delegate_->OnAuthRequired(socket, auth_info);
 }
 
 void WebSocketJob::OnError(const SocketStream* socket, int error) {
+  if (delegate_ && error != ERR_PROTOCOL_SWITCHED)
+    delegate_->OnError(socket, error);
+}
+
+void WebSocketJob::OnCreatedSpdyStream(int result) {
+  DCHECK(spdy_websocket_stream_.get());
+  DCHECK(socket_.get());
+  DCHECK_NE(ERR_IO_PENDING, result);
+
+  if (state_ == CLOSED) {
+    result = ERR_ABORTED;
+  } else if (result == OK) {
+    state_ = CONNECTING;
+    result = ERR_PROTOCOL_SWITCHED;
+  } else {
+    spdy_websocket_stream_.reset();
+  }
+
+  CompleteIO(result);
+}
+
+void WebSocketJob::OnSentSpdyHeaders(int result) {
+  DCHECK_NE(INITIALIZED, state_);
+  if (state_ != CONNECTING)
+    return;
   if (delegate_)
-    delegate_->OnError(socket, error);
+    delegate_->OnSentData(socket_, handshake_request_->original_length());
+  handshake_request_.reset();
+}
+
+int WebSocketJob::OnReceivedSpdyResponseHeader(
+    const spdy::SpdyHeaderBlock& headers, int status) {
+  DCHECK_NE(INITIALIZED, state_);
+  if (state_ != CONNECTING)
+    return status;
+  if (status != OK)
+    return status;
+  // TODO(toyoshim): Fallback to non-spdy connection?
+  handshake_response_->ParseResponseHeaderBlock(headers, challenge_);
+
+  SaveCookiesAndNotifyHeaderComplete();
+  return OK;
+}
+
+void WebSocketJob::OnSentSpdyData(int amount_sent) {
+  DCHECK_NE(INITIALIZED, state_);
+  DCHECK_NE(CONNECTING, state_);
+  if (state_ == CLOSED)
+    return;
+  if (!spdy_websocket_stream_.get())
+    return;
+  OnSentData(socket_, amount_sent);
+}
+
+void WebSocketJob::OnReceivedSpdyData(const char* data, int length) {
+  DCHECK_NE(INITIALIZED, state_);
+  DCHECK_NE(CONNECTING, state_);
+  if (state_ == CLOSED)
+    return;
+  if (!spdy_websocket_stream_.get())
+    return;
+  OnReceivedData(socket_, data, length);
+}
+
+void WebSocketJob::OnCloseSpdyStream() {
+  spdy_websocket_stream_.reset();
+  OnClose(socket_);
 }
 
 bool WebSocketJob::SendHandshakeRequest(const char* data, int len) {
   DCHECK_EQ(state_, CONNECTING);
   if (started_to_send_handshake_request_)
     return false;
   if (!handshake_request_->ParseRequest(data, len))
     return false;
 
   // handshake message is completed.
@@ skipping 19 matching lines @@
         CookieOptions cookie_options;
         cookie_options.set_include_httponly();
         std::string cookie =
             socket_->context()->cookie_store()->GetCookiesWithOptions(
                 GetURLForCookies(), cookie_options);
         if (!cookie.empty())
           handshake_request_->AppendHeaderIfMissing("Cookie", cookie);
       }
     }
 
-    const std::string& handshake_request = handshake_request_->GetRawRequest();
-    handshake_request_sent_ = 0;
-    socket_->net_log()->AddEvent(
-        NetLog::TYPE_WEB_SOCKET_SEND_REQUEST_HEADERS,
-        make_scoped_refptr(
-            new NetLogWebSocketHandshakeParameter(handshake_request)));
-    socket_->SendData(handshake_request.data(),
-                      handshake_request.size());
+    if (spdy_websocket_stream_.get()) {
+      linked_ptr<spdy::SpdyHeaderBlock> headers(new spdy::SpdyHeaderBlock);
+      handshake_request_->GetRequestHeaderBlock(
+          socket_->url(), headers.get(), &challenge_);
+      spdy_websocket_stream_->SendRequest(headers);
+    } else {
+      const std::string& handshake_request =
+          handshake_request_->GetRawRequest();
+      handshake_request_sent_ = 0;
+      socket_->net_log()->AddEvent(
+          NetLog::TYPE_WEB_SOCKET_SEND_REQUEST_HEADERS,
+          make_scoped_refptr(
+              new NetLogWebSocketHandshakeParameter(handshake_request)));
+      socket_->SendData(handshake_request.data(),
+                        handshake_request.size());
+    }
   }
 }
 
 void WebSocketJob::OnSentHandshakeRequest(
     SocketStream* socket, int amount_sent) {
   DCHECK_EQ(state_, CONNECTING);
   handshake_request_sent_ += amount_sent;
   DCHECK_LE(handshake_request_sent_, handshake_request_->raw_length());
   if (handshake_request_sent_ >= handshake_request_->raw_length()) {
     // handshake request has been sent.
@@ skipping 119 matching lines @@
     // If so, use it to create the websocket stream.
     HttpTransactionFactory* factory =
         socket_->context()->http_transaction_factory();
     if (factory) {
       scoped_refptr<HttpNetworkSession> session = factory->GetSession();
       if (session.get()) {
         SpdySessionPool* spdy_pool = session->spdy_session_pool();
         const HostPortProxyPair pair(HostPortPair::FromURL(socket_->url()),
                                      socket_->proxy_server());
         if (spdy_pool->HasSession(pair)) {
-          // TODO(toyoshim): Switch to SpdyWebSocketStream here by returning
-          // ERR_PROTOCOL_SWITCHED.
+          scoped_refptr<SpdySession> spdy_session =
+              spdy_pool->Get(pair, *socket_->net_log());
+
+          SSLInfo ssl_info;
+          bool was_npn_negotiated;
+          bool use_ssl =
+              spdy_session->GetSSLInfo(&ssl_info, &was_npn_negotiated);
+
+          // Forbid wss downgrade to SPDY without SSL.
+          if (!socket_->is_secure() || use_ssl) {

[Yuta Kitamura, 2011/07/08 07:40:21]

This seems right, but an identical expression "!(socket_->is_secure() &&
!use_ssl)" may be a bit more straightforward.

[Takashi Toyoshima, 2011/07/08 08:35:14]

Done.

+            spdy_websocket_stream_.reset(
+                new SpdyWebSocketStream(spdy_session, this));
+
+            int result = spdy_websocket_stream_->InitializeStream(
+                socket_->url(), MEDIUM, *socket_->net_log());
+            if (result == OK) {
+              OnConnected(socket_, kMaxPendingSendAllowed);
+              return ERR_PROTOCOL_SWITCHED;
+            }
+            if (result == ERR_IO_PENDING)
+              return ERR_IO_PENDING;
+
+            // Fallback to WebSocket wire protocol.
+            spdy_websocket_stream_.reset();
+          }

[Yuta Kitamura, 2011/07/08 07:40:21]

(Optional) These blocks seem nested too deeply.

Generally speaking, it is preferred to use "early exits" style rather than "late
exits".

Preferred:
  if (...)
    return status;
  ...
  if (...)
    return status;
  ...
  if (...)
    return status;
  ...

Not preferred:
  if (...) {
    if (...) {
      if (...) {
        ...
      }
    }
  }
  return status;

This helps readability a lot. However, this is not required by our style guide,
so this is optional.

[Takashi Toyoshima, 2011/07/08 08:35:14]

I love it.
Done!

         }
       }
     }
   }
   // No SPDY session was available.
   // Fallback to connecting a new socket.
   return OK;
 }
 
 void WebSocketJob::SetWaiting() {
   waiting_ = true;
 }
 
 bool WebSocketJob::IsWaiting() const {
   return waiting_;
 }
 
 void WebSocketJob::Wakeup() {
   if (!waiting_)
     return;
   waiting_ = false;
   DCHECK(callback_);
   MessageLoopForIO::current()->PostTask(
       FROM_HERE,
       NewRunnableMethod(this, &WebSocketJob::RetryPendingIO));
 }
 
 void WebSocketJob::RetryPendingIO() {
-  int result = TrySpdyStream();
+  CompleteIO(TrySpdyStream());

[Yuta Kitamura, 2011/07/08 07:40:21]

TrySpdyStream() can return ERR_IO_PENDING. Is it okay to call
CompleteIO(ERR_IO_PENDING)? (because ERR_IO_PENDING means the I/O in question
has not finished yet)

[Takashi Toyoshima, 2011/07/08 08:35:14]

Oh!
It's a bug to cause DCHECK failure !!
In that case, we should call CompleteIO() from OnCreatedSpdyStream().
Thanks.

I have a plan to add tests which can check this case.
But, I'd like to split it into another change.
The bug is fixed in the next change set.

+}
+
+void WebSocketJob::CompleteIO(int result) {
   // |callback_| may be NULL if OnClose() or DetachDelegate() was called.
   if (callback_) {
     net::CompletionCallback* callback = callback_;
     callback_ = NULL;
     callback->Run(result);
     Release();  // Balanced with OnStartOpenConnection().
   }
 }
 
 bool WebSocketJob::SendDataInternal(const char* data, int length) {
-  // TODO(toyoshim): Call protocol specific SendData().
+  if (spdy_websocket_stream_.get())
+    return ERR_IO_PENDING == spdy_websocket_stream_->SendData(data, length);
   return socket_->SendData(data, length);
 }
 
 void WebSocketJob::CloseInternal() {
-  // TODO(toyoshim): Call protocol specific Close().
+  if (spdy_websocket_stream_.get())
+    spdy_websocket_stream_->Close();
   socket_->Close();
 }
 
 void WebSocketJob::SendPending() {
   if (current_buffer_)
     return;
   // Current buffer is done.  Try next buffer if any.
   // Don't buffer sending data. See comment on case OPEN in SendData().
   if (send_frame_handler_->UpdateCurrentBuffer(false) <= 0) {
     // No more data to send.
     if (state_ == CLOSING)
       CloseInternal();
     return;
   }
   current_buffer_ = new DrainableIOBuffer(
       send_frame_handler_->GetCurrentBuffer(),
       send_frame_handler_->GetCurrentBufferSize());
   SendDataInternal(current_buffer_->data(), current_buffer_->BytesRemaining());
 }
 
 }  // namespace net