Make DownloadFileImpl handle multiple byte streams.

content/browser/download/download_file_unittest.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <utility>
 #include <vector>
 
 #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"
 #include "base/test/test_file_util.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "build/build_config.h"
 #include "content/browser/byte_stream.h"
 #include "content/browser/download/download_create_info.h"
 #include "content/browser/download/download_destination_observer.h"
 #include "content/browser/download/download_file_impl.h"
@@ skipping 13 matching lines @@
 using ::testing::AnyNumber;
 using ::testing::DoAll;
 using ::testing::InSequence;
 using ::testing::Return;
 using ::testing::SetArgPointee;
 using ::testing::StrictMock;
 
 namespace content {
 namespace {
 
+// Struct for SourceStream states verification.
+struct SourceStreamTestData {
+  SourceStreamTestData(int64_t offset, int64_t bytes_written, bool finished)
+      : offset(offset), bytes_written(bytes_written), finished(finished) {}
+  int64_t offset;
+  int64_t bytes_written;
+  bool finished;
+};
+
 std::string GetHexEncodedHashValue(crypto::SecureHash* hash_state) {
   if (!hash_state)
     return std::string();
   std::vector<char> hash_value(hash_state->GetHashLength());
   hash_state->Finish(&hash_value.front(), hash_value.size());
   return base::HexEncode(&hash_value.front(), hash_value.size());
 }
 
 class MockByteStreamReader : public ByteStreamReader {
  public:
@@ skipping 39 matching lines @@
 enum DownloadFileRenameMethodType { RENAME_AND_UNIQUIFY, RENAME_AND_ANNOTATE };
 
 // This is a test DownloadFileImpl that has no retry delay and, on Posix,
 // retries renames failed due to ACCESS_DENIED.
 class TestDownloadFileImpl : public DownloadFileImpl {
  public:
   TestDownloadFileImpl(std::unique_ptr<DownloadSaveInfo> save_info,
                        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:
   base::TimeDelta GetRetryDelayForFailedRename(int attempt_count) override {
     return base::TimeDelta::FromMilliseconds(0);
   }
 
 #if !defined(OS_WIN)
   // On Posix, we don't encounter transient errors during renames, except
   // possibly EAGAIN, which is difficult to replicate reliably. So we resort to
   // simulating a transient error using ACCESS_DENIED instead.
   bool ShouldRetryFailedRename(DownloadInterruptReason reason) override {
     return reason == DOWNLOAD_INTERRUPT_REASON_FILE_ACCESS_DENIED;
   }
 #endif
 };
 
 }  // namespace
 
 class DownloadFileTest : public testing::Test {
  public:
   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;
   static const int kDummyChildId;
   static const int kDummyRequestId;
 
   DownloadFileTest()
       : observer_(new StrictMock<MockDownloadDestinationObserver>),
         observer_factory_(observer_.get()),
         input_stream_(NULL),
+        input_stream_1_(NULL),
         bytes_(-1),
         bytes_per_sec_(-1) {}
 
   ~DownloadFileTest() override {}
 
   void SetUpdateDownloadInfo(int64_t bytes, int64_t bytes_per_sec) {
     bytes_ = bytes;
     bytes_per_sec_ = bytes_per_sec;
   }
 
@@ skipping 12 matching lines @@
     sink_callback_ = sink_callback;
   }
 
   void SetInterruptReasonCallback(const base::Closure& closure,
                                   DownloadInterruptReason* reason_p,
                                   DownloadInterruptReason reason) {
     *reason_p = reason;
     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());
 
     input_stream_ = new StrictMock<MockByteStreamReader>();
 
     // TODO: Need to actually create a function that'll set the variables
     // based on the inputs from the callback.
     EXPECT_CALL(*input_stream_, RegisterCallback(_))
         .WillOnce(Invoke(this, &DownloadFileTest::RegisterCallback))
         .RetiresOnSaturation();
 
     std::unique_ptr<DownloadSaveInfo> save_info(new DownloadSaveInfo());
     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))
         .RetiresOnSaturation();
 
     base::WeakPtrFactory<DownloadFileTest> weak_ptr_factory(this);
     DownloadInterruptReason result = DOWNLOAD_INTERRUPT_REASON_NONE;
     base::RunLoop loop_runner;
     download_file_->Initialize(base::Bind(
         &DownloadFileTest::SetInterruptReasonCallback,
@@ skipping 10 matching lines @@
     // Make sure the data has been properly written to disk.
     std::string disk_data;
     EXPECT_TRUE(base::ReadFileToString(download_file_->FullPath(), &disk_data));
     EXPECT_EQ(expected_data_, disk_data);
 
     // Make sure the Browser and File threads outlive the DownloadFile
     // to satisfy thread checks inside it.
     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;
     }
   }
 
   void VerifyStreamAndSize() {
     ::testing::Mock::VerifyAndClearExpectations(input_stream_);
     int64_t size;
     EXPECT_TRUE(base::GetFileSize(download_file_->FullPath(), &size));
     EXPECT_EQ(expected_data_.size(), static_cast<size_t>(size));
   }
 
   // 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))
         .RetiresOnSaturation();
     sink_callback_.Run();
     VerifyStreamAndSize();
   }
 
   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) {
       EXPECT_CALL(*(observer_.get()),
                   MockDestinationCompleted(_, expected_hash));
       base::RunLoop().RunUntilIdle();
       ::testing::Mock::VerifyAndClearExpectations(observer_.get());
       EXPECT_CALL(*(observer_.get()), DestinationUpdate(_, _))
           .Times(AnyNumber())
           .WillRepeatedly(
@@ skipping 53 matching lines @@
         base::Bind(&DownloadFileTest::SetRenameResult,
                    base::Unretained(this),
                    loop_runner.QuitClosure(),
                    &result_reason,
                    result_path_p);
     InvokeRenameMethod(method, full_path, completion_callback);
     loop_runner.Run();
     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 =
+        download_file_->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();
+  }
+
+  void VerifySourceStreamsStates(const SourceStreamTestData& data) {
+    DCHECK(download_file_->source_streams_.find(data.offset) !=
+           download_file_->source_streams_.end());
+    DownloadFileImpl::SourceStream* stream =
+        download_file_->source_streams_[data.offset].get();
+    DCHECK(stream);
+    EXPECT_EQ(data.offset, stream->offset());
+    EXPECT_EQ(data.bytes_written, stream->bytes_written());
+    EXPECT_EQ(data.finished, stream->is_finished());
+  }
+
+  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_;
 
   // Latest update sent to the observer.
   int64_t bytes_;
   int64_t bytes_per_sec_;
 
  private:
   void SetRenameResult(const base::Closure& closure,
                        DownloadInterruptReason* reason_p,
@@ skipping 37 matching lines @@
 // the value parameter.
 INSTANTIATE_TEST_CASE_P(DownloadFile,
                         DownloadFileTestWithRename,
                         ::testing::Values(RENAME_AND_ANNOTATE,
                                           RENAME_AND_UNIQUIFY));
 
 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[] =
     "E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855";
 
 const uint32_t DownloadFileTest::kDummyDownloadId = 23;
 const int DownloadFileTest::kDummyChildId = 3;
 const int DownloadFileTest::kDummyRequestId = 67;
 
 // Rename the file before any data is downloaded, after some has, after it all
@@ skipping 310 matching lines @@
   DestroyDownloadFile(0);
 }
 
 TEST_F(DownloadFileTest, StreamNonEmptySuccess) {
   ASSERT_TRUE(CreateDownloadFile(0, true));
   base::FilePath initial_path(download_file_->FullPath());
   EXPECT_TRUE(base::PathExists(initial_path));
 
   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();
   base::RunLoop().RunUntilIdle();
   DestroyDownloadFile(0);
 }
 
 TEST_F(DownloadFileTest, StreamNonEmptyError) {
   ASSERT_TRUE(CreateDownloadFile(0, true));
   base::FilePath initial_path(download_file_->FullPath());
   EXPECT_TRUE(base::PathExists(initial_path));
 
   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(
                   DOWNLOAD_INTERRUPT_REASON_NETWORK_DISCONNECTED, _, _))
       .WillOnce(InvokeWithoutArgs(
           this, &DownloadFileTest::ConfirmUpdateDownloadInfo));
 
   // If this next EXPECT_CALL fails flakily, it's probably a real failure.
   // We'll be getting a stream of UpdateDownload calls from the timer, and
   // the last one may have the correct information even if the failure
   // doesn't produce an update, as the timer update may have triggered at the
   // same time.
   EXPECT_CALL(*(observer_.get()),
               CurrentUpdateStatus(strlen(kTestData1) + strlen(kTestData2), _));
 
   sink_callback_.Run();
   base::RunLoop().RunUntilIdle();
   VerifyStreamAndSize();
   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(_, _));
