Prototype of chunked transfer encoded POST.

net/base/upload_data_stream.cc

 // 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/base/upload_data_stream.h"
 
 #include "base/file_util.h"
 #include "base/logging.h"
 #include "net/base/file_stream.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 
 namespace net {
 
 UploadDataStream* UploadDataStream::Create(UploadData* data, int* error_code) {
   scoped_ptr<UploadDataStream> stream(new UploadDataStream(data));
   int rv = stream->FillBuf();
   if (error_code)
     *error_code = rv;
   if (rv != OK)
     return NULL;
 
   return stream.release();
 }
 
 UploadDataStream::UploadDataStream(UploadData* data)
     : data_(data),
       buf_(new IOBuffer(kBufSize)),
       buf_len_(0),
-      next_element_(data->elements()->begin()),
+      next_element_(0),
       next_element_offset_(0),
       next_element_remaining_(0),
       total_size_(data->GetContentLength()),
       current_position_(0),
+      waiting_for_data_(false),
+      ALLOW_THIS_IN_INITIALIZER_LIST(
+          upload_data_callback_(this, &UploadDataStream::OnDataAvailable)),
+      data_callback_(NULL),
       eof_(false) {
+  if (data_->is_chunked()) {
+    waiting_for_data_ = true;
+    data_->set_data_callback(&upload_data_callback_);
+  }
 }
 
 UploadDataStream::~UploadDataStream() {
 }
 
 void UploadDataStream::DidConsume(size_t num_bytes) {
   DCHECK_LE(num_bytes, buf_len_);
   DCHECK(!eof_);
 
   buf_len_ -= num_bytes;
   if (buf_len_)
     memmove(buf_->data(), buf_->data() + num_bytes, buf_len_);
 
   FillBuf();
 
   current_position_ += num_bytes;
 }
 
+void UploadDataStream::set_data_callback(CompletionCallback* callback) {
+  DCHECK(!eof() && waiting_for_data());
+  data_callback_ = callback;
+}
+
+void UploadDataStream::OnDataAvailable(int result) {
+  if (!waiting_for_data_)
+    return;
+
+  waiting_for_data_ = false;
+  FillBuf();
+
+  // Callback may be NULL if we received more data before the previously
+  // received data was read.
+  if (data_callback_) {
+    CompletionCallback* c = data_callback_;
+    data_callback_ = NULL;
+    c->Run(0);

[wtc, 2011/01/12 02:39:02]

Should you pass 'result' instead of 0 to c->Run()?

Alternatively, I think the callback should be a Task
instead of a CompletionCallback.

[Satish, 2011/01/13 17:43:27]

No. A non-zero value is given only by the socket write code to indicate that so
many bytes can really be discarded from the input stream. Here we just want to
trigger the same code path to read more data in, but there is no change in how
many bytes were sent out so nothing to discard/consume from the buffer.
Changed to ChunkCallback interface now since this can just be set once and
invoked repeatedly like in UploadData

+  }
+}
+
 int UploadDataStream::FillBuf() {
-  std::vector<UploadData::Element>::iterator end =
-      data_->elements()->end();
+  std::vector<UploadData::Element>& elements = *data_->elements();
+  if (elements.size() == 0 && data_->is_chunked()) {
+    // We are waiting for data from UploadData, so nothing to do.
+    return OK;
+  }
 
-  while (buf_len_ < kBufSize && next_element_ != end) {
+  while (buf_len_ < kBufSize && next_element_ < elements.size()) {
     bool advance_to_next_element = false;
 
-    UploadData::Element& element = *next_element_;
+    UploadData::Element& element = elements[next_element_];
 
     size_t size_remaining = kBufSize - buf_len_;
-    if (element.type() == UploadData::TYPE_BYTES) {
+    if (element.type() == UploadData::TYPE_BYTES ||
+        element.type() == UploadData::TYPE_CHUNK) {
       const std::vector<char>& d = element.bytes();
       size_t count = d.size() - next_element_offset_;
 
       size_t bytes_copied = std::min(count, size_remaining);
 
-      memcpy(buf_->data() + buf_len_, &d[next_element_offset_], bytes_copied);
+      if (bytes_copied > 0)
+        memcpy(buf_->data() + buf_len_, &d[next_element_offset_], bytes_copied);
       buf_len_ += bytes_copied;
 
       if (bytes_copied == count) {
         advance_to_next_element = true;
       } else {
         next_element_offset_ += bytes_copied;
       }
     } else {
       DCHECK(element.type() == UploadData::TYPE_FILE);
 
@@ skipping 38 matching lines @@
     }
 
     if (advance_to_next_element) {
       ++next_element_;
       next_element_offset_ = 0;
       next_element_remaining_ = 0;
       next_element_stream_.reset();
     }
   }
 
-  if (next_element_ == end && !buf_len_)
-    eof_ = true;
+  if (next_element_ == elements.size() && !buf_len_) {
+    // If the last element in the list is a CHUNK buffer, we have to wait
+    // for more data to come in.
+    if (elements.size() > 0 &&
+        elements[elements.size() - 1].type() == UploadData::TYPE_CHUNK) {
+      waiting_for_data_ = true;
+    } else {
+      eof_ = true;
+    }
+  }
 
   return OK;
 }
 
 }  // namespace net