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

net/spdy/spdy_http_stream.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 <algorithm>
 #include <list>
 #include <string>
 
@@ skipping 18 matching lines @@
                                bool direct)
     : ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
       stream_(NULL),
       spdy_session_(spdy_session),
       response_info_(NULL),
       download_finished_(false),
       response_headers_received_(false),
       user_buffer_len_(0),
       buffered_read_callback_pending_(false),
       more_read_data_pending_(false),
-      direct_(direct) { }
+      direct_(direct),
+      waiting_for_chunk_(false) { }
 
 void SpdyHttpStream::InitializeWithExistingStream(SpdyStream* spdy_stream) {
   stream_ = spdy_stream;
   stream_->SetDelegate(this);
   response_headers_received_ = true;
 }
 
 SpdyHttpStream::~SpdyHttpStream() {
+  if (request_body_stream_ != NULL)
+    request_body_stream_->set_chunk_callback(NULL);
   if (stream_)
     stream_->DetachDelegate();
 }
 
 int SpdyHttpStream::InitializeStream(const HttpRequestInfo* request_info,
                                      const BoundNetLog& stream_net_log,
                                      const CompletionCallback& callback) {
   DCHECK(!stream_.get());
   if (spdy_session_->IsClosed())
    return ERR_CONNECTION_CLOSED;
@@ skipping 117 matching lines @@
 
 void SpdyHttpStream::SetConnectionReused() {
   // SPDY doesn't need an indicator here.
 }
 
 bool SpdyHttpStream::IsConnectionReusable() const {
   // SPDY streams aren't considered reusable.
   return false;
 }
 
-void SpdyHttpStream::set_chunk_callback(ChunkCallback* callback) {
-  if (request_body_stream_ != NULL)
-    request_body_stream_->set_chunk_callback(callback);
-}
-
 int SpdyHttpStream::SendRequest(const HttpRequestHeaders& request_headers,
                                 scoped_ptr<UploadDataStream> request_body,
                                 HttpResponseInfo* response,
                                 const CompletionCallback& callback) {
   base::Time request_time = base::Time::Now();
   CHECK(stream_.get());
 
   stream_->SetDelegate(this);
 
   linked_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
   CreateSpdyHeadersFromHttpRequest(*request_info_, request_headers,
                                    headers.get(), stream_->GetProtocolVersion(),
                                    direct_);
   stream_->set_spdy_headers(headers);
 
   stream_->SetRequestTime(request_time);
   // This should only get called in the case of a request occurring
   // during server push that has already begun but hasn't finished,
   // so we set the response's request time to be the actual one
   if (response_info_)
     response_info_->request_time = request_time;
 
   CHECK(!request_body_stream_.get());
   if (request_body != NULL) {
     if (request_body->size() || request_body->is_chunked()) {
       request_body_stream_.reset(request_body.release());
+      request_body_stream_->set_chunk_callback(this);
       // Use kMaxSpdyFrameChunkSize as the buffer size, since the request
       // body data is written with this size at a time.
       raw_request_body_buf_ = new IOBufferWithSize(kMaxSpdyFrameChunkSize);
       // The request body buffer is empty at first.
       request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_, 0);
     }
   }
 
   CHECK(!callback.is_null());
   CHECK(!stream_->cancelled());
@@ skipping 10 matching lines @@
   //
   // a) A client initiated request. In this case, |response_info_| should be
   //    NULL to start with.
   // b) A client request which matches a response that the server has already
   //    pushed.
   if (push_response_info_.get()) {
     *response = *(push_response_info_.get());
     push_response_info_.reset();
   }
   else
     DCHECK_EQ(static_cast<HttpResponseInfo*>(NULL), response_info_);

[Ryan Sleevi, 2012/07/05 06:52:06]

style nit: this should be

} else {
  DCHECK_EQ(...)
}

http://google-styleguide.googlecode.com/svn/trunk/cppguide.xml?showone=Condit...
- "However, if one part" ...

[ramant (doing other things), 2012/07/06 22:55:42]

Done.

 
   response_info_ = response;
 
   // Put the peer's IP address and port into the response.
   IPEndPoint address;
   int result = stream_->GetPeerAddress(&address);
   if (result != OK)
     return result;
   response_info_->socket_address = HostPortPair::FromIPEndPoint(address);
 
   bool has_upload_data = request_body_stream_.get() != NULL;
   result = stream_->SendRequest(has_upload_data);
   if (result == ERR_IO_PENDING) {
     CHECK(callback_.is_null());
     callback_ = callback;
   }
   return result;
 }
 
 void SpdyHttpStream::Cancel() {
   if (spdy_session_)
     spdy_session_->CancelPendingCreateStreams(&stream_);
   callback_.Reset();
   if (stream_)
     stream_->Cancel();
 }
 
