WebSocket over SPDY: protocol switch

This change enable to switch protocol from WebSocket to SPDY.
SpdyWebsocketStream class is not included, then SPDY connection
always fail now.

BUG=42320
TEST=net_unittests --gtest_filter=WebSocket\*

Review URL: http://codereview.chromium.org/7062043

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@89318 0039d316-1c4b-4281-b951-d872f2087c98

4 files changed, 91 insertions, 23 deletions

diff --git a/net/websockets/websocket_job.cc b/net/websockets/websocket_job.cc
index fc67f9a..77fc058 100644
--- a/net/websockets/websocket_job.cc
+++ b/net/websockets/websocket_job.cc
@@ -13,7 +13,11 @@
 #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"
@@ -110,7 +114,8 @@ bool WebSocketJob::SendData(const char* data, int len) {
         // pending bytes less than max_pending_send_allowed, so when sending
         // larger message than max_pending_send_allowed should not be buffered.
         // If we don't call OnSentData, WebCore::SocketStreamHandle would stop
-        // sending more data when pending data reaches max_pending_send_allowed.
+        // sending more data when pending data reaches
+        // max_pending_send_allowed.
         // TODO(ukai): Fix this to support compression for larger message.
         int err = 0;
         if (!send_frame_handler_->GetCurrentBuffer() &&
@@ -119,7 +124,7 @@ bool WebSocketJob::SendData(const char* data, int len) {
           current_buffer_ = new DrainableIOBuffer(
               send_frame_handler_->GetCurrentBuffer(),
               send_frame_handler_->GetCurrentBufferSize());
-          return socket_->SendData(
+          return SendDataInternal(
               current_buffer_->data(), current_buffer_->BytesRemaining());
         }
         return err >= 0;
@@ -139,7 +144,7 @@ void WebSocketJob::Close() {
     return;
   }
   state_ = CLOSED;
-  socket_->Close();
+  CloseInternal();
 }
 
 void WebSocketJob::RestartWithAuth(
@@ -173,11 +178,14 @@ int WebSocketJob::OnStartOpenConnection(
   state_ = CONNECTING;
   addresses_ = socket->address_list();
   WebSocketThrottle::GetInstance()->PutInQueue(this);
-  if (!waiting_)
-    return OK;
+  if (!waiting_) {
+    int result = TrySpdyStream();
+    if (result != ERR_IO_PENDING)
+      return result;
+  }
   callback_ = callback;
   AddRef();  // Balanced when callback_ becomes NULL.
-  return ERR_IO_PENDING;
+  return ERR_IO_PENDING;  // Wakeup will be called later.
 }
 
 void WebSocketJob::OnConnected(
@@ -440,6 +448,33 @@ const AddressList& WebSocketJob::address_list() const {
   return addresses_;
 }
 
+int WebSocketJob::TrySpdyStream() {
+  if (!socket_.get())
+    return ERR_FAILED;
+
+  if (websocket_over_spdy_enabled_) {
+    // Check if we have a SPDY session available.
+    // 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.
+        }
+      }
+    }
+  }
+  // No SPDY session was available.
+  // Fallback to connecting a new socket.
+  return OK;
+}
+
 void WebSocketJob::SetWaiting() {
   waiting_ = true;
 }
@@ -455,20 +490,30 @@ void WebSocketJob::Wakeup() {
   DCHECK(callback_);
   MessageLoopForIO::current()->PostTask(
       FROM_HERE,
-      NewRunnableMethod(this,
-                        &WebSocketJob::DoCallback));
+      NewRunnableMethod(this, &WebSocketJob::RetryPendingIO));
 }
 
