Add WebSocket over SPDY experimental implementation.

(Re-land with memory leak fix)

- Realize WebSocketJob's internal protocol switch to SPDY using SpdyWebSocketStream
- Add simple test to verify connection over SPDY

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

Committed: http://src.chromium.org/viewvc/chrome?view=rev&revision=91997

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

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

3 files changed, 362 insertions, 62 deletions

diff --git a/net/websockets/websocket_job.cc b/net/websockets/websocket_job.cc
index 77fc058..b627a07 100644
--- a/net/websockets/websocket_job.cc
+++ b/net/websockets/websocket_job.cc
@@ -24,6 +24,8 @@
 #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.
@@ -178,14 +180,14 @@ int WebSocketJob::OnStartOpenConnection(
   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(
@@ -276,10 +278,75 @@ void WebSocketJob::OnAuthRequired(
 }
 
 void WebSocketJob::OnError(const SocketStream* socket, int error) {
-  if (delegate_)
+  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_->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_)
@@ -317,14 +384,22 @@ void WebSocketJob::AddCookieHeaderAndSend() {
       }
     }
 
-    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());
+    }
   }
 }
 
@@ -452,27 +527,48 @@ 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.
-        }
-      }
-    }
+  if (!websocket_over_spdy_enabled_)
+    return OK;
+
+  // Check if we have a SPDY session available.
+  HttpTransactionFactory* factory =
+      socket_->context()->http_transaction_factory();
+  if (!factory)
+    return OK;
+  scoped_refptr<HttpNetworkSession> session = factory->GetSession();
+  if (!session.get())
+    return OK;
+  SpdySessionPool* spdy_pool = session->spdy_session_pool();
+  const HostPortProxyPair pair(HostPortPair::FromURL(socket_->url()),
+                               socket_->proxy_server());
+  if (!spdy_pool->HasSession(pair))
+    return OK;
+
+  // Forbid wss downgrade to SPDY without SSL.
+  // TODO(toyoshim): Does it realize the same policy with HTTP?
+  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);
+  if (socket_->is_secure() && !use_ssl)
+    return OK;
+
+  // Create SpdyWebSocketStream.
+  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;
   }
