Make DownloadFileImpl handle multiple byte streams.

content/browser/download/download_file_unittest.cc

Index: content/browser/download/download_file_unittest.cc
diff --git a/content/browser/download/download_file_unittest.cc b/content/browser/download/download_file_unittest.cc
index d4e9ddc9cb1870f998dd5c1fe78d5bb6686f0105..ef0b2916e0f20eb160eca5cf56cace9e15e1d9ed 100644
--- a/content/browser/download/download_file_unittest.cc
+++ b/content/browser/download/download_file_unittest.cc
@@ -11,6 +11,7 @@
 #include "base/files/file.h"
 #include "base/files/file_util.h"
 #include "base/location.h"
+#include "base/memory/ptr_util.h"
 #include "base/run_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_number_conversions.h"
@@ -103,11 +104,13 @@ class TestDownloadFileImpl : public DownloadFileImpl {
                        const base::FilePath& default_downloads_directory,
                        std::unique_ptr<ByteStreamReader> stream,
                        const net::NetLogWithSource& net_log,
+                       bool is_sparse_file,
                        base::WeakPtr<DownloadDestinationObserver> observer)
       : DownloadFileImpl(std::move(save_info),
                          default_downloads_directory,
                          std::move(stream),
                          net_log,
+                         is_sparse_file,
                          observer) {}
 
  protected:
