SPDY - chunked upload - speech recognition doesn't work with SPDY/3

net/spdy/spdy_http_stream_spdy3_unittest.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/spdy/spdy_http_stream.h"
 
 #include "base/memory/scoped_ptr.h"
 #include "base/threading/sequenced_worker_pool.h"
 #include "crypto/ec_private_key.h"
 #include "crypto/ec_signature_creator.h"
@@ skipping 16 matching lines @@
   OrderedSocketData* data() { return data_.get(); }
  protected:
   SpdyHttpStreamSpdy3Test() {}
 
   virtual void SetUp() {
     SpdySession::set_default_protocol(kProtoSPDY3);
   }
 
   virtual void TearDown() {
     crypto::ECSignatureCreator::SetFactoryForTesting(NULL);
+    UploadDataStream::ResetMergeChunks();
     MessageLoop::current()->RunAllPending();
   }
 
+  void set_merge_chunks(bool merge) {
+    UploadDataStream::set_merge_chunks(merge);
+  }
+
   int InitSession(MockRead* reads, size_t reads_count,
                   MockWrite* writes, size_t writes_count,
                   HostPortPair& host_port_pair) {
     HostPortProxyPair pair(host_port_pair, ProxyServer::Direct());
     data_.reset(new OrderedSocketData(reads, reads_count,
                                       writes, writes_count));
     session_deps_.socket_factory->AddSocketDataProvider(data_.get());
     http_session_ = SpdySessionDependencies::SpdyCreateSession(&session_deps_);
     session_ = http_session_->spdy_session_pool()->Get(pair, BoundNetLog());
     transport_params_ = new TransportSocketParams(host_port_pair,
@@ skipping 69 matching lines @@
   data()->CompleteRead();
 
   // Because we abandoned the stream, we don't expect to find a session in the
   // pool anymore.
   EXPECT_FALSE(http_session_->spdy_session_pool()->HasSession(pair));
   EXPECT_TRUE(data()->at_read_eof());
   EXPECT_TRUE(data()->at_write_eof());
 }
 
 TEST_F(SpdyHttpStreamSpdy3Test, SendChunkedPost) {
-  UploadDataStream::set_merge_chunks(false);
+  set_merge_chunks(false);
 
   scoped_ptr<SpdyFrame> req(ConstructChunkedSpdyPost(NULL, 0));
   scoped_ptr<SpdyFrame> chunk1(ConstructSpdyBodyFrame(1, false));
   scoped_ptr<SpdyFrame> chunk2(ConstructSpdyBodyFrame(1, true));
   MockWrite writes[] = {
-    CreateMockWrite(*req.get(), 1),
-    CreateMockWrite(*chunk1, 2),  // POST upload frames
-    CreateMockWrite(*chunk2, 3),
+    CreateMockWrite(*req.get(), 0),
+    CreateMockWrite(*chunk1, 1),  // POST upload frames
+    CreateMockWrite(*chunk2, 2),
   };
   scoped_ptr<SpdyFrame> resp(ConstructSpdyPostSynReply(NULL, 0));
   MockRead reads[] = {
-    CreateMockRead(*resp, 4),
-    CreateMockRead(*chunk1, 5),
+    CreateMockRead(*resp, 3),
+    CreateMockRead(*chunk1, 4),
     CreateMockRead(*chunk2, 5),
     MockRead(SYNCHRONOUS, 0, 6)  // EOF
   };
 
   HostPortPair host_port_pair("www.google.com", 80);
   HostPortProxyPair pair(host_port_pair, ProxyServer::Direct());
   EXPECT_EQ(OK, InitSession(reads, arraysize(reads), writes, arraysize(writes),
                             host_port_pair));
 
   HttpRequestInfo request;
@@ skipping 26 matching lines @@
   data()->CompleteRead();
   MessageLoop::current()->RunAllPending();
 
   // Because we abandoned the stream, we don't expect to find a session in the
   // pool anymore.
   EXPECT_FALSE(http_session_->spdy_session_pool()->HasSession(pair));
   EXPECT_TRUE(data()->at_read_eof());
   EXPECT_TRUE(data()->at_write_eof());
 }
 
+// Test to ensure the SpdyStream state machine does not get confused when
+// sending a request with a chunked body, where chunks become available
+// asynchronously, over a socket where writes may also complete asynchronously.
+TEST_F(SpdyHttpStreamSpdy3Test, DelayedSendChunkedPost) {
+  set_merge_chunks(false);
+
+  const char kUploadData1[] = "12345678";
+  const int kUploadData1Size = arraysize(kUploadData1)-1;
+  scoped_ptr<SpdyFrame> req(ConstructChunkedSpdyPost(NULL, 0));
+  scoped_ptr<SpdyFrame> chunk1(ConstructSpdyBodyFrame(1, false));
+  scoped_ptr<SpdyFrame> chunk2(
+      ConstructSpdyBodyFrame(1, kUploadData1, kUploadData1Size, false));
+  scoped_ptr<SpdyFrame> chunk3(ConstructSpdyBodyFrame(1, true));
+  MockWrite writes[] = {
+    CreateMockWrite(*req.get(), 0),
+    CreateMockWrite(*chunk1, 1),  // POST upload frames
+    CreateMockWrite(*chunk2, 2),
+    CreateMockWrite(*chunk3, 3),
+  };
+  scoped_ptr<SpdyFrame> resp(ConstructSpdyPostSynReply(NULL, 0));
+  MockRead reads[] = {
+    CreateMockRead(*resp, 4),
+    CreateMockRead(*chunk1, 5),
+    CreateMockRead(*chunk2, 6),
+    CreateMockRead(*chunk3, 7),
+    MockRead(ASYNC, 0, 8)  // EOF
+  };
+
+  HostPortPair host_port_pair("www.google.com", 80);
+  HostPortProxyPair pair(host_port_pair, ProxyServer::Direct());
+
+  scoped_ptr<DeterministicSocketData> data(
+      new DeterministicSocketData(reads, arraysize(reads),
+                                  writes, arraysize(writes)));
+
+  DeterministicMockClientSocketFactory* socket_factory =
+      session_deps_.deterministic_socket_factory.get();
+  socket_factory->AddSocketDataProvider(data.get());
+
+  http_session_ = SpdySessionDependencies::SpdyCreateSessionDeterministic(
+      &session_deps_);
+  session_ = http_session_->spdy_session_pool()->Get(pair, BoundNetLog());
+  transport_params_ = new TransportSocketParams(host_port_pair,
+                                                MEDIUM, false, false,
+                                                OnHostResolutionCallback());
+
+  TestCompletionCallback callback;
+  scoped_ptr<ClientSocketHandle> connection(new ClientSocketHandle);
+
+  EXPECT_EQ(ERR_IO_PENDING,
+            connection->Init(host_port_pair.ToString(),
+                             transport_params_,
+                             MEDIUM,
+                             callback.callback(),
+                             http_session_->GetTransportSocketPool(
+                                 HttpNetworkSession::NORMAL_SOCKET_POOL),
+                             BoundNetLog()));
+
+  callback.WaitForResult();
+  EXPECT_EQ(OK,
+            session_->InitializeWithSocket(connection.release(), false, OK));
+
+  HttpRequestInfo request;
+  request.method = "POST";
+  request.url = GURL("http://www.google.com/");
+  request.upload_data = new UploadData();
+  request.upload_data->set_is_chunked(true);
+
+  BoundNetLog net_log;
+  scoped_ptr<SpdyHttpStream> http_stream(
+      new SpdyHttpStream(session_.get(), true));
+  ASSERT_EQ(OK,
+            http_stream->InitializeStream(&request,
+                                          net_log,
+                                          CompletionCallback()));
+
+  scoped_ptr<UploadDataStream> upload_stream(
+      new UploadDataStream(request.upload_data));
+  ASSERT_EQ(OK, upload_stream->Init());
+
+  request.upload_data->AppendChunk(kUploadData, kUploadDataSize, false);
+
+  HttpRequestHeaders headers;
+  HttpResponseInfo response;
+  // This will attempt to Write() the initial request and headers, which will
+  // complete asynchronously.
+  EXPECT_EQ(ERR_IO_PENDING,
+            http_stream->SendRequest(headers,
+                                     upload_stream.Pass(),
+                                     &response,
+                                     callback.callback()));
+  EXPECT_TRUE(http_session_->spdy_session_pool()->HasSession(pair));
+
+  // Complete the initial request write and the first chunk.
+  data->RunFor(2);
+  ASSERT_TRUE(callback.have_result());
+  EXPECT_GT(callback.WaitForResult(), 0);
+
+  // Now append the second chunk. This will enqueue another write.
+  request.upload_data->AppendChunk(kUploadData1, kUploadData1Size, false);
+
+  // Now append final chunk. This will enqueue another write.
+  request.upload_data->AppendChunk(kUploadData, kUploadDataSize, true);
+
+  // Finish writing all the chunks.
+  data->RunFor(2);
+
+  // Read response headers.
+  data->RunFor(1);
+  ASSERT_EQ(OK, http_stream->ReadResponseHeaders(callback.callback()));
+
+  // Read and check |chunk1| response.
+  data->RunFor(1);
+  scoped_refptr<IOBuffer> buf1(new IOBuffer(kUploadDataSize));
+  int rv = http_stream->ReadResponseBody(buf1,
+                                         kUploadDataSize,
+                                         callback.callback());
+  ASSERT_EQ(kUploadDataSize, rv);
+  EXPECT_EQ(kUploadData, std::string(buf1->data(), kUploadDataSize));
+
+  // Read and check |chunk2| response.
+  data->RunFor(1);
+  scoped_refptr<IOBuffer> buf2(new IOBuffer(kUploadData1Size));
+  rv = http_stream->ReadResponseBody(buf2,
+                                     kUploadData1Size,
+                                     callback.callback());
+  ASSERT_EQ(kUploadData1Size, rv);
+  EXPECT_EQ(kUploadData1, std::string(buf2->data(), kUploadData1Size));
+
+  // Read and check |chunk3| response.
+  data->RunFor(1);
+  scoped_refptr<IOBuffer> buf3(new IOBuffer(kUploadDataSize));
+  rv = http_stream->ReadResponseBody(buf3,
+                                     kUploadDataSize,
+                                     callback.callback());
+  ASSERT_EQ(kUploadDataSize, rv);
+  EXPECT_EQ(kUploadData, std::string(buf3->data(), kUploadDataSize));
+
+  // Finish reading the |EOF|.
+  data->RunFor(1);
+  ASSERT_TRUE(response.headers.get());
+  ASSERT_EQ(200, response.headers->response_code());
+  EXPECT_TRUE(data->at_read_eof());
+  EXPECT_TRUE(data->at_write_eof());
+}
+
+// Test the receipt of a WINDOW_UPDATE frame while waiting for a chunk to be
+// made available is handled correctly.
+TEST_F(SpdyHttpStreamSpdy3Test, DelayedSendChunkedPostWithWindowUpdate) {
+  set_merge_chunks(false);
+
+  const char kUploadData1[] = "12345678";
+  const int kUploadData1Size = arraysize(kUploadData1)-1;
+  scoped_ptr<SpdyFrame> req(ConstructChunkedSpdyPost(NULL, 0));
+  scoped_ptr<SpdyFrame> chunk1(ConstructSpdyBodyFrame(1, false));
+  scoped_ptr<SpdyFrame> chunk2(
+      ConstructSpdyBodyFrame(1, kUploadData1, kUploadData1Size, false));
+  scoped_ptr<SpdyFrame> chunk3(ConstructSpdyBodyFrame(1, true));
+  MockWrite writes[] = {
+    CreateMockWrite(*req.get(), 0),
+    CreateMockWrite(*chunk1, 1),  // POST upload frames
+    CreateMockWrite(*chunk2, 3),
+    CreateMockWrite(*chunk3, 4),
+  };
+  scoped_ptr<SpdyFrame> resp(ConstructSpdyPostSynReply(NULL, 0));
+  scoped_ptr<SpdyFrame> window_update(
+      ConstructSpdyWindowUpdate(1, kUploadDataSize));
+  MockRead reads[] = {
+    CreateMockRead(*window_update, 2),
+    CreateMockRead(*resp, 5),
+    CreateMockRead(*chunk1, 6),
+    CreateMockRead(*chunk2, 7),
+    CreateMockRead(*chunk3, 8),
+    MockRead(ASYNC, 0, 9)  // EOF
+  };
+
+  HostPortPair host_port_pair("www.google.com", 80);
+  HostPortProxyPair pair(host_port_pair, ProxyServer::Direct());
+
+  scoped_ptr<DeterministicSocketData> data(
+      new DeterministicSocketData(reads, arraysize(reads),
+                                  writes, arraysize(writes)));
+
+  DeterministicMockClientSocketFactory* socket_factory =
+      session_deps_.deterministic_socket_factory.get();
+  socket_factory->AddSocketDataProvider(data.get());
+
+  http_session_ = SpdySessionDependencies::SpdyCreateSessionDeterministic(
+      &session_deps_);
+  session_ = http_session_->spdy_session_pool()->Get(pair, BoundNetLog());
+  transport_params_ = new TransportSocketParams(host_port_pair,
+                                                MEDIUM, false, false,
+                                                OnHostResolutionCallback());
+
+  TestCompletionCallback callback;
+  scoped_ptr<ClientSocketHandle> connection(new ClientSocketHandle);
+
+  EXPECT_EQ(ERR_IO_PENDING,
+            connection->Init(host_port_pair.ToString(),
+                             transport_params_,
+                             MEDIUM,
+                             callback.callback(),
+                             http_session_->GetTransportSocketPool(
+                                 HttpNetworkSession::NORMAL_SOCKET_POOL),
+                             BoundNetLog()));
+
+  callback.WaitForResult();
+  EXPECT_EQ(OK,
+            session_->InitializeWithSocket(connection.release(), false, OK));
+
+  HttpRequestInfo request;
+  request.method = "POST";
+  request.url = GURL("http://www.google.com/");
+  request.upload_data = new UploadData();
+  request.upload_data->set_is_chunked(true);
+
+  BoundNetLog net_log;
+  scoped_ptr<SpdyHttpStream> http_stream(
+      new SpdyHttpStream(session_.get(), true));
+  ASSERT_EQ(OK,
+            http_stream->InitializeStream(&request,
+                                          net_log,
+                                          CompletionCallback()));
+
+  scoped_ptr<UploadDataStream> upload_stream(
+      new UploadDataStream(request.upload_data));
+  ASSERT_EQ(OK, upload_stream->Init());
+
+  request.upload_data->AppendChunk(kUploadData, kUploadDataSize, false);
+
+  HttpRequestHeaders headers;
+  HttpResponseInfo response;
+  // This will attempt to Write() the initial request and headers, which will
+  // complete asynchronously.
+  EXPECT_EQ(ERR_IO_PENDING,
+            http_stream->SendRequest(headers,
+                                     upload_stream.Pass(),
+                                     &response,
+                                     callback.callback()));
+  EXPECT_TRUE(http_session_->spdy_session_pool()->HasSession(pair));
+
+  // Complete the initial request write, first chunk and window_update.
+  data->RunFor(2);
+  ASSERT_TRUE(callback.have_result());
+  EXPECT_GT(callback.WaitForResult(), 0);
+
+  // Verify that the window size has decreased.
+  ASSERT_TRUE(http_stream->stream() != NULL);
+  EXPECT_NE(static_cast<int>(kSpdyStreamInitialWindowSize),
+            http_stream->stream()->send_window_size());
+
+  // Read window update.
+  data->RunFor(1);
+
+  // Verify the window update.
+  ASSERT_TRUE(http_stream->stream() != NULL);
+  EXPECT_EQ(static_cast<int>(kSpdyStreamInitialWindowSize),
+            http_stream->stream()->send_window_size());
+
+  // Now append the second chunk. This will enqueue another write.
+  request.upload_data->AppendChunk(kUploadData1, kUploadData1Size, false);
+
+  // Now append final chunk. This will enqueue another write.
+  request.upload_data->AppendChunk(kUploadData, kUploadDataSize, true);
+
+  // Finish writing all the chunks.
+  data->RunFor(2);
+
+  // Read response headers.
+  data->RunFor(1);
+  ASSERT_EQ(OK, http_stream->ReadResponseHeaders(callback.callback()));
+
+  // Read and check |chunk1| response.
+  data->RunFor(1);
+  scoped_refptr<IOBuffer> buf1(new IOBuffer(kUploadDataSize));
+  int rv = http_stream->ReadResponseBody(buf1,
+                                         kUploadDataSize,
+                                         callback.callback());
+  ASSERT_EQ(kUploadDataSize, rv);
+  EXPECT_EQ(kUploadData, std::string(buf1->data(), kUploadDataSize));
+
+  // Read and check |chunk2| response.
+  data->RunFor(1);
+  scoped_refptr<IOBuffer> buf2(new IOBuffer(kUploadData1Size));
+  rv = http_stream->ReadResponseBody(buf2,
+                                     kUploadData1Size,
+                                     callback.callback());
+  ASSERT_EQ(kUploadData1Size, rv);
+  EXPECT_EQ(kUploadData1, std::string(buf2->data(), kUploadData1Size));
+
+  // Read and check |chunk3| response.
+  data->RunFor(1);
+  scoped_refptr<IOBuffer> buf3(new IOBuffer(kUploadDataSize));
+  rv = http_stream->ReadResponseBody(buf3,
+                                     kUploadDataSize,
+                                     callback.callback());
+  ASSERT_EQ(kUploadDataSize, rv);
+  EXPECT_EQ(kUploadData, std::string(buf3->data(), kUploadDataSize));
+
+  // Finish reading the |EOF|.
+  data->RunFor(1);
+  ASSERT_TRUE(response.headers.get());
+  ASSERT_EQ(200, response.headers->response_code());
+  EXPECT_TRUE(data->at_read_eof());
+  EXPECT_TRUE(data->at_write_eof());
+}
+
 // Test case for bug: http://code.google.com/p/chromium/issues/detail?id=50058
 TEST_F(SpdyHttpStreamSpdy3Test, SpdyURLTest) {
   const char * const full_url = "http://www.google.com/foo?query=what#anchor";
   const char * const base_url = "http://www.google.com/foo?query=what";
   scoped_ptr<SpdyFrame> req(ConstructSpdyGet(base_url, false, 1, LOWEST));
   MockWrite writes[] = {
     CreateMockWrite(*req.get(), 1),
   };
   scoped_ptr<SpdyFrame> resp(ConstructSpdyGetSynReply(NULL, 0, 1));
   MockRead reads[] = {
@@ skipping 328 matching lines @@
 
   sequenced_worker_pool->Shutdown();
 }
 
 #endif  // !defined(USE_OPENSSL)
 
 // TODO(willchan): Write a longer test for SpdyStream that exercises all
 // methods.
 
 }  // namespace net