Implement stream based cross file system copy.

This CL implements the FileStreamReader/Writer based cross file system copy
operation.

BUG=279278
TEST=Ran content_unittests and unit_tests

Review URL: https://codereview.chromium.org/23463048

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@225378 0039d316-1c4b-4281-b951-d872f2087c98

4 files changed, 390 insertions, 8 deletions

diff --git a/webkit/browser/fileapi/copy_or_move_operation_delegate.cc b/webkit/browser/fileapi/copy_or_move_operation_delegate.cc
index 845e33b..fed1abd 100644
--- a/webkit/browser/fileapi/copy_or_move_operation_delegate.cc
+++ b/webkit/browser/fileapi/copy_or_move_operation_delegate.cc
@@ -6,7 +6,11 @@
 
 #include "base/bind.h"
 #include "base/files/file_path.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "webkit/browser/blob/file_stream_reader.h"
 #include "webkit/browser/fileapi/copy_or_move_file_validator.h"
+#include "webkit/browser/fileapi/file_stream_writer.h"
 #include "webkit/browser/fileapi/file_system_context.h"
 #include "webkit/browser/fileapi/file_system_operation_runner.h"
 #include "webkit/browser/fileapi/file_system_url.h"
@@ -289,8 +293,230 @@ class SnapshotCopyOrMoveImpl
   DISALLOW_COPY_AND_ASSIGN(SnapshotCopyOrMoveImpl);
 };
 
+// The size of buffer for StreamCopyHelper.
+const int kReadBufferSize = 32768;
+
+// To avoid too many progress callbacks, it should be called less
+// frequently than 50ms.
+const int kMinProgressCallbackInvocationSpanInMilliseconds = 50;
+
+// Specifically for cross file system copy/move operation, this class uses
+// stream reader and writer for copying. Validator is not supported, so if
+// necessary SnapshotCopyOrMoveImpl should be used.
+class StreamCopyOrMoveImpl
+    : public CopyOrMoveOperationDelegate::CopyOrMoveImpl {
+ public:
+  StreamCopyOrMoveImpl(
+      FileSystemOperationRunner* operation_runner,
+      CopyOrMoveOperationDelegate::OperationType operation_type,
+      const FileSystemURL& src_url,
+      const FileSystemURL& dest_url,
+      scoped_ptr<webkit_blob::FileStreamReader> reader,
+      scoped_ptr<FileStreamWriter> writer,
+      const FileSystemOperation::CopyFileProgressCallback&
+          file_progress_callback)
+      : operation_runner_(operation_runner),
+        operation_type_(operation_type),
+        src_url_(src_url),
+        dest_url_(dest_url),
+        reader_(reader.Pass()),
+        writer_(writer.Pass()),
+        file_progress_callback_(file_progress_callback),
+        weak_factory_(this) {
+  }
+
+  virtual void Run(
+      const CopyOrMoveOperationDelegate::StatusCallback& callback) OVERRIDE {
+    // Reader can be created even if the entry does not exist or the entry is
+    // a directory. To check errors before destination file creation,
+    // check metadata first.
+    operation_runner_->GetMetadata(
+        src_url_,
+        base::Bind(&StreamCopyOrMoveImpl::RunAfterGetMetadataForSource,
+                   weak_factory_.GetWeakPtr(), callback));
+  }
+
+ private:
+  void RunAfterGetMetadataForSource(
+      const CopyOrMoveOperationDelegate::StatusCallback& callback,
+      base::PlatformFileError error,
+      const base::PlatformFileInfo& file_info) {
+    if (error != base::PLATFORM_FILE_OK) {
+      callback.Run(error);
+      return;
+    }
+    if (file_info.is_directory) {
+      // If not a directory, failed with appropriate error code.
+      callback.Run(base::PLATFORM_FILE_ERROR_NOT_A_FILE);
+      return;
+    }
+
+    // To use FileStreamWriter, we need to ensure the destination file exists.
+    operation_runner_->CreateFile(
+        dest_url_, false /* exclusive */,
+        base::Bind(&StreamCopyOrMoveImpl::RunAfterCreateFileForDestination,
+                   weak_factory_.GetWeakPtr(), callback));
+  }
+
+  void RunAfterCreateFileForDestination(
+      const CopyOrMoveOperationDelegate::StatusCallback& callback,
+      base::PlatformFileError error) {
+    if (error != base::PLATFORM_FILE_OK) {
+      callback.Run(error);
+      return;
+    }
+
+    DCHECK(!copy_helper_);
+    copy_helper_.reset(
+        new CopyOrMoveOperationDelegate::StreamCopyHelper(
+            reader_.Pass(), writer_.Pass(),
+            kReadBufferSize,
+            file_progress_callback_,
+            base::TimeDelta::FromMilliseconds(
+                kMinProgressCallbackInvocationSpanInMilliseconds)));
+    copy_helper_->Run(
+        base::Bind(&StreamCopyOrMoveImpl::RunAfterStreamCopy,
+                   weak_factory_.GetWeakPtr(), callback));
+  }
+
+  void RunAfterStreamCopy(
+      const CopyOrMoveOperationDelegate::StatusCallback& callback,
+      base::PlatformFileError error) {
+    if (error != base::PLATFORM_FILE_OK) {
+      callback.Run(error);
+      return;
+    }
+
+    if (operation_type_ == CopyOrMoveOperationDelegate::OPERATION_COPY) {
+      callback.Run(base::PLATFORM_FILE_OK);
+      return;
+    }
+
+    DCHECK_EQ(CopyOrMoveOperationDelegate::OPERATION_MOVE, operation_type_);
+
+    // Remove the source for finalizing move operation.
+    operation_runner_->Remove(
+        src_url_, false /* recursive */,
+        base::Bind(&StreamCopyOrMoveImpl::RunAfterRemoveForMove,
+                   weak_factory_.GetWeakPtr(), callback));
+  }
+
+  void RunAfterRemoveForMove(
+      const CopyOrMoveOperationDelegate::StatusCallback& callback,
+      base::PlatformFileError error) {
+    if (error == base::PLATFORM_FILE_ERROR_NOT_FOUND)
+      error = base::PLATFORM_FILE_OK;
+    callback.Run(error);
+  }
+
+  FileSystemOperationRunner* operation_runner_;
+  CopyOrMoveOperationDelegate::OperationType operation_type_;
+  FileSystemURL src_url_;
+  FileSystemURL dest_url_;
+  scoped_ptr<webkit_blob::FileStreamReader> reader_;
+  scoped_ptr<FileStreamWriter> writer_;
+  FileSystemOperation::CopyFileProgressCallback file_progress_callback_;
+  scoped_ptr<CopyOrMoveOperationDelegate::StreamCopyHelper> copy_helper_;
+
+  base::WeakPtrFactory<StreamCopyOrMoveImpl> weak_factory_;
+  DISALLOW_COPY_AND_ASSIGN(StreamCopyOrMoveImpl);
+};
+
 }  // namespace
 