+int SpdyHttpStream::SendData() {
+  CHECK(request_body_stream_.get());
+  CHECK_EQ(0, request_body_buf_->BytesRemaining());
+
+  // Read the data from the request body stream.
+  const int bytes_read = request_body_stream_->Read(
+      raw_request_body_buf_, raw_request_body_buf_->size());
+  DCHECK(!waiting_for_chunk_ || bytes_read != ERR_IO_PENDING);
+
+  if (request_body_stream_->is_chunked() && bytes_read == ERR_IO_PENDING) {
+    waiting_for_chunk_ = true;
+    return ERR_IO_PENDING;
+  }
+
+  waiting_for_chunk_ = false;
+
+  // ERR_IO_PENDING with chunked encoding is the only possible error.
+  DCHECK_GE(bytes_read, 0);
+
+  request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_,
+                                            bytes_read);
+
+  const bool eof = request_body_stream_->IsEOF();
+  return stream_->WriteStreamData(
+      request_body_buf_,
+      request_body_buf_->BytesRemaining(),
+      eof ? DATA_FLAG_FIN : DATA_FLAG_NONE);
+}
+
 bool SpdyHttpStream::OnSendHeadersComplete(int status) {
   if (!callback_.is_null())
     DoCallback(status);
   return request_body_stream_.get() == NULL;
 }
 
 int SpdyHttpStream::OnSendBody() {
   CHECK(request_body_stream_.get());
+  const bool eof = request_body_stream_->IsEOF();
+  if (request_body_buf_->BytesRemaining() > 0) {
+    return stream_->WriteStreamData(
+        request_body_buf_,
+        request_body_buf_->BytesRemaining(),
+        eof ? DATA_FLAG_FIN : DATA_FLAG_NONE);
+  }
 
-  // TODO(satorux): Clean up the logic here. This behavior is weird. Reading
-  // of upload data should happen in OnSendBody(). crbug.com/113107.
-  //
-  // Nothing to send. This happens when OnSendBody() is first called.
-  // A read of the upload data stream is initiated in OnSendBodyComplete().
-  if (request_body_buf_->BytesRemaining() == 0)
+  // The entire body data has been sent.
+  if (eof)
     return OK;
 
-  const bool eof = request_body_stream_->IsEOF();
-  return stream_->WriteStreamData(
-      request_body_buf_,
-      request_body_buf_->BytesRemaining(),
-      eof ? DATA_FLAG_FIN : DATA_FLAG_NONE);
+  return SendData();
 }
 
 int SpdyHttpStream::OnSendBodyComplete(int status, bool* eof) {
   // |status| is the number of bytes written to the SPDY stream.
   CHECK(request_body_stream_.get());
   *eof = false;
 
   if (status > 0) {
     request_body_buf_->DidConsume(status);
     if (request_body_buf_->BytesRemaining()) {
       // Go back to OnSendBody() to send the remaining data.
       return OK;
     }
   }
 
   // Check if the entire body data has been sent.
   *eof = (request_body_stream_->IsEOF() &&
           !request_body_buf_->BytesRemaining());
-  if (*eof)
-    return OK;
-
-  // Read the data from the request body stream.
-  const int bytes_read = request_body_stream_->Read(
-      raw_request_body_buf_, raw_request_body_buf_->size());
-  if (request_body_stream_->is_chunked() && bytes_read == ERR_IO_PENDING)
-    return ERR_IO_PENDING;
-  // ERR_IO_PENDING with chunked encoding is the only possible error.
-  DCHECK_GE(bytes_read, 0);
-
-  request_body_buf_ = new DrainableIOBuffer(raw_request_body_buf_,
-                                            bytes_read);
   return OK;
 }
 
 int SpdyHttpStream::OnResponseReceived(const SpdyHeaderBlock& response,
                                        base::Time response_time,
                                        int status) {
   if (!response_info_) {
     DCHECK(stream_->pushed());
     push_response_info_.reset(new HttpResponseInfo);
     response_info_ = push_response_info_.get();
@@ skipping 62 matching lines @@
 }
 
 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 (request_body_stream_ != NULL)
+    request_body_stream_->set_chunk_callback(NULL);

[Ryan Sleevi, 2012/07/05 06:52:06]

Does Cancel also need to be updated?

Not sure the lifetime semantics there, but it is resetting stream_

[ramant (doing other things), 2012/07/06 22:55:42]

No need. Cancel calls reset spdy stream, which deletes the stream (which calls
OnClose and which calls SpdyHttpStream's OnClose).

   if (status == net::OK) {
     // We need to complete any pending buffered read now.
     invoked_callback = DoBufferedReadCallback();
   }
   if (!invoked_callback && !callback_.is_null())
     DoCallback(status);
 }
 
+void SpdyHttpStream::OnChunkAvailable() {
+  if (!waiting_for_chunk_)
+    return;
+  DCHECK(request_body_stream_->is_chunked());
+  SendData();
+}
+
 void SpdyHttpStream::ScheduleBufferedReadCallback() {
   // If there is already a scheduled DoBufferedReadCallback, don't issue
   // another one.  Mark that we have received more data and return.
   if (buffered_read_callback_pending_) {
     more_read_data_pending_ = true;
     return;
   }
 
   more_read_data_pending_ = false;
   buffered_read_callback_pending_ = true;
@@ skipping 77 matching lines @@
 bool SpdyHttpStream::IsSpdyHttpStream() const {
   return true;
 }
 
 void SpdyHttpStream::Drain(HttpNetworkSession* session) {
   Close(false);
   delete this;
 }
 
 }  // namespace net