@@ -132,6 +135,8 @@ class DownloadFileTest : public testing::Test {
   static const char kTestData1[];
   static const char kTestData2[];
   static const char kTestData3[];
+  static const char kTestData4[];
+  static const char kTestData5[];
   static const char kDataHash[];
   static const char kEmptyHash[];
   static const uint32_t kDummyDownloadId;
@@ -142,6 +147,7 @@ class DownloadFileTest : public testing::Test {
       : observer_(new StrictMock<MockDownloadDestinationObserver>),
         observer_factory_(observer_.get()),
         input_stream_(NULL),
+        input_stream_1_(NULL),
         bytes_(-1),
         bytes_per_sec_(-1) {}
 
@@ -174,7 +180,9 @@ class DownloadFileTest : public testing::Test {
     closure.Run();
   }
 
-  bool CreateDownloadFile(int offset, bool calculate_hash) {
+  bool CreateDownloadFile(int offset,
+                          bool calculate_hash,
+                          bool is_sparse_file = false) {
     // There can be only one.
     DCHECK(!download_file_.get());
 
@@ -190,7 +198,8 @@ class DownloadFileTest : public testing::Test {
     download_file_.reset(new TestDownloadFileImpl(
         std::move(save_info), base::FilePath(),
         std::unique_ptr<ByteStreamReader>(input_stream_),
-        net::NetLogWithSource(), observer_factory_.GetWeakPtr()));
+        net::NetLogWithSource(), is_sparse_file,
+        observer_factory_.GetWeakPtr()));
 
     EXPECT_CALL(*input_stream_, Read(_, _))
         .WillOnce(Return(ByteStreamReader::STREAM_EMPTY))
@@ -221,23 +230,38 @@ class DownloadFileTest : public testing::Test {
     download_file_.reset();
   }
 
-  // Setup the stream to do be a data append; don't actually trigger
-  // the callback or do verifications.
-  void SetupDataAppend(const char **data_chunks, size_t num_chunks,
-                       ::testing::Sequence s) {
-    DCHECK(input_stream_);
+  // Setup the stream to append data or write from |offset| to the file.
+  // Don't actually trigger the callback or do verifications.
+  void SetupDataAppend(const char** data_chunks,
+                       size_t num_chunks,
+                       MockByteStreamReader* stream_reader,
+                       ::testing::Sequence s,
+                       int64_t offset = -1) {
+    DCHECK(stream_reader);
+    size_t current_pos = static_cast<size_t>(offset);
     for (size_t i = 0; i < num_chunks; i++) {
       const char *source_data = data_chunks[i];
       size_t length = strlen(source_data);
       scoped_refptr<net::IOBuffer> data = new net::IOBuffer(length);
       memcpy(data->data(), source_data, length);
-      EXPECT_CALL(*input_stream_, Read(_, _))
+      EXPECT_CALL(*stream_reader, Read(_, _))
           .InSequence(s)
-          .WillOnce(DoAll(SetArgPointee<0>(data),
-                          SetArgPointee<1>(length),
+          .WillOnce(DoAll(SetArgPointee<0>(data), SetArgPointee<1>(length),
                           Return(ByteStreamReader::STREAM_HAS_DATA)))
           .RetiresOnSaturation();
-      expected_data_ += source_data;
+
+      if (offset < 0) {
+        // Append data.
+        expected_data_ += source_data;
+        continue;
+      }
+
+      // Write from offset. May fill holes with '\0'.
+      size_t new_len = current_pos + length;
+      if (new_len > expected_data_.size())
+        expected_data_.append(new_len - expected_data_.size(), '\0');
+      expected_data_.replace(current_pos, length, source_data);
+      current_pos += length;
     }
   }
 
@@ -251,7 +275,7 @@ class DownloadFileTest : public testing::Test {
   // TODO(rdsmith): Manage full percentage issues properly.
   void AppendDataToFile(const char **data_chunks, size_t num_chunks) {
     ::testing::Sequence s1;
-    SetupDataAppend(data_chunks, num_chunks, s1);
+    SetupDataAppend(data_chunks, num_chunks, input_stream_, s1);
     EXPECT_CALL(*input_stream_, Read(_, _))
         .InSequence(s1)
         .WillOnce(Return(ByteStreamReader::STREAM_EMPTY))
@@ -261,24 +285,24 @@ class DownloadFileTest : public testing::Test {
   }
 
   void SetupFinishStream(DownloadInterruptReason interrupt_reason,
-                       ::testing::Sequence s) {
-    EXPECT_CALL(*input_stream_, Read(_, _))
+                         MockByteStreamReader* stream_reader,
+                         ::testing::Sequence s) {
+    EXPECT_CALL(*stream_reader, Read(_, _))
         .InSequence(s)
         .WillOnce(Return(ByteStreamReader::STREAM_COMPLETE))
         .RetiresOnSaturation();
-    EXPECT_CALL(*input_stream_, GetStatus())
+    EXPECT_CALL(*stream_reader, GetStatus())
         .InSequence(s)
         .WillOnce(Return(interrupt_reason))
         .RetiresOnSaturation();
-    EXPECT_CALL(*input_stream_, RegisterCallback(_))
-        .RetiresOnSaturation();
+    EXPECT_CALL(*stream_reader, RegisterCallback(_)).RetiresOnSaturation();
   }
 
   void FinishStream(DownloadInterruptReason interrupt_reason,
                     bool check_observer,
                     const std::string& expected_hash) {
     ::testing::Sequence s1;
-    SetupFinishStream(interrupt_reason, s1);
+    SetupFinishStream(interrupt_reason, input_stream_, s1);
     sink_callback_.Run();
     VerifyStreamAndSize();
     if (check_observer) {
@@ -352,16 +376,65 @@ class DownloadFileTest : public testing::Test {
     return result_reason;
   }
 
+  // Prepare two byte streams to write to the same file sink.
+  void PrepareMultipleStreams() {
+    // Create a sparse file.
+    ASSERT_TRUE(CreateDownloadFile(0, true, true));
+    base::FilePath initial_path(download_file_->FullPath());
+    EXPECT_TRUE(base::PathExists(initial_path));
+    DCHECK(download_file_);
+
+    const char* stream_0_data[] = {kTestData1, kTestData2};
+    const char* stream_1_data[] = {kTestData4, kTestData5};
+    size_t stream_1_offset = strlen(kTestData1) + strlen(kTestData2);
+
+    // Register second SourceStream entry for the second stream.
+    // The first stream should be registered in ctor of DownloadFile.
+    DownloadFileImpl::SourceStreams& source_streams =
+        DownloadFileTest::source_streams();
+    EXPECT_EQ(static_cast<size_t>(1), source_streams.size());
+    source_streams[stream_1_offset] =
+        base::MakeUnique<DownloadFileImpl::SourceStream>(stream_1_offset, 0);
+
+    // Create the second byte stream. Will be moved to DownloadFile.
+    input_stream_1_ = new MockByteStreamReader();
+
+    // Fill up two streams and setup expectation.
+    ::testing::Sequence s0;
+    ::testing::Sequence s1;
+    SetupDataAppend(stream_1_data, 2, input_stream_1_, s1, stream_1_offset);
+    SetupDataAppend(stream_0_data, 2, input_stream_, s0, 0);
+    SetupFinishStream(DOWNLOAD_INTERRUPT_REASON_NONE, input_stream_1_, s1);
+    SetupFinishStream(DOWNLOAD_INTERRUPT_REASON_NONE, input_stream_, s0);
+
+    // Every ByteStreamReader should call RegisterCallback twice.
+    // One to set the callback, the other to null the callback.
+    EXPECT_CALL(*input_stream_1_, RegisterCallback(_)).RetiresOnSaturation();
+  }
+
+  DownloadFileImpl::SourceStreams& source_streams() {
+    DCHECK(download_file_);
+    return download_file_->source_streams_;
+  }
+
+  int64_t TotalBytesReceived() const {
+    DCHECK(download_file_);
+    return download_file_->TotalBytesReceived();
+  }
+
   std::unique_ptr<StrictMock<MockDownloadDestinationObserver>> observer_;
   base::WeakPtrFactory<DownloadDestinationObserver> observer_factory_;
 
   // DownloadFile instance we are testing.
-  std::unique_ptr<DownloadFile> download_file_;
+  std::unique_ptr<DownloadFileImpl> download_file_;
 
   // Stream for sending data into the download file.
   // Owned by download_file_; will be alive for lifetime of download_file_.
   StrictMock<MockByteStreamReader>* input_stream_;
 
+  // A second byte stream to test multiple stream write.
+  MockByteStreamReader* input_stream_1_;
+
   // Sink callback data for stream.
   base::Closure sink_callback_;
 
@@ -419,6 +492,8 @@ const char DownloadFileTest::kTestData1[] =
     "Let's write some data to the file!\n";
 const char DownloadFileTest::kTestData2[] = "Writing more data.\n";
 const char DownloadFileTest::kTestData3[] = "Final line.";
+const char DownloadFileTest::kTestData4[] = "abcdefg";
+const char DownloadFileTest::kTestData5[] = "01234";
 const char DownloadFileTest::kDataHash[] =
     "CBF68BF10F8003DB86B31343AFAC8C7175BD03FB5FC905650F8C80AF087443A8";
 const char DownloadFileTest::kEmptyHash[] =
@@ -749,8 +824,8 @@ TEST_F(DownloadFileTest, StreamNonEmptySuccess) {
 
   const char* chunks1[] = { kTestData1, kTestData2 };
   ::testing::Sequence s1;
-  SetupDataAppend(chunks1, 2, s1);
-  SetupFinishStream(DOWNLOAD_INTERRUPT_REASON_NONE, s1);
+  SetupDataAppend(chunks1, 2, input_stream_, s1);
+  SetupFinishStream(DOWNLOAD_INTERRUPT_REASON_NONE, input_stream_, s1);
   EXPECT_CALL(*(observer_.get()), MockDestinationCompleted(_, _));
   sink_callback_.Run();
   VerifyStreamAndSize();
@@ -765,8 +840,9 @@ TEST_F(DownloadFileTest, StreamNonEmptyError) {
 
   const char* chunks1[] = { kTestData1, kTestData2 };
   ::testing::Sequence s1;
-  SetupDataAppend(chunks1, 2, s1);
-  SetupFinishStream(DOWNLOAD_INTERRUPT_REASON_NETWORK_DISCONNECTED, s1);
+  SetupDataAppend(chunks1, 2, input_stream_, s1);
+  SetupFinishStream(DOWNLOAD_INTERRUPT_REASON_NETWORK_DISCONNECTED,
+                    input_stream_, s1);
 
   EXPECT_CALL(*(observer_.get()),
               MockDestinationError(
@@ -788,4 +864,82 @@ TEST_F(DownloadFileTest, StreamNonEmptyError) {
   DestroyDownloadFile(0);
 }
 
+// Tests for concurrent streams handling, used for parallel download.
+//
+// Activate both streams at the same time.
+TEST_F(DownloadFileTest, MutipleStreamsWrite) {
+  PrepareMultipleStreams();
+  EXPECT_CALL(*(observer_.get()), MockDestinationCompleted(_, _));
+
+  DownloadFileImpl::SourceStreams& source_streams =
+      DownloadFileTest::source_streams();
+  int64_t stream_0_length =
+      static_cast<int64_t>(strlen(kTestData1) + strlen(kTestData2));
+  int64_t stream_1_length =
+      static_cast<int64_t>(strlen(kTestData4) + strlen(kTestData5));
+  DownloadFileImpl::SourceStream* stream0 = source_streams[0].get();
+  DownloadFileImpl::SourceStream* stream1 =
+      source_streams[stream_0_length].get();
+
+  download_file_->AddByteStream(
+      std::unique_ptr<MockByteStreamReader>(input_stream_1_), stream_0_length);
+  sink_callback_.Run();
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_EQ(stream_0_length, stream0->bytes_received());
+  EXPECT_EQ(stream_1_length, stream1->bytes_received());
+  EXPECT_TRUE(stream0->is_finished());
+  EXPECT_TRUE(stream1->is_finished());
+  EXPECT_EQ(0, stream0->offset());
+  EXPECT_EQ(stream_0_length, stream1->offset());
+  EXPECT_EQ(stream0->bytes_received() + stream1->bytes_received(),
+            TotalBytesReceived());
+
+  DestroyDownloadFile(0);
+}
+
+// Activate and deplete one stream, later add the second stream.
+TEST_F(DownloadFileTest, MutipleStreamsOneStreamFirst) {
+  PrepareMultipleStreams();
+
+  DownloadFileImpl::SourceStreams& source_streams =
+      DownloadFileTest::source_streams();
+  int64_t stream_0_length =
+      static_cast<int64_t>(strlen(kTestData1) + strlen(kTestData2));
+  int64_t stream_1_length =
+      static_cast<int64_t>(strlen(kTestData4) + strlen(kTestData5));
+  DownloadFileImpl::SourceStream* stream0 = source_streams[0].get();
+  DownloadFileImpl::SourceStream* stream1 =
+      source_streams[stream_0_length].get();
+
+  // Deplete the first stream.
+  sink_callback_.Run();
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(stream_0_length, stream0->bytes_received());
+  EXPECT_EQ(0, stream1->bytes_received());
+  EXPECT_TRUE(stream0->is_finished());
+  EXPECT_FALSE(stream1->is_finished());
+  EXPECT_EQ(stream_0_length, stream1->offset());
+  EXPECT_EQ(stream0->bytes_received(), TotalBytesReceived());
+
+  // Won't inform the observer utill the second stream is depleted.
+  EXPECT_CALL(*(observer_.get()), MockDestinationCompleted(_, _));
+
+  // Drain the second stream after the first stream is depleted.
+  download_file_->AddByteStream(
+      std::unique_ptr<MockByteStreamReader>(input_stream_1_), stream_0_length);
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_EQ(stream_0_length, stream0->bytes_received());
+  EXPECT_EQ(stream_1_length, stream1->bytes_received());
+  EXPECT_TRUE(stream0->is_finished());
+  EXPECT_TRUE(stream1->is_finished());
+  EXPECT_EQ(0, stream0->offset());
+  EXPECT_EQ(stream_0_length, stream1->offset());
+  EXPECT_EQ(stream0->bytes_received() + stream1->bytes_received(),
+            TotalBytesReceived());
+
+  DestroyDownloadFile(0);
+}
+
 }  // namespace content