+
+  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));
+
+  download_file_->AddByteStream(
+      std::unique_ptr<MockByteStreamReader>(input_stream_1_), stream_0_length);
+  sink_callback_.Run();
+  base::RunLoop().RunUntilIdle();
+
+  SourceStreamTestData stream_data_0(0, stream_0_length, true);
+  SourceStreamTestData stream_data_1(stream_0_length, stream_1_length, true);
+  VerifySourceStreamsStates(stream_data_0);
+  VerifySourceStreamsStates(stream_data_1);
+  EXPECT_EQ(stream_0_length + stream_1_length, TotalBytesReceived());
+
+  DestroyDownloadFile(0);
+}
+
+// Activate and deplete one stream, later add the second stream.
+TEST_F(DownloadFileTest, MutipleStreamsOneStreamFirst) {
+  PrepareMultipleStreams();
+
+  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));
+
+  // Deplete the first stream.
+  sink_callback_.Run();
+  base::RunLoop().RunUntilIdle();
+
+  SourceStreamTestData stream_data_0(0, stream_0_length, true);
+  SourceStreamTestData stream_data_1(stream_0_length, 0, false);
+  VerifySourceStreamsStates(stream_data_0);
+  VerifySourceStreamsStates(stream_data_1);
+  EXPECT_EQ(stream_0_length, TotalBytesReceived());
+
+  // Won't inform the observer until 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();
+
+  stream_data_1.bytes_written = stream_1_length;
+  stream_data_1.finished = true;
+  VerifySourceStreamsStates(stream_data_0);
+  VerifySourceStreamsStates(stream_data_1);
+  EXPECT_EQ(stream_0_length + stream_1_length, TotalBytesReceived());
+
+  DestroyDownloadFile(0);
+}
+
 }  // namespace content