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. |
+ 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 |