net: Make UploadDataStream::Read() asynchronous

net/base/upload_data_stream.cc

Index: net/base/upload_data_stream.cc
diff --git a/net/base/upload_data_stream.cc b/net/base/upload_data_stream.cc
index a62528d9bf816adacef194387cd06d21178f2fab..6620a50d7ce6d321fdac895b1e9ed48c9ab859b5 100644
--- a/net/base/upload_data_stream.cc
+++ b/net/base/upload_data_stream.cc
@@ -5,6 +5,7 @@
 #include "net/base/upload_data_stream.h"
 
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/base/upload_element_reader.h"
@@ -62,7 +63,21 @@ int UploadDataStream::InitSync() {
   return OK;
 }
 
-int UploadDataStream::Read(IOBuffer* buf, int buf_len) {
+int UploadDataStream::Read(IOBuffer* buf,
+                           int buf_len,
+                           const CompletionCallback& callback) {
+  DCHECK(initialized_successfully_);
+
+  // Use fast path when all data is in memory, chunked data is also processed
+  // with ReadSync() since it requires special logic.

[mmenke, 2012/10/12 21:30:29]

"since it requires special logic, and is already in memory"

[hashimoto, 2012/10/15 11:21:37]

Done.

+  if (IsInMemory() || is_chunked())
+    return ReadSync(buf, buf_len);
+
+  const bool invoked_asynchronously = false;
+  return ReadInternal(buf, buf_len, 0, invoked_asynchronously, callback, 0);
+}
+
+int UploadDataStream::ReadSync(IOBuffer* buf, int buf_len) {
   DCHECK(initialized_successfully_);
 
   // Initialize readers for newly appended chunks.
@@ -84,8 +99,10 @@ int UploadDataStream::Read(IOBuffer* buf, int buf_len) {
   int bytes_copied = 0;
   while (bytes_copied < buf_len && element_index_ < element_readers_.size()) {
     UploadElementReader* reader = element_readers_[element_index_];
-    bytes_copied += reader->ReadSync(buf->data() + bytes_copied,
-                                     buf_len - bytes_copied);
+    scoped_refptr<DrainableIOBuffer> sub_buffer =
+        new DrainableIOBuffer(buf, buf_len);
+    sub_buffer->SetOffset(bytes_copied);
+    bytes_copied += reader->ReadSync(sub_buffer, sub_buffer->BytesRemaining());
     if (reader->BytesRemaining() == 0)
       ++element_index_;
 
@@ -179,4 +196,50 @@ void UploadDataStream::FinalizeInitialization() {
   initialized_successfully_ = true;
 }
 
+int UploadDataStream::ReadInternal(scoped_refptr<IOBuffer> buf,
+                                   int buf_len,
+                                   int bytes_copied,
+                                   bool invoked_asynchronously,
+                                   const CompletionCallback& callback,
+                                   int previous_result) {
+  DCHECK(initialized_successfully_);
+  DCHECK_GE(previous_result, 0);
+
+  // Add the last result.
+  bytes_copied += previous_result;
+
+  while (bytes_copied < buf_len && element_index_ < element_readers_.size()) {
+    UploadElementReader* reader = element_readers_[element_index_];
+
+    if (reader->BytesRemaining() == 0) {
+      ++element_index_;
+      continue;
+    }
+    scoped_refptr<DrainableIOBuffer> sub_buffer =
+        new DrainableIOBuffer(buf, buf_len);
+    sub_buffer->SetOffset(bytes_copied);
+    const int result = reader->Read(
+        sub_buffer,
+        sub_buffer->BytesRemaining(),
+        base::Bind(base::IgnoreResult(&UploadDataStream::ReadInternal),
+                   weak_ptr_factory_.GetWeakPtr(),
+                   buf,
+                   buf_len,
+                   bytes_copied,
+                   true,  // invoked_asynchronously
+                   callback));
+    if (result == ERR_IO_PENDING)
+      return ERR_IO_PENDING;
+
+    DCHECK_GE(bytes_copied, 0);
+    bytes_copied += result;
+  }
+  current_position_ += bytes_copied;
+
+  // When invoked asynchronously, callback is the only way to return the result.
+  if (invoked_asynchronously)
+    callback.Run(bytes_copied);
+  return bytes_copied;
+}
+
 }  // namespace net