-  // No SPDY session was available.
-  // Fallback to connecting a new socket.
-  return OK;
+  if (result != ERR_IO_PENDING) {
+    spdy_websocket_stream_.reset();
+    return OK;
+  }
+
+  return ERR_IO_PENDING;
 }
 
 void WebSocketJob::SetWaiting() {
@@ -495,6 +591,14 @@ void WebSocketJob::Wakeup() {
 
 void WebSocketJob::RetryPendingIO() {
   int result = TrySpdyStream();
+
+  // In the case of ERR_IO_PENDING, CompleteIO() will be called from
+  // OnCreatedSpdyStream().
+  if (result != ERR_IO_PENDING)
+    CompleteIO(result);
+}
+
+void WebSocketJob::CompleteIO(int result) {
   // |callback_| may be NULL if OnClose() or DetachDelegate() was called.
   if (callback_) {
     net::CompletionCallback* callback = callback_;
@@ -505,12 +609,14 @@ void WebSocketJob::RetryPendingIO() {
 }
 
 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();
 }
 
diff --git a/net/websockets/websocket_job.h b/net/websockets/websocket_job.h
index 369ef56..f0e67c6 100644
--- a/net/websockets/websocket_job.h
+++ b/net/websockets/websocket_job.h
@@ -13,6 +13,7 @@
 #include "net/base/address_list.h"
 #include "net/base/completion_callback.h"
 #include "net/socket_stream/socket_stream_job.h"
+#include "net/spdy/spdy_websocket_stream.h"
 
 class GURL;
 
@@ -31,7 +32,8 @@ class WebSocketHandshakeResponseHandler;
 // TODO(ukai): refactor websocket.cc to use this.
 class NET_API WebSocketJob
     : public SocketStreamJob,
-      public SocketStream::Delegate {
+      public SocketStream::Delegate,
+      public SpdyWebSocketStream::Delegate {
  public:
   // This is state of WebSocket, not SocketStream.
   enum State {
@@ -70,6 +72,15 @@ class NET_API WebSocketJob
       SocketStream* socket, AuthChallengeInfo* auth_info);
   virtual void OnError(const SocketStream* socket, int error);
 
+  // SpdyWebSocketStream::Delegate methods.
+  virtual void OnCreatedSpdyStream(int status);
+  virtual void OnSentSpdyHeaders(int status);
+  virtual int OnReceivedSpdyResponseHeader(
+      const spdy::SpdyHeaderBlock& headers, int status);
+  virtual void OnSentSpdyData(int amount_sent);
+  virtual void OnReceivedSpdyData(const char* data, int length);
+  virtual void OnCloseSpdyStream();
+
  private:
   friend class WebSocketThrottle;
   friend class WebSocketJobTest;
@@ -92,6 +103,7 @@ class NET_API WebSocketJob
   bool IsWaiting() const;
   void Wakeup();
   void RetryPendingIO();
+  void CompleteIO(int result);
 
   bool SendDataInternal(const char* data, int length);
   void CloseInternal();
@@ -118,6 +130,9 @@ class NET_API WebSocketJob
   scoped_refptr<DrainableIOBuffer> current_buffer_;
   scoped_ptr<WebSocketFrameHandler> receive_frame_handler_;
 
+  scoped_ptr<SpdyWebSocketStream> spdy_websocket_stream_;
+  std::string challenge_;
+
   DISALLOW_COPY_AND_ASSIGN(WebSocketJob);
 };
 
diff --git a/net/websockets/websocket_job_unittest.cc b/net/websockets/websocket_job_unittest.cc
index e863faf..f36d771 100644
--- a/net/websockets/websocket_job_unittest.cc
+++ b/net/websockets/websocket_job_unittest.cc
@@ -19,9 +19,13 @@
 #include "net/base/sys_addrinfo.h"
 #include "net/base/test_completion_callback.h"
 #include "net/base/transport_security_state.h"
+#include "net/http/http_transaction_factory.h"
 #include "net/proxy/proxy_service.h"
 #include "net/socket/socket_test_util.h"
 #include "net/socket_stream/socket_stream.h"
+#include "net/spdy/spdy_session.h"
+#include "net/spdy/spdy_test_util.h"
+#include "net/spdy/spdy_websocket_test_util.h"
 #include "net/url_request/url_request_context.h"
 #include "net/websockets/websocket_throttle.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -198,6 +202,64 @@ class MockURLRequestContext : public net::URLRequestContext {
   scoped_refptr<net::TransportSecurityState> transport_security_state_;
 };
 
+class MockHttpTransactionFactory : public net::HttpTransactionFactory {
+ public:
+  MockHttpTransactionFactory(scoped_refptr<net::OrderedSocketData>& data) {
+    data_ = data;
+    net::MockConnect connect_data(false, net::OK);
+    data_->set_connect_data(connect_data);
+    session_deps_.reset(new net::SpdySessionDependencies);
+    session_deps_->socket_factory->AddSocketDataProvider(data_.get());
+    http_session_ =
+        net::SpdySessionDependencies::SpdyCreateSession(session_deps_.get());
+    host_port_pair_.set_host("example.com");
+    host_port_pair_.set_port(80);
+    host_port_proxy_pair_.first = host_port_pair_;
+    host_port_proxy_pair_.second = net::ProxyServer::Direct();
+    net::SpdySessionPool* spdy_session_pool =
+        http_session_->spdy_session_pool();
+    DCHECK(spdy_session_pool);
+    EXPECT_FALSE(spdy_session_pool->HasSession(host_port_proxy_pair_));
+    session_ =
+        spdy_session_pool->Get(host_port_proxy_pair_, net::BoundNetLog());
+    EXPECT_TRUE(spdy_session_pool->HasSession(host_port_proxy_pair_));
+
+    transport_params_ = new net::TransportSocketParams(host_port_pair_,
+                                                       net::MEDIUM,
+                                                       GURL(),
+                                                       false,
+                                                       false);
+    net::ClientSocketHandle* connection = new net::ClientSocketHandle;
+    EXPECT_EQ(net::OK, connection->Init(host_port_pair_.ToString(),
+                                        transport_params_,
+                                        net::MEDIUM,
+                                        NULL,
+                                        http_session_->transport_socket_pool(),
+                                        net::BoundNetLog()));
+    EXPECT_EQ(net::OK,
+              session_->InitializeWithSocket(connection, false, net::OK));
+  }
+  virtual int CreateTransaction(scoped_ptr<net::HttpTransaction>* trans) {
+    NOTREACHED();
+    return net::ERR_UNEXPECTED;
+  }
+  virtual net::HttpCache* GetCache() {
+    NOTREACHED();
+    return NULL;
+  }
+  virtual net::HttpNetworkSession* GetSession() {
+    return http_session_.get();
+  }
+ private:
+  scoped_refptr<net::OrderedSocketData> data_;
+  scoped_ptr<net::SpdySessionDependencies> session_deps_;
+  scoped_refptr<net::HttpNetworkSession> http_session_;
+  scoped_refptr<net::TransportSocketParams> transport_params_;
+  scoped_refptr<net::SpdySession> session_;
+  net::HostPortPair host_port_pair_;
+  net::HostPortProxyPair host_port_proxy_pair_;
+};
+
 }
 
 namespace net {
@@ -205,6 +267,7 @@ namespace net {
 class WebSocketJobTest : public PlatformTest {
  public:
   virtual void SetUp() {
+    spdy::SpdyFramer::set_enable_compression_default(false);
     stream_type_ = STREAM_INVALID;
     cookie_store_ = new MockCookieStore;
     context_ = new MockURLRequestContext(cookie_store_.get());
@@ -239,37 +302,33 @@ class WebSocketJobTest : public PlatformTest {
   void InitWebSocketJob(const GURL& url,
                         MockSocketStreamDelegate* delegate,
                         StreamType stream_type) {
+    DCHECK_NE(STREAM_INVALID, stream_type);
     stream_type_ = stream_type;
     websocket_ = new WebSocketJob(delegate);
 
-    if (stream_type == STREAM_SOCKET ||
-        stream_type == STREAM_SPDY_WEBSOCKET) {
+    if (stream_type == STREAM_MOCK_SOCKET)
+      socket_ = new MockSocketStream(url, websocket_.get());
+
+    if (stream_type == STREAM_SOCKET || stream_type == STREAM_SPDY_WEBSOCKET) {
+      if (stream_type == STREAM_SPDY_WEBSOCKET) {
+        http_factory_.reset(new MockHttpTransactionFactory(data_));
+        context_->set_http_transaction_factory(http_factory_.get());
+      }
+
       ssl_config_service_ = new MockSSLConfigService();
       context_->set_ssl_config_service(ssl_config_service_);
       proxy_service_.reset(net::ProxyService::CreateDirect());
       context_->set_proxy_service(proxy_service_.get());
       host_resolver_.reset(new net::MockHostResolver);
       context_->set_host_resolver(host_resolver_.get());
-    }
 
-    switch (stream_type) {
-      case STREAM_INVALID:
-        NOTREACHED();
-        break;
-      case STREAM_MOCK_SOCKET:
-        socket_ = new MockSocketStream(url, websocket_.get());
-        break;
-      case STREAM_SOCKET:
-        socket_ = new SocketStream(url, websocket_.get());
-        socket_factory_.reset(new MockClientSocketFactory);
-        DCHECK(data_.get());
-        socket_factory_->AddSocketDataProvider(data_.get());
-        socket_->SetClientSocketFactory(socket_factory_.get());
-        break;
-      case STREAM_SPDY_WEBSOCKET:
-        // TODO(toyoshim): Support SpdyWebSocketStream.
-        break;
+      socket_ = new SocketStream(url, websocket_.get());
+      socket_factory_.reset(new MockClientSocketFactory);
+      DCHECK(data_.get());
+      socket_factory_->AddSocketDataProvider(data_.get());
+      socket_->SetClientSocketFactory(socket_factory_.get());
     }
+
     websocket_->InitSocketStream(socket_.get());
     websocket_->set_context(context_.get());
     struct addrinfo addr;
@@ -324,6 +383,7 @@ class WebSocketJobTest : public PlatformTest {
   void TestHSTSUpgrade();
   void TestInvalidSendData();
   void TestConnectByWebSocket();
+  void TestConnectBySpdy(bool use_spdy);
 
   StreamType stream_type_;
   scoped_refptr<MockCookieStore> cookie_store_;
@@ -336,6 +396,7 @@ class WebSocketJobTest : public PlatformTest {
   scoped_refptr<MockSSLConfigService> ssl_config_service_;
   scoped_ptr<net::ProxyService> proxy_service_;
   scoped_ptr<net::MockHostResolver> host_resolver_;
+  scoped_ptr<MockHttpTransactionFactory> http_factory_;
 
   static const char kHandshakeRequestWithoutCookie[];
   static const char kHandshakeRequestWithCookie[];
@@ -344,6 +405,8 @@ class WebSocketJobTest : public PlatformTest {
   static const char kHandshakeResponseWithCookie[];
   static const char kDataHello[];
   static const char kDataWorld[];
+  static const char* const kHandshakeRequestForSpdy[];
+  static const char* const kHandshakeResponseForSpdy[];
   static const size_t kHandshakeRequestWithoutCookieLength;
   static const size_t kHandshakeRequestWithCookieLength;
   static const size_t kHandshakeRequestWithFilteredCookieLength;
@@ -416,6 +479,22 @@ const char WebSocketJobTest::kDataHello[] = "Hello, ";
 
 const char WebSocketJobTest::kDataWorld[] = "World!\n";
 
+// TODO(toyoshim): I should clarify which WebSocket headers for handshake must
+// be exported to SPDY SYN_STREAM and SYN_REPLY.
+// Because it depends on HyBi versions, just define it as follow for now.
+const char* const WebSocketJobTest::kHandshakeRequestForSpdy[] = {
+  "host", "example.com",
+  "origin", "http://example.com",
+  "sec-websocket-protocol", "sample",
+  "url", "ws://example.com/demo"
+};
+
+const char* const WebSocketJobTest::kHandshakeResponseForSpdy[] = {
+  "sec-websocket-origin", "http://example.com",
+  "sec-websocket-location", "ws://example.com/demo",
+  "sec-websocket-protocol", "sample",
+};
+
 const size_t WebSocketJobTest::kHandshakeRequestWithoutCookieLength =
     arraysize(kHandshakeRequestWithoutCookie) - 1;
 const size_t WebSocketJobTest::kHandshakeRequestWithCookieLength =
@@ -624,7 +703,6 @@ void WebSocketJobTest::TestInvalidSendData() {
 // OrderedSocketData provide socket level verifiation by checking out-going
 // packets in comparison with the MockWrite array and emulating in-coming
 // packets with MockRead array.
-// TODO(toyoshim): Add tests which verify protocol switch and ERR_IO_PENDING.
 
 void WebSocketJobTest::TestConnectByWebSocket() {
   // This is a test for verifying cooperation between WebSocketJob and
@@ -650,8 +728,8 @@ void WebSocketJobTest::TestConnectByWebSocket() {
              4),
     MockRead(false, 0, 5)  // EOF
   };
-  data_ = (new OrderedSocketData(
-      reads, arraysize(reads), writes, arraysize(writes)));
+  data_ = new OrderedSocketData(
+      reads, arraysize(reads), writes, arraysize(writes));
 
   GURL url("ws://example.com/demo");
   MockSocketStreamDelegate delegate;
@@ -671,6 +749,94 @@ void WebSocketJobTest::TestConnectByWebSocket() {
   EXPECT_EQ(WebSocketJob::CLOSED, GetWebSocketJobState());
 }
 
+void WebSocketJobTest::TestConnectBySpdy(bool use_spdy) {
+  // This is a test for verifying cooperation between WebSocketJob and
+  // SocketStream in the situation we have SPDY session to the server.
+  MockWrite writes_websocket[] = {
+    MockWrite(true,
+              kHandshakeRequestWithoutCookie,
+              kHandshakeRequestWithoutCookieLength,
+              1),
+    MockWrite(true,
+              kDataHello,
+              kDataHelloLength,
+              3)
+  };
+  MockRead reads_websocket[] = {
+    MockRead(true,
+             kHandshakeResponseWithoutCookie,
+             kHandshakeResponseWithoutCookieLength,
+             2),
+    MockRead(true,
+             kDataWorld,
+             kDataWorldLength,
+             4),
+    MockRead(false, 0, 5)  // EOF
+  };
+
+  const spdy::SpdyStreamId kStreamId = 1;
+  scoped_ptr<spdy::SpdyFrame> request_frame(
+      ConstructSpdyWebSocketHandshakeRequestFrame(
+          kHandshakeRequestForSpdy,
+          arraysize(kHandshakeRequestForSpdy) / 2,
+          kStreamId,
+          MEDIUM));
+  scoped_ptr<spdy::SpdyFrame> response_frame(
+      ConstructSpdyWebSocketHandshakeResponseFrame(
+          kHandshakeResponseForSpdy,
+          arraysize(kHandshakeResponseForSpdy) / 2,
+          kStreamId,
+          MEDIUM));
+  scoped_ptr<spdy::SpdyFrame> data_hello_frame(
+      ConstructSpdyWebSocketDataFrame(
+          kDataHello,
+          kDataHelloLength,
+          kStreamId,
+          false));
+  scoped_ptr<spdy::SpdyFrame> data_world_frame(
+      ConstructSpdyWebSocketDataFrame(
+          kDataWorld,
+          kDataWorldLength,
+          kStreamId,
+          false));
+  MockWrite writes_spdy[] = {
+    CreateMockWrite(*request_frame.get(), 1),
+    CreateMockWrite(*data_hello_frame.get(), 3),
+  };
+  MockRead reads_spdy[] = {
+    CreateMockRead(*response_frame.get(), 2),
+    CreateMockRead(*data_world_frame.get(), 4),
+    MockRead(false, 0, 5)  // EOF
+  };
+
+  if (use_spdy)
+    data_ = new OrderedSocketData(
+        reads_spdy, arraysize(reads_spdy),
+        writes_spdy, arraysize(writes_spdy));
+  else
+    data_ = new OrderedSocketData(
+        reads_websocket, arraysize(reads_websocket),
+        writes_websocket, arraysize(writes_websocket));
+
+  GURL url("ws://example.com/demo");
+  MockSocketStreamDelegate delegate;
+  WebSocketJobTest* test = this;
+  delegate.SetOnConnected(
+      NewCallback(test, &WebSocketJobTest::DoSendRequest));
+  delegate.SetOnReceivedData(
+      NewCallback(test, &WebSocketJobTest::DoSendData));
+  delegate.SetOnClose(
+      NewCallback(test, &WebSocketJobTest::DoSync));
+  InitWebSocketJob(url, &delegate, STREAM_SPDY_WEBSOCKET);
+
+  websocket_->Connect();
+  EXPECT_EQ(OK, WaitForResult());
+
+  EXPECT_TRUE(data_->at_read_eof());
+  EXPECT_TRUE(data_->at_write_eof());
+  EXPECT_EQ(WebSocketJob::CLOSED, GetWebSocketJobState());
+}
+
 // Execute tests in both spdy-disabled mode and spdy-enabled mode.
 TEST_F(WebSocketJobTest, SimpleHandshake) {
   WebSocketJob::set_websocket_over_spdy_enabled(false);
@@ -742,4 +908,17 @@ TEST_F(WebSocketJobTest, ConnectByWebSocketSpdyEnabled) {
   TestConnectByWebSocket();
 }
 
+TEST_F(WebSocketJobTest, ConnectBySpdy) {
+  WebSocketJob::set_websocket_over_spdy_enabled(false);
+  TestConnectBySpdy(false);
+}
+
+TEST_F(WebSocketJobTest, ConnectBySpdySpdyEnabled) {
+  WebSocketJob::set_websocket_over_spdy_enabled(true);
+  TestConnectBySpdy(true);
+}
+
+// TODO(toyoshim): Add tests to verify throttling, SPDY stream limitation.
+// TODO(toyoshim,yutak): Add tests to verify closing handshake.
+
 }  // namespace net