-void WebSocketJob::DoCallback() {
+void WebSocketJob::RetryPendingIO() {
+  int result = TrySpdyStream();
   // |callback_| may be NULL if OnClose() or DetachDelegate() was called.
   if (callback_) {
     net::CompletionCallback* callback = callback_;
     callback_ = NULL;
-    callback->Run(net::OK);
+    callback->Run(result);
     Release();  // Balanced with OnStartOpenConnection().
   }
 }
 
+bool WebSocketJob::SendDataInternal(const char* data, int length) {
+  // TODO(toyoshim): Call protocol specific SendData().
+  return socket_->SendData(data, length);
+}
+
+void WebSocketJob::CloseInternal() {
+  // TODO(toyoshim): Call protocol specific Close().
+  socket_->Close();
+}
+
 void WebSocketJob::SendPending() {
   if (current_buffer_)
     return;
@@ -477,13 +522,13 @@ void WebSocketJob::SendPending() {
   if (send_frame_handler_->UpdateCurrentBuffer(false) <= 0) {
     // No more data to send.
     if (state_ == CLOSING)
-      socket_->Close();
+      CloseInternal();
     return;
   }
   current_buffer_ = new DrainableIOBuffer(
       send_frame_handler_->GetCurrentBuffer(),
       send_frame_handler_->GetCurrentBufferSize());
-  socket_->SendData(current_buffer_->data(), current_buffer_->BytesRemaining());
+  SendDataInternal(current_buffer_->data(), current_buffer_->BytesRemaining());
 }
 
 }  // namespace net
diff --git a/net/websockets/websocket_job.h b/net/websockets/websocket_job.h
index b554469..369ef56 100644
--- a/net/websockets/websocket_job.h
+++ b/net/websockets/websocket_job.h
@@ -62,17 +62,13 @@ class NET_API WebSocketJob
   // SocketStream::Delegate methods.
   virtual int OnStartOpenConnection(
       SocketStream* socket, CompletionCallback* callback);
-  virtual void OnConnected(
-      SocketStream* socket, int max_pending_send_allowed);
-  virtual void OnSentData(
-      SocketStream* socket, int amount_sent);
-  virtual void OnReceivedData(
-      SocketStream* socket, const char* data, int len);
+  virtual void OnConnected(SocketStream* socket, int max_pending_send_allowed);
+  virtual void OnSentData(SocketStream* socket, int amount_sent);
+  virtual void OnReceivedData(SocketStream* socket, const char* data, int len);
   virtual void OnClose(SocketStream* socket);
   virtual void OnAuthRequired(
       SocketStream* socket, AuthChallengeInfo* auth_info);
-  virtual void OnError(
-      const SocketStream* socket, int error);
+  virtual void OnError(const SocketStream* socket, int error);
 
  private:
   friend class WebSocketThrottle;
@@ -91,11 +87,14 @@ class NET_API WebSocketJob
   GURL GetURLForCookies() const;
 
   const AddressList& address_list() const;
+  int TrySpdyStream();
   void SetWaiting();
   bool IsWaiting() const;
   void Wakeup();
-  void DoCallback();
+  void RetryPendingIO();
 
+  bool SendDataInternal(const char* data, int length);
+  void CloseInternal();
   void SendPending();
 
   static bool websocket_over_spdy_enabled_;
diff --git a/net/websockets/websocket_job_unittest.cc b/net/websockets/websocket_job_unittest.cc
index 8951a4e..b19fb37 100644
--- a/net/websockets/websocket_job_unittest.cc
+++ b/net/websockets/websocket_job_unittest.cc
@@ -59,7 +59,8 @@ class MockSocketStreamDelegate : public SocketStream::Delegate {
   }
   virtual ~MockSocketStreamDelegate() {}
 
-  virtual void OnConnected(SocketStream* socket, int max_pending_send_allowed) {
+  virtual void OnConnected(SocketStream* socket,
+                           int max_pending_send_allowed) {
   }
   virtual void OnSentData(SocketStream* socket, int amount_sent) {
     amount_sent_ += amount_sent;
@@ -278,6 +279,9 @@ const char* WebSocketJobTest::kHandshakeResponseWithCookie =
     "8jKS'y:G*Co,Wxa-";
 
 TEST_F(WebSocketJobTest, SimpleHandshake) {
+  // TODO(toyoshim): We need to consider both spdy-enabled and spdy-disabled
+  // configuration.
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
   GURL url("ws://example.com/demo");
   MockSocketStreamDelegate delegate;
   InitWebSocketJob(url, &delegate);
@@ -302,6 +306,7 @@ TEST_F(WebSocketJobTest, SimpleHandshake) {
 }
 
 TEST_F(WebSocketJobTest, SlowHandshake) {
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
   GURL url("ws://example.com/demo");
   MockSocketStreamDelegate delegate;
   InitWebSocketJob(url, &delegate);
@@ -341,6 +346,7 @@ TEST_F(WebSocketJobTest, SlowHandshake) {
 }
 
 TEST_F(WebSocketJobTest, HandshakeWithCookie) {
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
   GURL url("ws://example.com/demo");
   GURL cookieUrl("http://example.com/demo");
   CookieOptions cookie_options;
@@ -382,6 +388,7 @@ TEST_F(WebSocketJobTest, HandshakeWithCookie) {
 }
 
 TEST_F(WebSocketJobTest, HandshakeWithCookieButNotAllowed) {
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
   GURL url("ws://example.com/demo");
   GURL cookieUrl("http://example.com/demo");
   CookieOptions cookie_options;
@@ -422,6 +429,7 @@ TEST_F(WebSocketJobTest, HandshakeWithCookieButNotAllowed) {
 }
 
 TEST_F(WebSocketJobTest, HSTSUpgrade) {
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
   GURL url("ws://upgrademe.com/");
   MockSocketStreamDelegate delegate;
   scoped_refptr<SocketStreamJob> job = SocketStreamJob::CreateSocketStreamJob(
diff --git a/net/websockets/websocket_throttle_unittest.cc b/net/websockets/websocket_throttle_unittest.cc
index 7eb18fa5..e89b2c1 100644
--- a/net/websockets/websocket_throttle_unittest.cc
+++ b/net/websockets/websocket_throttle_unittest.cc
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -10,6 +10,7 @@
 #include "net/base/sys_addrinfo.h"
 #include "net/base/test_completion_callback.h"
 #include "net/socket_stream/socket_stream.h"
+#include "net/url_request/url_request_test_util.h"
 #include "net/websockets/websocket_job.h"
 #include "net/websockets/websocket_throttle.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -64,6 +65,8 @@ class WebSocketThrottleTest : public PlatformTest {
   static void MockSocketStreamConnect(
       SocketStream* socket, struct addrinfo* head) {
     socket->CopyAddrInfo(head);
+    // TODO(toyoshim): We should introduce additional tests on cases via proxy.
+    socket->proxy_info_.UseDirect();
     // In SocketStream::Connect(), it adds reference to socket, which is
     // balanced with SocketStream::Finish() that is finally called from
     // SocketStream::Close() or SocketStream::DetachDelegate(), when
@@ -77,7 +80,11 @@ class WebSocketThrottleTest : public PlatformTest {
 };
 
 TEST_F(WebSocketThrottleTest, Throttle) {
+  scoped_refptr<URLRequestContext> context(new TestURLRequestContext);
   DummySocketStreamDelegate delegate;
+  // TODO(toyoshim): We need to consider both spdy-enabled and spdy-disabled
+  // configuration.
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
 
   // For host1: 1.2.3.4, 1.2.3.5, 1.2.3.6
   struct addrinfo* addr = AddAddr(1, 2, 3, 4, NULL);
@@ -86,6 +93,7 @@ TEST_F(WebSocketThrottleTest, Throttle) {
   scoped_refptr<WebSocketJob> w1(new WebSocketJob(&delegate));
   scoped_refptr<SocketStream> s1(
       new SocketStream(GURL("ws://host1/"), w1.get()));
+  s1->set_context(context.get());
   w1->InitSocketStream(s1.get());
   WebSocketThrottleTest::MockSocketStreamConnect(s1, addr);
   DeleteAddrInfo(addr);
@@ -106,6 +114,7 @@ TEST_F(WebSocketThrottleTest, Throttle) {
   scoped_refptr<WebSocketJob> w2(new WebSocketJob(&delegate));
   scoped_refptr<SocketStream> s2(
       new SocketStream(GURL("ws://host2/"), w2.get()));
+  s2->set_context(context.get());
   w2->InitSocketStream(s2.get());
   WebSocketThrottleTest::MockSocketStreamConnect(s2, addr);
   DeleteAddrInfo(addr);
@@ -125,6 +134,7 @@ TEST_F(WebSocketThrottleTest, Throttle) {
   scoped_refptr<WebSocketJob> w3(new WebSocketJob(&delegate));
   scoped_refptr<SocketStream> s3(
       new SocketStream(GURL("ws://host3/"), w3.get()));
+  s3->set_context(context.get());
   w3->InitSocketStream(s3.get());
   WebSocketThrottleTest::MockSocketStreamConnect(s3, addr);
   DeleteAddrInfo(addr);
@@ -144,6 +154,7 @@ TEST_F(WebSocketThrottleTest, Throttle) {
   scoped_refptr<WebSocketJob> w4(new WebSocketJob(&delegate));
   scoped_refptr<SocketStream> s4(
       new SocketStream(GURL("ws://host4/"), w4.get()));
+  s4->set_context(context.get());
   w4->InitSocketStream(s4.get());
   WebSocketThrottleTest::MockSocketStreamConnect(s4, addr);
   DeleteAddrInfo(addr);
@@ -162,6 +173,7 @@ TEST_F(WebSocketThrottleTest, Throttle) {
   scoped_refptr<WebSocketJob> w5(new WebSocketJob(&delegate));
   scoped_refptr<SocketStream> s5(
       new SocketStream(GURL("ws://host5/"), w5.get()));
+  s5->set_context(context.get());
   w5->InitSocketStream(s5.get());
   WebSocketThrottleTest::MockSocketStreamConnect(s5, addr);
   DeleteAddrInfo(addr);
@@ -180,6 +192,7 @@ TEST_F(WebSocketThrottleTest, Throttle) {
   scoped_refptr<WebSocketJob> w6(new WebSocketJob(&delegate));
   scoped_refptr<SocketStream> s6(
       new SocketStream(GURL("ws://host6/"), w6.get()));
+  s6->set_context(context.get());
   w6->InitSocketStream(s6.get());
   WebSocketThrottleTest::MockSocketStreamConnect(s6, addr);
   DeleteAddrInfo(addr);
@@ -281,7 +294,9 @@ TEST_F(WebSocketThrottleTest, Throttle) {
 }
 
 TEST_F(WebSocketThrottleTest, NoThrottleForDuplicateAddress) {
+  scoped_refptr<URLRequestContext> context(new TestURLRequestContext);
   DummySocketStreamDelegate delegate;
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
 
   // For localhost: 127.0.0.1, 127.0.0.1
   struct addrinfo* addr = AddAddr(127, 0, 0, 1, NULL);
@@ -289,6 +304,7 @@ TEST_F(WebSocketThrottleTest, NoThrottleForDuplicateAddress) {
   scoped_refptr<WebSocketJob> w1(new WebSocketJob(&delegate));
   scoped_refptr<SocketStream> s1(
       new SocketStream(GURL("ws://localhost/"), w1.get()));
+  s1->set_context(context.get());
   w1->InitSocketStream(s1.get());
   WebSocketThrottleTest::MockSocketStreamConnect(s1, addr);
   DeleteAddrInfo(addr);