Initial implementation of the ByteStream refactor.

content/browser/download/download_file_impl.cc

Index: content/browser/download/download_file_impl.cc
diff --git a/content/browser/download/download_file_impl.cc b/content/browser/download/download_file_impl.cc
index 5fb7ffc330bed912ce0dd68bdc46db65fbc2e36f..2638789361af1fb8199394d2daf08520b3b75714 100644
--- a/content/browser/download/download_file_impl.cc
+++ b/content/browser/download/download_file_impl.cc
@@ -6,10 +6,15 @@
 
 #include <string>
 
+#include "base/bind.h"
 #include "base/file_util.h"
+#include "base/message_loop_proxy.h"
 #include "content/browser/download/download_create_info.h"
 #include "content/public/browser/browser_thread.h"
+#include "content/browser/download/download_interrupt_reasons_impl.h"
 #include "content/public/browser/download_manager.h"
+#include "content/browser/download/download_stats.h"
+#include "net/base/io_buffer.h"
 
 using content::BrowserThread;
 using content::DownloadId;
@@ -29,6 +34,7 @@ DownloadFileImpl::DownloadFileImpl(
                 info->save_info.hash_state,
                 info->save_info.file_stream,
                 bound_net_log),
+          input_pipe_(info->pipe),
           id_(info->download_id),
           request_handle_(request_handle),
           download_manager_(download_manager) {
@@ -37,11 +43,28 @@ DownloadFileImpl::DownloadFileImpl(
 
 DownloadFileImpl::~DownloadFileImpl() {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::FILE));
+  // Prevent any future callbacks from reaching us.
+  if (input_pipe_.get())
+    input_pipe_->RegisterSinkCallback(scoped_refptr<base::TaskRunner>(),
+                                      base::Closure());
 }
 
 // BaseFile delegated functions.
 net::Error DownloadFileImpl::Initialize() {
-  return file_.Initialize();
+  net::Error result = file_.Initialize();
+  if (result != net::OK)
+    return result;
+
+  input_pipe_->RegisterSinkCallback(
+      base::MessageLoopProxy::current(),
+      // Unretained is safe because the callback is nulled in the
+      // destructor.
+      base::Bind(&DownloadFileImpl::PipeActive, base::Unretained(this)));
+
+  // Initial pull from the straw.
+  PipeActive();
+
+  return result;
 }
 
 net::Error DownloadFileImpl::AppendDataToFile(const char* data,
@@ -122,3 +145,72 @@ std::string DownloadFileImpl::DebugString() const {
                             request_handle_->DebugString().c_str(),
                             file_.DebugString().c_str());
 }
+
+void DownloadFileImpl::PipeActive() {
+  scoped_refptr<net::IOBuffer> incoming_data;
+  size_t length = 0;
+  size_t total_length = 0;
+  content::ByteStream::StreamState state(content::ByteStream::STREAM_EMPTY);
+  content::DownloadInterruptReason reason =
+      content::DOWNLOAD_INTERRUPT_REASON_NONE;
+
+  // Take care of any file local activity required.
+  do {
+    state = input_pipe_->GetData(&incoming_data, &length);
+
+    net::Error result = net::OK;
+    switch (state) {
+      case content::ByteStream::STREAM_EMPTY:
+        break;
+      case content::ByteStream::STREAM_NON_EMPTY:
+        result = AppendDataToFile(incoming_data.get()->data(), length);
+        total_length += length;
+        reason = content::ConvertNetErrorToInterruptReason(
+            result, content::DOWNLOAD_INTERRUPT_FROM_DISK);
+        break;
+      case content::ByteStream::STREAM_COMPLETE:
+        reason = input_pipe_->GetSourceResult();
+        if (reason != content::DOWNLOAD_INTERRUPT_REASON_NONE)
+          Finish();
+        break;
+      default:
+        NOTREACHED();
+        break;
+    }
+  } while (state == content::ByteStream::STREAM_NON_EMPTY &&
+           reason == content::DOWNLOAD_INTERRUPT_REASON_NONE);
+
+  if (total_length)
+    download_stats::RecordFileThreadReceiveBuffers(total_length);
+
+  // Take care of communication with our controller.
+  if (reason != content::DOWNLOAD_INTERRUPT_REASON_NONE) {
+    // Error case for both upstream source and file write.
+    // Shut down processing and signal an error to our controller.
+    // Our controller will clean us up.
+    input_pipe_->RegisterSinkCallback(scoped_refptr<base::TaskRunner>(),
+                                      base::Closure());
+    if (download_manager_.get()) {
+      BrowserThread::PostTask(
+          BrowserThread::UI, FROM_HERE,
+          base::Bind(&DownloadManager::OnDownloadInterrupted,
+                     download_manager_, id_.local(),
+                     BytesSoFar(), GetHashState(), reason));
+    }
+  } else if (state == content::ByteStream::STREAM_COMPLETE) {
+    // Signal successful completion.  Processing shutdown shouldn't
+    // be necessary, but it can't hurt to make sure.
+    input_pipe_->RegisterSinkCallback(scoped_refptr<base::TaskRunner>(),
+                                      base::Closure());
+    if (download_manager_.get()) {
+      std::string hash;
+      if (!GetHash(&hash) || file_.IsEmptyHash(hash))
+        hash.clear();
+      BrowserThread::PostTask(
+          BrowserThread::UI, FROM_HERE,
+          base::Bind(&DownloadManager::OnResponseCompleted,
+                     download_manager_, id_.local(),
+                     BytesSoFar(), hash));
+    }
+  }
+}