+CopyOrMoveOperationDelegate::StreamCopyHelper::StreamCopyHelper(
+    scoped_ptr<webkit_blob::FileStreamReader> reader,
+    scoped_ptr<FileStreamWriter> writer,
+    int buffer_size,
+    const FileSystemOperation::CopyFileProgressCallback&
+        file_progress_callback,
+    const base::TimeDelta& min_progress_callback_invocation_span)
+    : reader_(reader.Pass()),
+      writer_(writer.Pass()),
+      file_progress_callback_(file_progress_callback),
+      io_buffer_(new net::IOBufferWithSize(buffer_size)),
+      num_copied_bytes_(0),
+      min_progress_callback_invocation_span_(
+          min_progress_callback_invocation_span),
+      weak_factory_(this) {
+}
+
+CopyOrMoveOperationDelegate::StreamCopyHelper::~StreamCopyHelper() {
+}
+
+void CopyOrMoveOperationDelegate::StreamCopyHelper::Run(
+    const StatusCallback& callback) {
+  file_progress_callback_.Run(0);
+  last_progress_callback_invocation_time_ = base::Time::Now();
+  Read(callback);
+}
+
+void CopyOrMoveOperationDelegate::StreamCopyHelper::Read(
+    const StatusCallback& callback) {
+  int result = reader_->Read(
+      io_buffer_.get(), io_buffer_->size(),
+      base::Bind(&StreamCopyHelper::DidRead,
+                 weak_factory_.GetWeakPtr(), callback));
+  if (result != net::ERR_IO_PENDING)
+    DidRead(callback, result);
+}
+
+void CopyOrMoveOperationDelegate::StreamCopyHelper::DidRead(
+    const StatusCallback& callback, int result) {
+  if (result < 0) {
+    callback.Run(NetErrorToPlatformFileError(result));
+    return;
+  }
+
+  if (result == 0) {
+    // Here is the EOF.
+    callback.Run(base::PLATFORM_FILE_OK);
+    return;
+  }
+
+  Write(callback, new net::DrainableIOBuffer(io_buffer_.get(), result));
+}
+
+void CopyOrMoveOperationDelegate::StreamCopyHelper::Write(
+    const StatusCallback& callback,
+    scoped_refptr<net::DrainableIOBuffer> buffer) {
+  DCHECK_GT(buffer->BytesRemaining(), 0);
+
+  int result = writer_->Write(
+      buffer.get(), buffer->BytesRemaining(),
+      base::Bind(&StreamCopyHelper::DidWrite,
+                 weak_factory_.GetWeakPtr(), callback, buffer));
+  if (result != net::ERR_IO_PENDING)
+    DidWrite(callback, buffer, result);
+}
+
+void CopyOrMoveOperationDelegate::StreamCopyHelper::DidWrite(
+    const StatusCallback& callback,
+    scoped_refptr<net::DrainableIOBuffer> buffer,
+    int result) {
+  if (result < 0) {
+    callback.Run(NetErrorToPlatformFileError(result));
+    return;
+  }
+
+  buffer->DidConsume(result);
+  num_copied_bytes_ += result;
+
+  // Check the elapsed time since last |file_progress_callback_| invocation.
+  base::Time now = base::Time::Now();
+  if (now - last_progress_callback_invocation_time_ >=
+      min_progress_callback_invocation_span_) {
+    file_progress_callback_.Run(num_copied_bytes_);
+    last_progress_callback_invocation_time_ = now;
+  }
+
+  if (buffer->BytesRemaining() > 0) {
+    Write(callback, buffer);
+    return;
+  }
+
+  Read(callback);
+}
 
 CopyOrMoveOperationDelegate::CopyOrMoveOperationDelegate(
     FileSystemContext* file_system_context,
@@ -359,7 +585,6 @@ void CopyOrMoveOperationDelegate::ProcessFile(
                    weak_factory_.GetWeakPtr(), src_url));
   } else {
     // Cross filesystem case.
-    // TODO(hidehiko): Support stream based copy. crbug.com/279287.
     base::PlatformFileError error = base::PLATFORM_FILE_ERROR_FAILED;
     CopyOrMoveFileValidatorFactory* validator_factory =
         file_system_context()->GetCopyOrMoveFileValidatorFactory(
@@ -369,11 +594,28 @@ void CopyOrMoveOperationDelegate::ProcessFile(
       return;
     }
 
-    impl = new SnapshotCopyOrMoveImpl(
-        operation_runner(), operation_type_, src_url, dest_url, option_,
-        validator_factory,
-        base::Bind(&CopyOrMoveOperationDelegate::OnCopyFileProgress,
-                   weak_factory_.GetWeakPtr(), src_url));
+    if (!validator_factory) {
+      scoped_ptr<webkit_blob::FileStreamReader> reader =
+          file_system_context()->CreateFileStreamReader(
+              src_url, 0, base::Time());
+      scoped_ptr<FileStreamWriter> writer =
+          file_system_context()->CreateFileStreamWriter(dest_url, 0);
+      if (reader && writer) {
+        impl = new StreamCopyOrMoveImpl(
+            operation_runner(), operation_type_, src_url, dest_url,
+            reader.Pass(), writer.Pass(),
+            base::Bind(&CopyOrMoveOperationDelegate::OnCopyFileProgress,
+                       weak_factory_.GetWeakPtr(), src_url));
+      }
+    }
+
+    if (!impl) {
+      impl = new SnapshotCopyOrMoveImpl(
+          operation_runner(), operation_type_, src_url, dest_url, option_,
+          validator_factory,
+          base::Bind(&CopyOrMoveOperationDelegate::OnCopyFileProgress,
+                     weak_factory_.GetWeakPtr(), src_url));
+    }
   }
 
   // Register the running task.
diff --git a/webkit/browser/fileapi/copy_or_move_operation_delegate.h b/webkit/browser/fileapi/copy_or_move_operation_delegate.h
index 1f82a34a..597e05e 100644
--- a/webkit/browser/fileapi/copy_or_move_operation_delegate.h
+++ b/webkit/browser/fileapi/copy_or_move_operation_delegate.h
@@ -9,15 +9,23 @@
 
 #include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/time/time.h"
 #include "webkit/browser/fileapi/recursive_operation_delegate.h"
 
+namespace net {
+class DrainableIOBuffer;
+class IOBufferWithSize;
+}
+
 namespace webkit_blob {
+class FileStreamReader;
 class ShareableFileReference;
 }
 
 namespace fileapi {
 
 class CopyOrMoveFileValidator;
+class FileStreamWriter;
 
 // A delegate class for recursive copy or move operations.
 class CopyOrMoveOperationDelegate
@@ -32,6 +40,43 @@ class CopyOrMoveOperationDelegate
     OPERATION_MOVE
   };
 
+  // Helper to copy a file by reader and writer streams.
+  // Export for testing.
+  class WEBKIT_STORAGE_BROWSER_EXPORT StreamCopyHelper {
+   public:
+    StreamCopyHelper(
+        scoped_ptr<webkit_blob::FileStreamReader> reader,
+        scoped_ptr<FileStreamWriter> writer,
+        int buffer_size,
+        const FileSystemOperation::CopyFileProgressCallback&
+            file_progress_callback,
+        const base::TimeDelta& min_progress_callback_invocation_span);
+    ~StreamCopyHelper();
+
+    void Run(const StatusCallback& callback);
+
+   private:
+    // Reads the content from the |reader_|.
+    void Read(const StatusCallback& callback);
+    void DidRead(const StatusCallback& callback, int result);
+
+    // Writes the content in |buffer| to |writer_|.
+    void Write(const StatusCallback& callback,
+               scoped_refptr<net::DrainableIOBuffer> buffer);
+    void DidWrite(const StatusCallback& callback,
+                  scoped_refptr<net::DrainableIOBuffer> buffer, int result);
+
+    scoped_ptr<webkit_blob::FileStreamReader> reader_;
+    scoped_ptr<FileStreamWriter> writer_;
+    FileSystemOperation::CopyFileProgressCallback file_progress_callback_;
+    scoped_refptr<net::IOBufferWithSize> io_buffer_;
+    int64 num_copied_bytes_;
+    base::Time last_progress_callback_invocation_time_;
+    base::TimeDelta min_progress_callback_invocation_span_;
+    base::WeakPtrFactory<StreamCopyHelper> weak_factory_;
+    DISALLOW_COPY_AND_ASSIGN(StreamCopyHelper);
+  };
+
   CopyOrMoveOperationDelegate(
       FileSystemContext* file_system_context,
       const FileSystemURL& src_root,
diff --git a/webkit/browser/fileapi/copy_or_move_operation_delegate_unittest.cc b/webkit/browser/fileapi/copy_or_move_operation_delegate_unittest.cc
index 322341d..5c9534d 100644
--- a/webkit/browser/fileapi/copy_or_move_operation_delegate_unittest.cc
+++ b/webkit/browser/fileapi/copy_or_move_operation_delegate_unittest.cc
@@ -7,13 +7,17 @@
 
 #include "base/basictypes.h"
 #include "base/bind.h"
+#include "base/file_util.h"
 #include "base/files/scoped_temp_dir.h"
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "base/stl_util.h"
 #include "testing/gtest/include/gtest/gtest.h"
+#include "webkit/browser/blob/file_stream_reader.h"
 #include "webkit/browser/fileapi/async_file_test_helper.h"
 #include "webkit/browser/fileapi/copy_or_move_file_validator.h"
+#include "webkit/browser/fileapi/copy_or_move_operation_delegate.h"
+#include "webkit/browser/fileapi/file_stream_writer.h"
 #include "webkit/browser/fileapi/file_system_backend.h"
 #include "webkit/browser/fileapi/file_system_context.h"
 #include "webkit/browser/fileapi/file_system_operation.h"
@@ -114,6 +118,41 @@ void RecordProgressCallback(std::vector<ProgressRecord>* records,
   records->push_back(record);
 }
 
+void RecordFileProgressCallback(std::vector<int64>* records,
+                                int64 progress) {
+  records->push_back(progress);
+}
+
+void AssignAndQuit(base::RunLoop* run_loop,
+                   base::PlatformFileError* result_out,
+                   base::PlatformFileError result) {
+  *result_out = result;
+  run_loop->Quit();
+}
+
+class ScopedThreadStopper {
+ public:
+  ScopedThreadStopper(base::Thread* thread) : thread_(thread) {
+  }
+
+  ~ScopedThreadStopper() {
+    if (thread_) {
+      // Give another chance for deleted streams to perform Close.
+      base::RunLoop run_loop;
+      thread_->message_loop_proxy()->PostTaskAndReply(
+          FROM_HERE, base::Bind(&base::DoNothing), run_loop.QuitClosure());
+      run_loop.Run();
+      thread_->Stop();
+    }
+  }
+
+  bool is_valid() const { return thread_; }
+
+ private:
+  base::Thread* thread_;
+  DISALLOW_COPY_AND_ASSIGN(ScopedThreadStopper);
+};
+
 }  // namespace
 
 class CopyOrMoveOperationTestHelper {
@@ -124,7 +163,8 @@ class CopyOrMoveOperationTestHelper {
       FileSystemType dest_type)
       : origin_(origin),
         src_type_(src_type),
-        dest_type_(dest_type) {}
+        dest_type_(dest_type),
+        message_loop_(base::MessageLoop::TYPE_IO) {}
 
   ~CopyOrMoveOperationTestHelper() {
     file_system_context_ = NULL;
@@ -662,4 +702,58 @@ TEST(LocalFileSystemCopyOrMoveOperationTest, ProgressCallback) {
   }
 }
 
+
+TEST(LocalFileSystemCopyOrMoveOperationTest, StreamCopyHelper) {
+  base::ScopedTempDir temp_dir;
+  ASSERT_TRUE(temp_dir.CreateUniqueTempDir());
+  base::FilePath source_path = temp_dir.path().AppendASCII("source");
+  const char kTestData[] = "abcdefghijklmnopqrstuvwxyz0123456789";
+  file_util::WriteFile(source_path, kTestData,
+                       arraysize(kTestData) - 1);  // Exclude trailing '\0'.
+
+  base::FilePath dest_path = temp_dir.path().AppendASCII("dest");
+  // LocalFileWriter requires the file exists. So create an empty file here.
+  file_util::WriteFile(dest_path, "", 0);
+
+  base::MessageLoop message_loop(base::MessageLoop::TYPE_IO);
+  base::Thread file_thread("file_thread");
+  ASSERT_TRUE(file_thread.Start());
+  ScopedThreadStopper thread_stopper(&file_thread);
+  ASSERT_TRUE(thread_stopper.is_valid());
+
+  scoped_refptr<base::MessageLoopProxy> task_runner =
+      file_thread.message_loop_proxy();
+
+  scoped_ptr<webkit_blob::FileStreamReader> reader(
+      webkit_blob::FileStreamReader::CreateForLocalFile(
+          task_runner.get(), source_path, 0, base::Time()));
+
+  scoped_ptr<FileStreamWriter> writer(
+      FileStreamWriter::CreateForLocalFile(task_runner.get(), dest_path, 0));
+
+  std::vector<int64> progress;
+  CopyOrMoveOperationDelegate::StreamCopyHelper helper(
+      reader.Pass(), writer.Pass(),
+      10,  // buffer size
+      base::Bind(&RecordFileProgressCallback, base::Unretained(&progress)),
+      base::TimeDelta());  // For testing, we need all the progress.
+
+  base::PlatformFileError error = base::PLATFORM_FILE_ERROR_FAILED;
+  base::RunLoop run_loop;
+  helper.Run(base::Bind(&AssignAndQuit, &run_loop, &error));
+  run_loop.Run();
+
+  EXPECT_EQ(base::PLATFORM_FILE_OK, error);
+  ASSERT_EQ(5U, progress.size());
+  EXPECT_EQ(0, progress[0]);
+  EXPECT_EQ(10, progress[1]);
+  EXPECT_EQ(20, progress[2]);
+  EXPECT_EQ(30, progress[3]);
+  EXPECT_EQ(36, progress[4]);
+
+  std::string content;
+  ASSERT_TRUE(base::ReadFileToString(dest_path, &content));
+  EXPECT_EQ(kTestData, content);
+}
+
 }  // namespace fileapi
diff --git a/webkit/browser/fileapi/dragged_file_util_unittest.cc b/webkit/browser/fileapi/dragged_file_util_unittest.cc
index af15af8..f4c029b 100644
--- a/webkit/browser/fileapi/dragged_file_util_unittest.cc
+++ b/webkit/browser/fileapi/dragged_file_util_unittest.cc
@@ -89,7 +89,8 @@ FileSystemURL GetOtherURL(FileSystemContext* file_system_context,
 
 class DraggedFileUtilTest : public testing::Test {
  public:
-  DraggedFileUtilTest() {}
+  DraggedFileUtilTest()
+      : message_loop_(base::MessageLoop::TYPE_IO) {}
 
   virtual void SetUp() {
     ASSERT_TRUE(data_dir_.CreateUniqueTempDir());