Refactor SpdyStream state for WebSocket support

In HTTP, it doesn't use STATE_READ_BODY/STATE_READ_BODY_COMPLETE states. Reading body message is handled in OnDataReceived() and DoLoop() is not involved in this state.

In WebSocket, it will send frame after handshake has been finished, and need to get how many data has been written.
STATE_OPEN handles this.

BUG=none
TEST=none

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

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

8 files changed, 359 insertions, 51 deletions

diff --git a/net/spdy/spdy_framer.h b/net/spdy/spdy_framer.h
index a0ea64d..1122474 100644
--- a/net/spdy/spdy_framer.h
+++ b/net/spdy/spdy_framer.h
@@ -28,6 +28,7 @@ class HttpNetworkLayer;
 class HttpNetworkTransactionTest;
 class SpdyNetworkTransactionTest;
 class SpdySessionTest;
+class SpdyStreamTest;
 }
 
 namespace spdy {
@@ -248,6 +249,7 @@ class SpdyFramer {
   friend class net::HttpNetworkTransactionTest;
   friend class net::HttpNetworkLayer;  // This is temporary for the server.
   friend class net::SpdySessionTest;
+  friend class net::SpdyStreamTest;
   friend class test::TestSpdyVisitor;
   friend void test::FramerSetEnableCompressionHelper(SpdyFramer* framer,
                                                      bool compress);
diff --git a/net/spdy/spdy_http_stream.cc b/net/spdy/spdy_http_stream.cc
index 68e5e2a..ffc8330 100644
--- a/net/spdy/spdy_http_stream.cc
+++ b/net/spdy/spdy_http_stream.cc
@@ -317,7 +317,8 @@ int SpdyHttpStream::OnSendBody() {
   int buf_len = static_cast<int>(request_body_stream_->buf_len());
   if (!buf_len)
     return OK;
-  return stream_->WriteStreamData(request_body_stream_->buf(), buf_len);
+  return stream_->WriteStreamData(request_body_stream_->buf(), buf_len,
+                                  spdy::DATA_FLAG_FIN);
 }
 
 bool SpdyHttpStream::OnSendBodyComplete(int status) {
@@ -370,6 +371,12 @@ void SpdyHttpStream::OnDataReceived(const char* data, int length) {
   }
 }
 
+void SpdyHttpStream::OnDataSent(int length) {
+  // For HTTP streams, no data is sent from the client while in the OPEN state,
+  // so it is never called.
+  NOTREACHED();
+}
+
 void SpdyHttpStream::OnClose(int status) {
   bool invoked_callback = false;
   if (status == net::OK) {
diff --git a/net/spdy/spdy_http_stream.h b/net/spdy/spdy_http_stream.h
index a16c3ca..3926501 100644
--- a/net/spdy/spdy_http_stream.h
+++ b/net/spdy/spdy_http_stream.h
@@ -90,6 +90,10 @@ class SpdyHttpStream : public SpdyStream::Delegate {
   // SpdyHttpSession schedule to call back |callback| set by ReadResponseBody.
   virtual void OnDataReceived(const char* buffer, int bytes);
 
+  // For HTTP streams, no data is sent from the client while in the OPEN state,
+  // so OnDataSent is never called.
+  virtual void OnDataSent(int length);
+
   // Called by the SpdySession when the request is finished.  This callback
   // will always be called at the end of the request and signals to the
   // stream that the stream has no more network events.  No further callbacks
diff --git a/net/spdy/spdy_session.cc b/net/spdy/spdy_session.cc
index b938f53..bbbf9ba 100644
--- a/net/spdy/spdy_session.cc
+++ b/net/spdy/spdy_session.cc
@@ -420,7 +420,8 @@ int SpdySession::WriteSynStream(
 }
 
 int SpdySession::WriteStreamData(spdy::SpdyStreamId stream_id,
-                                 net::IOBuffer* data, int len) {
+                                 net::IOBuffer* data, int len,
+                                 spdy::SpdyDataFlags flags) {
   LOG(INFO) << "Writing Stream Data for stream " << stream_id << " (" << len
             << " bytes)";
   const int kMss = 1430;  // This is somewhat arbitrary and not really fixed,
@@ -437,11 +438,6 @@ int SpdySession::WriteStreamData(spdy::SpdyStreamId stream_id,
   if (!stream)
     return ERR_INVALID_SPDY_STREAM;
 
-  // TODO(mbelshe):  Setting of the FIN is assuming that the caller will pass
-  //                 all data to write in a single chunk.  Is this always true?
-
-  // Set the flags on the upload.
-  spdy::SpdyDataFlags flags = spdy::DATA_FLAG_FIN;
   if (len > kMaxSpdyFrameChunkSize) {
     len = kMaxSpdyFrameChunkSize;
     flags = spdy::DATA_FLAG_NONE;
diff --git a/net/spdy/spdy_session.h b/net/spdy/spdy_session.h
index b306831..00ac0a8 100644
--- a/net/spdy/spdy_session.h
+++ b/net/spdy/spdy_session.h
@@ -95,7 +95,8 @@ class SpdySession : public base::RefCounted<SpdySession>,
   // Write a data frame to the stream.
   // Used to create and queue a data frame for the given stream.
   int WriteStreamData(spdy::SpdyStreamId stream_id, net::IOBuffer* data,
-                      int len);
+                      int len,
+                      spdy::SpdyDataFlags flags);
 
   // Close a stream.
   void CloseStream(spdy::SpdyStreamId stream_id, int status);
diff --git a/net/spdy/spdy_stream.cc b/net/spdy/spdy_stream.cc
index 3d68d5d..7218699 100644
--- a/net/spdy/spdy_stream.cc
+++ b/net/spdy/spdy_stream.cc
@@ -82,6 +82,7 @@ void SpdyStream::SetRequestTime(base::Time t) {
 int SpdyStream::OnResponseReceived(const spdy::SpdyHeaderBlock& response) {
   int rv = OK;
   LOG(INFO) << "OnResponseReceived";
+  DCHECK_NE(io_state_, STATE_OPEN);
 
   metrics_.StartStream();
 
@@ -204,34 +205,45 @@ int SpdyStream::DoSendRequest(bool has_upload_data) {
 
   send_time_ = base::TimeTicks::Now();
 
-  DCHECK_EQ(io_state_, STATE_NONE);
-  if (!pushed_)
+  int result = OK;
+  if (!pushed_) {
+    DCHECK_EQ(io_state_, STATE_NONE);
     io_state_ = STATE_SEND_HEADERS;
-  else {
+  } else {
+    DCHECK(!has_upload_data);
     if (!response_->empty()) {
-      io_state_ = STATE_READ_BODY;
+      // We already have response headers, so we don't need to read the header.
+      // Pushed stream should not have upload data.
+      // We don't need to call DoLoop() in this state.
+      DCHECK_EQ(io_state_, STATE_OPEN);
+      return OK;
     } else {
       io_state_ = STATE_READ_HEADERS;
     }
   }
-  return DoLoop(OK);
+  return DoLoop(result);
 }
 
 int SpdyStream::DoReadResponseHeaders() {
-  CHECK_EQ(STATE_NONE, io_state_);
   CHECK(!cancelled_);
 
   // The SYN_REPLY has already been received.
-  if (!response_->empty())
+  if (!response_->empty()) {
+    CHECK_EQ(STATE_OPEN, io_state_);
     return OK;
+  } else {
+    CHECK_EQ(STATE_NONE, io_state_);
+  }
+
 
   io_state_ = STATE_READ_HEADERS;
   // Q: do we need to run DoLoop here?
   return ERR_IO_PENDING;
 }
 
-int SpdyStream::WriteStreamData(IOBuffer* data, int length) {
-  return session_->WriteStreamData(stream_id_, data, length);
+int SpdyStream::WriteStreamData(IOBuffer* data, int length,
+                                spdy::SpdyDataFlags flags) {
+  return session_->WriteStreamData(stream_id_, data, length, flags);
 }
 
 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info, bool* was_npn_negotiated) {
@@ -272,26 +284,33 @@ int SpdyStream::DoLoop(int result) {
         result = DoReadHeadersComplete(result);
         break;
 
-      // State machine 2: Read body.
-      // NOTE(willchan): Currently unused.  Currently we handle this stuff in
-      // the OnDataReceived()/OnClose()/ReadResponseHeaders()/etc.  Only reason
-      // to do this is for consistency with the Http code.
-      case STATE_READ_BODY:
-        net_log_.BeginEvent(NetLog::TYPE_SPDY_STREAM_READ_BODY, NULL);
-        result = DoReadBody();
-        break;
-      case STATE_READ_BODY_COMPLETE:
-        net_log_.EndEvent(NetLog::TYPE_SPDY_STREAM_READ_BODY, NULL);
-        result = DoReadBodyComplete(result);
+      // State machine 2: connection is established.
+      // In STATE_OPEN, OnResponseReceived has already been called.
+      // OnDataReceived, OnClose and OnWriteCompelte can be called.
+      // Only OnWriteCompletee calls DoLoop(().
+      //
+      // For HTTP streams, no data is sent from the client while in the OPEN
+      // state, so OnWriteComplete is never called here.  The HTTP body is
+      // handled in the OnDataReceived callback, which does not call into
+      // DoLoop.
+      //
+      // For WebSocket streams, which are bi-directional, we'll send and
+      // receive data once the connection is established.  Received data is
+      // handled in OnDataReceived.  Sent data is handled in OnWriteComplete,
+      // which calls DoOpen().
+      case STATE_OPEN:
+        result = DoOpen(result);
         break;
+
       case STATE_DONE:
         DCHECK(result != ERR_IO_PENDING);
         break;
       default:
-        NOTREACHED();
+        NOTREACHED() << io_state_;
         break;
     }
-  } while (result != ERR_IO_PENDING && io_state_ != STATE_NONE);
+  } while (result != ERR_IO_PENDING && io_state_ != STATE_NONE &&
+           io_state_ != STATE_OPEN);
 
   return result;
 }
@@ -363,23 +382,15 @@ int SpdyStream::DoReadHeaders() {
 }
 
 int SpdyStream::DoReadHeadersComplete(int result) {
+  io_state_ = STATE_OPEN;
   return result;
 }
 
-int SpdyStream::DoReadBody() {
-  // TODO(mbelshe): merge SpdyStreamParser with SpdyStream and then this
-  // makes sense.
-  if (response_complete_) {
-    io_state_ = STATE_READ_BODY_COMPLETE;
-    return OK;
-  }
-  return ERR_IO_PENDING;
-}
-
-int SpdyStream::DoReadBodyComplete(int result) {
-  // TODO(mbelshe): merge SpdyStreamParser with SpdyStream and then this
-  // makes sense.
-  return OK;
+int SpdyStream::DoOpen(int result) {
+  if (delegate_)
+    delegate_->OnDataSent(result);
+  io_state_ = STATE_OPEN;
+  return result;
 }
 
 void SpdyStream::UpdateHistograms() {
diff --git a/net/spdy/spdy_stream.h b/net/spdy/spdy_stream.h
index 6183be2..bcb38c7 100644
--- a/net/spdy/spdy_stream.h
+++ b/net/spdy/spdy_stream.h
@@ -57,9 +57,11 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
                                    int status) = 0;
 
     // Called when data is received.
-    // Returns true if data is successfully processed.
     virtual void OnDataReceived(const char* data, int length) = 0;
 
+    // Called when data is sent.
+    virtual void OnDataSent(int length) = 0;
+
     // Called when SpdyStream is closed.
     virtual void OnClose(int status) = 0;
 
@@ -152,11 +154,14 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
   int DoReadResponseHeaders();
 
   // Sends DATA frame.
-  int WriteStreamData(IOBuffer* data, int length);
+  int WriteStreamData(IOBuffer* data, int length,
+                      spdy::SpdyDataFlags flags);
 
   bool GetSSLInfo(SSLInfo* ssl_info, bool* was_npn_negotiated);
 
-  bool is_idle() const { return io_state_ == STATE_NONE; }
+  bool is_idle() const {
+    return io_state_ == STATE_NONE || io_state_ == STATE_OPEN;
+  }
   bool response_complete() const { return response_complete_; }
   int response_status() const { return response_status_; }
 
@@ -169,8 +174,7 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
     STATE_SEND_BODY_COMPLETE,
     STATE_READ_HEADERS,
     STATE_READ_HEADERS_COMPLETE,
-    STATE_READ_BODY,
-    STATE_READ_BODY_COMPLETE,
+    STATE_OPEN,
     STATE_DONE
   };
 
@@ -187,8 +191,7 @@ class SpdyStream : public base::RefCounted<SpdyStream> {
   int DoSendBodyComplete(int result);
   int DoReadHeaders();
   int DoReadHeadersComplete(int result);
-  int DoReadBody();
-  int DoReadBodyComplete(int result);
+  int DoOpen(int result);
 
   // Update the histograms.  Can safely be called repeatedly, but should only
   // be called after the stream has completed.
diff --git a/net/spdy/spdy_stream_unittest.cc b/net/spdy/spdy_stream_unittest.cc
new file mode 100644
index 0000000..4105aa6
--- /dev/null
+++ b/net/spdy/spdy_stream_unittest.cc
@@ -0,0 +1,284 @@
+// Copyright (c) 2010 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/spdy/spdy_stream.h"
+#include "base/ref_counted.h"
+#include "base/time.h"
+#include "net/base/mock_host_resolver.h"
+#include "net/base/net_errors.h"
+#include "net/base/net_log.h"
+#include "net/base/ssl_config_service.h"
+#include "net/base/ssl_config_service_defaults.h"
+#include "net/base/test_completion_callback.h"
+#include "net/http/http_auth_handler_factory.h"
+#include "net/http/http_network_session.h"
+#include "net/http/http_request_info.h"
+#include "net/http/http_response_info.h"
+#include "net/proxy/proxy_service.h"
+#include "net/socket/socket_test_util.h"
+#include "net/spdy/spdy_framer.h"
+#include "net/spdy/spdy_session.h"
+#include "net/spdy/spdy_session_pool.h"
+#include "net/spdy/spdy_test_util.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace net {
+
+// TODO(ukai): factor out common part with spdy_http_stream_unittest.cc
+class SpdySessionPoolPeer {
+ public:
+  explicit SpdySessionPoolPeer(const scoped_refptr<SpdySessionPool>& pool)
+      : pool_(pool) {}
+
+  void RemoveSpdySession(const scoped_refptr<SpdySession>& session) {
+    pool_->Remove(session);
+  }
+
+ private:
+  const scoped_refptr<SpdySessionPool> pool_;
+
+  DISALLOW_COPY_AND_ASSIGN(SpdySessionPoolPeer);
+};
+
+namespace {
+
+// Create a proxy service which fails on all requests (falls back to direct).
+ProxyService* CreateNullProxyService() {
+  return ProxyService::CreateNull();
+}
+
+// Helper to manage the lifetimes of the dependencies for a
+// SpdyNetworkTransaction.
+class SessionDependencies {
+ public:
+  // Default set of dependencies -- "null" proxy service.
+  SessionDependencies()
+      : host_resolver(new MockHostResolver),
+        proxy_service(CreateNullProxyService()),
+        ssl_config_service(new SSLConfigServiceDefaults),
+        http_auth_handler_factory(HttpAuthHandlerFactory::CreateDefault()),
+        spdy_session_pool(new SpdySessionPool()) {}
+
+  // Custom proxy service dependency.
+  explicit SessionDependencies(ProxyService* proxy_service)
+      : host_resolver(new MockHostResolver),
+        proxy_service(proxy_service),
+        ssl_config_service(new SSLConfigServiceDefaults),
+        http_auth_handler_factory(HttpAuthHandlerFactory::CreateDefault()),
+        spdy_session_pool(new SpdySessionPool()) {}
+
+  scoped_refptr<MockHostResolverBase> host_resolver;
+  scoped_refptr<ProxyService> proxy_service;
+  scoped_refptr<SSLConfigService> ssl_config_service;
+  MockClientSocketFactory socket_factory;
+  scoped_ptr<HttpAuthHandlerFactory> http_auth_handler_factory;
+  scoped_refptr<SpdySessionPool> spdy_session_pool;
+};
+
+HttpNetworkSession* CreateSession(SessionDependencies* session_deps) {
+  return new HttpNetworkSession(session_deps->host_resolver,
+                                session_deps->proxy_service,
+                                &session_deps->socket_factory,
+                                session_deps->ssl_config_service,
+                                session_deps->spdy_session_pool,
+                                session_deps->http_auth_handler_factory.get(),
+                                NULL,
+                                NULL);
+}
+
+class TestSpdyStreamDelegate : public SpdyStream::Delegate {
+ public:
+  TestSpdyStreamDelegate(SpdyStream* stream,
+                         IOBufferWithSize* buf,
+                         CompletionCallback* callback)
+      : stream_(stream),
+        buf_(buf),
+        callback_(callback),
+        send_headers_completed_(false),
+        response_(new spdy::SpdyHeaderBlock),
+        data_sent_(0),
+        closed_(false) {}
+  virtual ~TestSpdyStreamDelegate() {}
+
+  virtual bool OnSendHeadersComplete(int status) {
+    send_headers_completed_ = true;
+    return true;
+  }
+  virtual int OnSendBody() {
+    ADD_FAILURE() << "OnSendBody should not be called";
+    return ERR_UNEXPECTED;
+  }
+  virtual bool OnSendBodyComplete(int status) {
+    ADD_FAILURE() << "OnSendBodyComplete should not be called";
+    return true;
+  }
+  virtual int OnResponseReceived(const spdy::SpdyHeaderBlock& response,
+                                 base::Time response_time,
+                                 int status) {
+    EXPECT_TRUE(send_headers_completed_);
+    *response_ = response;
+    if (buf_) {
+      EXPECT_EQ(ERR_IO_PENDING,
+                stream_->WriteStreamData(buf_.get(), buf_->size(),
+                                         spdy::DATA_FLAG_NONE));
+    }
+    return status;
+  }
+  virtual void OnDataReceived(const char* buffer, int bytes) {
+    received_data_ += std::string(buffer, bytes);
+  }
+  virtual void OnDataSent(int length) {
+    data_sent_ += length;
+  }
+  virtual void OnClose(int status) {
+    closed_ = true;
+    CompletionCallback* callback = callback_;
+    callback_ = NULL;
+    callback->Run(OK);
+  }
+
+  bool send_headers_completed() const { return send_headers_completed_; }
+  const linked_ptr<spdy::SpdyHeaderBlock>& response() const {
+    return response_;
+  }
+  const std::string& received_data() const { return received_data_; }
+  int data_sent() const { return data_sent_; }
+  bool closed() const {  return closed_; }
+
+ private:
+  SpdyStream* stream_;
+  scoped_refptr<IOBufferWithSize> buf_;
+  CompletionCallback* callback_;
+  bool send_headers_completed_;
+  linked_ptr<spdy::SpdyHeaderBlock> response_;
+  std::string received_data_;
+  int data_sent_;
+  bool closed_;
+};
+
+spdy::SpdyFrame* ConstructSpdyBodyFrame(const char* data, int length) {
+  spdy::SpdyFramer framer;
+  return framer.CreateDataFrame(1, data, length, spdy::DATA_FLAG_NONE);
+}
+
+}  // anonymous namespace
+
+class SpdyStreamTest : public testing::Test {
+ protected:
+  SpdyStreamTest() {
+  }
+
+  scoped_refptr<SpdySession> CreateSpdySession() {
+    spdy::SpdyFramer::set_enable_compression_default(false);
+    HostPortPair host_port_pair("www.google.com", 80);
+    scoped_refptr<SpdySession> session(
+        session_->spdy_session_pool()->Get(
+            host_port_pair, session_, BoundNetLog()));
+    return session;
+  }
+
+  virtual void TearDown() {
+    MessageLoop::current()->RunAllPending();
+  }
+
+  scoped_refptr<HttpNetworkSession> session_;
+};
+
+TEST_F(SpdyStreamTest, SendDataAfterOpen) {
+  SessionDependencies session_deps;
+
+  session_ = CreateSession(&session_deps);
+  SpdySessionPoolPeer pool_peer_(session_->spdy_session_pool());
+
+  const SpdyHeaderInfo kSynStartHeader = {
+    spdy::SYN_STREAM,
+    1,
+    0,
+    SPDY_PRIORITY_LOWEST,
+    spdy::CONTROL_FLAG_NONE,
+    false,
+    spdy::INVALID,
+    NULL,
+    0,
+    spdy::DATA_FLAG_NONE
+  };
+  static const char* const kGetHeaders[] = {
+    "method",
+    "GET",
+    "url",
+    "http://www.google.com/",
+    "version",
+    "HTTP/1.1",
+  };
+  scoped_ptr<spdy::SpdyFrame> req(
+      ConstructSpdyPacket(
+          kSynStartHeader, NULL, 0, kGetHeaders, arraysize(kGetHeaders) / 2));
+  scoped_ptr<spdy::SpdyFrame> msg(
+      ConstructSpdyBodyFrame("\0hello!\xff", 8));
+  MockWrite writes[] = {
+    CreateMockWrite(*req),
+    CreateMockWrite(*msg),
+  };
+  writes[0].sequence_number = 0;
+  writes[1].sequence_number = 2;
+
+  scoped_ptr<spdy::SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
+  scoped_ptr<spdy::SpdyFrame> echo(
+      ConstructSpdyBodyFrame("\0hello!\xff", 8));
+  MockRead reads[] = {
+    CreateMockRead(*resp),
+    CreateMockRead(*echo),
+    MockRead(true, 0, 0), // EOF
+  };
+  reads[0].sequence_number = 1;
+  reads[1].sequence_number = 3;
+  reads[2].sequence_number = 4;
+
+  scoped_refptr<OrderedSocketData> data(
+      new OrderedSocketData(reads, arraysize(reads),
+                            writes, arraysize(writes)));
+
+  session_deps.socket_factory.AddSocketDataProvider(data.get());
+  SpdySession::SetSSLMode(false);
+
+  scoped_refptr<SpdySession> session(CreateSpdySession());
+  GURL url("http://www.google.com/");
+
+  HostPortPair host_port_pair("www.google.com", 80);
+  scoped_refptr<TCPSocketParams> tcp_params =
+      new TCPSocketParams(host_port_pair, LOWEST, GURL(), false);
+  EXPECT_EQ(OK, session->Connect("spdy.www.google.com", tcp_params,
+                                 LOWEST));
+
+  scoped_refptr<SpdyStream> stream;
+  ASSERT_EQ(
+      OK,
+      session->CreateStream(url, LOWEST, &stream, BoundNetLog(), NULL, NULL));
+  scoped_refptr<IOBufferWithSize> buf(new IOBufferWithSize(8));
+  memcpy(buf->data(), "\0hello!\xff", 8);
+  TestCompletionCallback callback;
+
+  scoped_ptr<TestSpdyStreamDelegate> delegate(
+      new TestSpdyStreamDelegate(stream.get(), buf.get(), &callback));
+  stream->SetDelegate(delegate.get());
+
+  linked_ptr<spdy::SpdyHeaderBlock> headers(new spdy::SpdyHeaderBlock);
+  (*headers)["method"] = "GET";
+  (*headers)["url"] = "http://www.google.com/";
+  (*headers)["version"] = "HTTP/1.1";
+  stream->set_spdy_headers(headers);
+
+  EXPECT_EQ(ERR_IO_PENDING, stream->DoSendRequest(true));
+
+  EXPECT_EQ(OK, callback.WaitForResult());
+
+  EXPECT_TRUE(delegate->send_headers_completed());
+  EXPECT_EQ("200", (*delegate->response())["status"]);
+  EXPECT_EQ("HTTP/1.1", (*delegate->response())["version"]);
+  EXPECT_EQ(std::string("\0hello!\xff", 8), delegate->received_data());
+  EXPECT_EQ(8, delegate->data_sent());
+  EXPECT_TRUE(delegate->closed());
+}
+
+}  // namespace net