Introduce BundledExchangesBlobDataSource

This class will be used to read partial body data of a web bundle
response from server.

Bug: 1018640
Change-Id: Ib945b9152527be2dfd2fc43dc1e6ef8fcc09b12f
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1886271Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Reviewed-by: Kunihiko Sakamoto <ksakamoto@chromium.org>
Commit-Queue: Tsuyoshi Horo <horo@chromium.org>
Cr-Commit-Position: refs/heads/master@{#711146}

content/browser/BUILD.gn

@@ -1866,6 +1866,8 @@ jumbo_source_set("browser") {
     "web_contents/web_drag_dest_mac.mm",
     "web_contents/web_drag_utils_win.cc",
     "web_contents/web_drag_utils_win.h",
+    "web_package/bundled_exchanges_blob_data_source.cc",
+    "web_package/bundled_exchanges_blob_data_source.h",
     "web_package/bundled_exchanges_handle.cc",
     "web_package/bundled_exchanges_handle.h",
     "web_package/bundled_exchanges_handle_tracker.cc",

content/browser/web_package/bundled_exchanges_blob_data_source.cc

+// Copyright 2019 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 "content/browser/web_package/bundled_exchanges_blob_data_source.h"
+
+#include "base/bit_cast.h"
+#include "base/callback_helpers.h"
+#include "base/memory/ref_counted.h"
+#include "base/numerics/checked_math.h"
+#include "base/numerics/safe_conversions.h"
+#include "base/task/post_task.h"
+#include "base/task/task_traits.h"
+#include "content/public/browser/browser_task_traits.h"
+#include "content/public/browser/browser_thread.h"
+#include "mojo/public/cpp/bindings/associated_remote.h"
+#include "net/base/io_buffer.h"
+#include "storage/browser/blob/blob_builder_from_stream.h"
+#include "storage/browser/blob/blob_data_handle.h"
+#include "storage/browser/blob/blob_reader.h"
+#include "storage/browser/blob/mojo_blob_reader.h"
+
+namespace content {
+namespace {
+
+class MojoBlobReaderDelegate : public storage::MojoBlobReader::Delegate {
+ public:
+  using CompletionCallback = base::OnceCallback<void(net::Error net_error)>;
+  explicit MojoBlobReaderDelegate(CompletionCallback completion_callback)
+      : completion_callback_(std::move(completion_callback)) {}
+  ~MojoBlobReaderDelegate() override = default;
+  RequestSideData DidCalculateSize(uint64_t total_size,
+                                   uint64_t content_size) override {
+    return DONT_REQUEST_SIDE_DATA;
+  }
+  void DidRead(int num_bytes) override {}
+  void OnComplete(net::Error result, uint64_t total_written_bytes) override {
+    DCHECK_CURRENTLY_ON(BrowserThread::IO);
+    std::move(completion_callback_).Run(result);
+  }
+
+ private:
+  CompletionCallback completion_callback_;
+  DISALLOW_COPY_AND_ASSIGN(MojoBlobReaderDelegate);
+};
+
+void OnReadComplete(
+    data_decoder::mojom::BundleDataSource::ReadCallback callback,
+    std::unique_ptr<storage::BlobReader> blob_reader,
+    scoped_refptr<net::IOBufferWithSize> io_buf,
+    int bytes_read) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (bytes_read != io_buf->size()) {
+    std::move(callback).Run(base::nullopt);
+    return;
+  }
+  std::vector<uint8_t> vec;
+  vec.assign(bit_cast<uint8_t*>(io_buf->data()),
+             bit_cast<uint8_t*>(io_buf->data()) + bytes_read);
+  std::move(callback).Run(std::move(vec));
+}
+
+void OnCalculateSizeComplete(
+    uint64_t offset,
+    uint64_t length,
+    data_decoder::mojom::BundleDataSource::ReadCallback callback,
+    std::unique_ptr<storage::BlobReader> blob_reader,
+    int net_error) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (net_error != net::OK) {
+    std::move(callback).Run(base::nullopt);
+    return;
+  }
+  auto set_read_range_status = blob_reader->SetReadRange(offset, length);
+  if (set_read_range_status != storage::BlobReader::Status::DONE) {
+    DCHECK_EQ(set_read_range_status, storage::BlobReader::Status::NET_ERROR);
+    std::move(callback).Run(base::nullopt);
+    return;
+  }
+  auto* raw_blob_reader = blob_reader.get();
+  auto io_buf =
+      base::MakeRefCounted<net::IOBufferWithSize>(static_cast<size_t>(length));
+  auto on_read_callback = base::AdaptCallbackForRepeating(base::BindOnce(
+      &OnReadComplete, std::move(callback), std::move(blob_reader), io_buf));
+  int bytes_read;
+  storage::BlobReader::Status read_status = raw_blob_reader->Read(
+      io_buf.get(), io_buf->size(), &bytes_read, on_read_callback);
+  if (read_status != storage::BlobReader::Status::IO_PENDING) {
+    on_read_callback.Run(bytes_read);
+  }
+}
+
+bool IsValidRange(uint64_t offset, uint64_t length, int64_t content_length) {
+  int64_t offset_plus_length;
+  if (!base::CheckAdd(offset, length).AssignIfValid(&offset_plus_length))
+    return false;
+  return offset_plus_length <= content_length;
+}
+
+}  // namespace
+
+BundledExchangesBlobDataSource::BundledExchangesBlobDataSource(
+    int64_t content_length,
+    mojo::ScopedDataPipeConsumerHandle outer_response_body,
+    network::mojom::URLLoaderClientEndpointsPtr endpoints,
+    BrowserContext::BlobContextGetter blob_context_getter,
+    mojo::PendingReceiver<data_decoder::mojom::BundleDataSource>
+        pending_receiver) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  DCHECK_GT(content_length, 0);
+  base::PostTask(
+      FROM_HERE, {BrowserThread::IO},
+      base::BindOnce(&BundledExchangesBlobDataSource::CreateCoreOnIO,
+                     weak_factory_.GetWeakPtr(), content_length,
+                     std::move(outer_response_body), std::move(endpoints),
+                     std::move(blob_context_getter),
+                     std::move(pending_receiver)));
+}
+
+BundledExchangesBlobDataSource::~BundledExchangesBlobDataSource() {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  DeleteCore(std::move(core_));
+
+  auto tasks = std::move(pending_get_core_tasks_);
+  for (auto& task : tasks) {
+    std::move(task).Run();
+  }
+}
+
+// static
+void BundledExchangesBlobDataSource::CreateCoreOnIO(
+    base::WeakPtr<BundledExchangesBlobDataSource> weak_ptr,
+    int64_t content_length,
+    mojo::ScopedDataPipeConsumerHandle outer_response_body,
+    network::mojom::URLLoaderClientEndpointsPtr endpoints,
+    BrowserContext::BlobContextGetter blob_context_getter,
+    mojo::PendingReceiver<data_decoder::mojom::BundleDataSource>
+        pending_receiver) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  auto core = std::make_unique<BlobDataSourceCore>(
+      content_length, std::move(endpoints), std::move(blob_context_getter),
+      std::move(pending_receiver));
+  core->Start(std::move(outer_response_body));
+  auto weak_core = core->GetWeakPtr();
+  base::PostTask(FROM_HERE, {BrowserThread::UI},
+                 base::BindOnce(&BundledExchangesBlobDataSource::SetCoreOnUI,
+                                std::move(weak_ptr), std::move(weak_core),
+                                std::move(core)));
+}
+
+// static
+void BundledExchangesBlobDataSource::SetCoreOnUI(
+    base::WeakPtr<BundledExchangesBlobDataSource> weak_ptr,
+    base::WeakPtr<BlobDataSourceCore> weak_core,
+    std::unique_ptr<BlobDataSourceCore> core) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  if (!weak_ptr) {
+    // This happens when the BundledExchangesBlobDataSource was deleted before
+    // SetCoreOnUI() is called.
+    DeleteCore(std::move(core));
+    return;
+  }
+  weak_ptr->SetCoreOnUIImpl(std::move(weak_core), std::move(core));
+}
+
+// static
+void BundledExchangesBlobDataSource::DeleteCore(
+    std::unique_ptr<BlobDataSourceCore> core) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  if (!core)
+    return;
+  base::PostTask(FROM_HERE, {BrowserThread::IO},
+                 base::BindOnce([](std::unique_ptr<BlobDataSourceCore> core) {},
+                                std::move(core)));
+}
+
+void BundledExchangesBlobDataSource::ReadToDataPipe(
+    uint64_t offset,
+    uint64_t length,
+    mojo::ScopedDataPipeProducerHandle producer_handle,
+    CompletionCallback callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  WaitForCore(
+      base::BindOnce(&BundledExchangesBlobDataSource::ReadToDataPipeImpl,
+                     base::Unretained(this), offset, length,
+                     std::move(producer_handle), std::move(callback)));
+}
+
+void BundledExchangesBlobDataSource::WaitForCore(base::OnceClosure callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  if (core_) {
+    std::move(callback).Run();
+    return;
+  }
+  pending_get_core_tasks_.push_back(std::move(callback));
+}
+
+void BundledExchangesBlobDataSource::SetCoreOnUIImpl(
+    base::WeakPtr<BlobDataSourceCore> weak_core,
+    std::unique_ptr<BlobDataSourceCore> core) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  weak_core_ = std::move(weak_core);
+  core_ = std::move(core);
+
+  auto tasks = std::move(pending_get_core_tasks_);
+  for (auto& task : tasks) {
+    std::move(task).Run();
+  }
+}
+
+void BundledExchangesBlobDataSource::ReadToDataPipeImpl(
+    uint64_t offset,
+    uint64_t length,
+    mojo::ScopedDataPipeProducerHandle producer_handle,
+    CompletionCallback callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  if (!core_) {
+    // This happens when |this| was deleted before SetCoreOnUI() is called.
+    std::move(callback).Run(net::ERR_FAILED);
+    return;
+  }
+  CompletionCallback wrapped_callback = base::BindOnce(
+      [](CompletionCallback callback, net::Error net_error) {
+        DCHECK_CURRENTLY_ON(BrowserThread::IO);
+        base::PostTask(FROM_HERE, {BrowserThread::UI},
+                       base::BindOnce(std::move(callback), net_error));
+      },
+      std::move(callback));
+  base::PostTask(FROM_HERE, {BrowserThread::IO},
+                 base::BindOnce(&BlobDataSourceCore::ReadToDataPipe, weak_core_,
+                                offset, length, std::move(producer_handle),
+                                std::move(wrapped_callback)));
+}
+
+BundledExchangesBlobDataSource::BlobDataSourceCore::BlobDataSourceCore(
+    int64_t content_length,
+    network::mojom::URLLoaderClientEndpointsPtr endpoints,
+    BrowserContext::BlobContextGetter blob_context_getter,
+    mojo::PendingReceiver<data_decoder::mojom::BundleDataSource>
+        pending_receiver)
+    : content_length_(content_length),
+      endpoints_(std::move(endpoints)),
+      receiver_(this, std::move(pending_receiver)),
+      blob_builder_from_stream_(
+          std::make_unique<storage::BlobBuilderFromStream>(
+              std::move(blob_context_getter).Run(),
+              "" /* content_type */,
+              "" /* content_disposition */,
+              base::BindOnce(&BundledExchangesBlobDataSource::
+                                 BlobDataSourceCore::StreamingBlobDone,
+                             base::Unretained(this)))) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  DCHECK_GT(content_length_, 0);
+}
+
+BundledExchangesBlobDataSource::BlobDataSourceCore::~BlobDataSourceCore() {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (blob_builder_from_stream_)
+    std::move(blob_builder_from_stream_)->Abort();
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::Start(
+    mojo::ScopedDataPipeConsumerHandle outer_response_body) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  blob_builder_from_stream_->Start(
+      content_length_, std::move(outer_response_body),
+      mojo::NullAssociatedRemote() /*  progress_client */);
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::ReadToDataPipe(
+    uint64_t offset,
+    uint64_t length,
+    mojo::ScopedDataPipeProducerHandle producer_handle,
+    CompletionCallback callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (!IsValidRange(offset, length, content_length_)) {
+    std::move(callback).Run(net::ERR_REQUEST_RANGE_NOT_SATISFIABLE);
+    return;
+  }
+  WaitForBlob(
+      base::BindOnce(&BundledExchangesBlobDataSource::BlobDataSourceCore::
+                         OnBlobReadyForReadToDataPipe,
+                     base::Unretained(this), offset, length,
+                     std::move(producer_handle), std::move(callback)));
+}
+
+base::WeakPtr<BundledExchangesBlobDataSource::BlobDataSourceCore>
+BundledExchangesBlobDataSource::BlobDataSourceCore::GetWeakPtr() {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  return weak_factory_.GetWeakPtr();
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::GetSize(
+    GetSizeCallback callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  std::move(callback).Run(content_length_);
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::Read(
+    uint64_t offset,
+    uint64_t length,
+    ReadCallback callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (!IsValidRange(offset, length, content_length_)) {
+    std::move(callback).Run(base::nullopt);
+    return;
+  }
+  WaitForBlob(base::BindOnce(
+      &BundledExchangesBlobDataSource::BlobDataSourceCore::OnBlobReadyForRead,
+      base::Unretained(this), offset, length, std::move(callback)));
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::StreamingBlobDone(
+    storage::BlobBuilderFromStream* builder,
+    std::unique_ptr<storage::BlobDataHandle> result) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  // Content length mismatch is treated as an error.
+  if (result && (result->size() == base::checked_cast<size_t>(content_length_)))
+    blob_ = std::move(result);
+  blob_builder_from_stream_.reset();
+
+  auto tasks = std::move(pending_get_blob_tasks_);
+  for (auto& task : tasks) {
+    std::move(task).Run();
+  }
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::WaitForBlob(
+    base::OnceClosure callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (!blob_builder_from_stream_) {
+    std::move(callback).Run();
+    return;
+  }
+  pending_get_blob_tasks_.push_back(std::move(callback));
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::OnBlobReadyForRead(
+    uint64_t offset,
+    uint64_t length,
+    ReadCallback callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (!blob_) {
+    std::move(callback).Run(base::nullopt);
+    return;
+  }
+  auto blob_reader = blob_->CreateReader();
+  auto* raw_blob_reader = blob_reader.get();
+  auto on_calculate_complete = base::AdaptCallbackForRepeating(
+      base::BindOnce(&OnCalculateSizeComplete, offset, length,
+                     std::move(callback), std::move(blob_reader)));
+  auto status = raw_blob_reader->CalculateSize(on_calculate_complete);
+  if (status != storage::BlobReader::Status::IO_PENDING) {
+    on_calculate_complete.Run(status == storage::BlobReader::Status::NET_ERROR
+                                  ? raw_blob_reader->net_error()
+                                  : net::OK);
+  }
+}
+
+void BundledExchangesBlobDataSource::BlobDataSourceCore::
+    OnBlobReadyForReadToDataPipe(
+        uint64_t offset,
+        uint64_t length,
+        mojo::ScopedDataPipeProducerHandle producer_handle,
+        CompletionCallback callback) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (!blob_) {
+    std::move(callback).Run(net::ERR_FAILED);
+    return;
+  }
+  storage::MojoBlobReader::Create(
+      blob_.get(), net::HttpByteRange::Bounded(offset, offset + length - 1),
+      std::make_unique<MojoBlobReaderDelegate>(std::move(callback)),
+      std::move(producer_handle));
+}
+
+}  // namespace content

content/browser/web_package/bundled_exchanges_blob_data_source.h

+// Copyright 2019 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.
+
+#ifndef CONTENT_BROWSER_WEB_PACKAGE_BUNDLED_EXCHANGES_BLOB_DATA_SOURCE_H_
+#define CONTENT_BROWSER_WEB_PACKAGE_BUNDLED_EXCHANGES_BLOB_DATA_SOURCE_H_
+
+#include <vector>
+
+#include "base/callback_forward.h"
+#include "base/macros.h"
+#include "content/common/content_export.h"
+#include "content/public/browser/browser_context.h"
+#include "mojo/public/cpp/bindings/pending_receiver.h"
+#include "mojo/public/cpp/bindings/receiver.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "net/base/net_errors.h"
+#include "services/data_decoder/public/mojom/bundled_exchanges_parser.mojom.h"
+#include "services/network/public/mojom/url_loader.mojom.h"
+
+namespace storage {
+class BlobBuilderFromStream;
+class BlobDataHandle;
+}  // namespace storage
+
+namespace content {
+
+// This class is used to read partial body data of a web bundle response from
+// server. Currently this class can't read any partial body data before
+// receiving the whole body. TODO(crbug/1018640): Support progressive loading.
+class CONTENT_EXPORT BundledExchangesBlobDataSource {
+ public:
+  using CompletionCallback = base::OnceCallback<void(net::Error net_error)>;
+
+  // This class keeps |endpoints| to keep the ongoing network request.
+  BundledExchangesBlobDataSource(
+      int64_t content_length,
+      mojo::ScopedDataPipeConsumerHandle outer_response_body,
+      network::mojom::URLLoaderClientEndpointsPtr endpoints,
+      BrowserContext::BlobContextGetter blob_context_getter,
+      mojo::PendingReceiver<data_decoder::mojom::BundleDataSource>
+          pending_receiver);
+  ~BundledExchangesBlobDataSource();
+
+  void ReadToDataPipe(uint64_t offset,
+                      uint64_t length,
+                      mojo::ScopedDataPipeProducerHandle producer_handle,
+                      CompletionCallback callback);
+
+ private:
+  // This class lives on the IO thread.
+  class BlobDataSourceCore : public data_decoder::mojom::BundleDataSource {
+   public:
+    BlobDataSourceCore(
+        int64_t content_length,
+        network::mojom::URLLoaderClientEndpointsPtr endpoints,
+        BrowserContext::BlobContextGetter blob_context_getter,
+        mojo::PendingReceiver<data_decoder::mojom::BundleDataSource>
+            pending_receiver);
+    ~BlobDataSourceCore() override;
+
+    void Start(mojo::ScopedDataPipeConsumerHandle outer_response_body);
+
+    void ReadToDataPipe(uint64_t offset,
+                        uint64_t length,
+                        mojo::ScopedDataPipeProducerHandle producer_handle,
+                        CompletionCallback callback);
+
+    base::WeakPtr<BlobDataSourceCore> GetWeakPtr();
+
+   private:
+    // Implements data_decoder::mojom::BundleDataSource.
+    void GetSize(GetSizeCallback callback) override;
+    void Read(uint64_t offset, uint64_t length, ReadCallback callback) override;
+
+    void StreamingBlobDone(storage::BlobBuilderFromStream* builder,
+                           std::unique_ptr<storage::BlobDataHandle> result);
+    void WaitForBlob(base::OnceClosure closure);
+
+    void OnBlobReadyForRead(uint64_t offset,
+                            uint64_t length,
+                            ReadCallback callback);
+    void OnBlobReadyForReadToDataPipe(
+        uint64_t offset,
+        uint64_t length,
+        mojo::ScopedDataPipeProducerHandle producer_handle,
+        CompletionCallback callback);
+
+    const int64_t content_length_;
+    // Used to keep the ongoing network request.
+    network::mojom::URLLoaderClientEndpointsPtr endpoints_;
+    mojo::Receiver<data_decoder::mojom::BundleDataSource> receiver_;
+    std::unique_ptr<storage::BlobBuilderFromStream> blob_builder_from_stream_;
+
+    std::unique_ptr<storage::BlobDataHandle> blob_;
+
+    // Used to wait StreamingBlobDone() to be called.
+    std::vector<base::OnceClosure> pending_get_blob_tasks_;
+
+    base::WeakPtrFactory<BlobDataSourceCore> weak_factory_{this};
+
+    DISALLOW_COPY_AND_ASSIGN(BlobDataSourceCore);
+  };
+
+  static void CreateCoreOnIO(
+      base::WeakPtr<BundledExchangesBlobDataSource> weak_ptr,
+      int64_t content_length,
+      mojo::ScopedDataPipeConsumerHandle outer_response_body,
+      network::mojom::URLLoaderClientEndpointsPtr endpoints,
+      BrowserContext::BlobContextGetter blob_context_getter,
+      mojo::PendingReceiver<data_decoder::mojom::BundleDataSource>
+          pending_receiver);
+
+  static void SetCoreOnUI(
+      base::WeakPtr<BundledExchangesBlobDataSource> weak_ptr,
+      base::WeakPtr<BlobDataSourceCore> weak_core,
+      std::unique_ptr<BlobDataSourceCore> core);
+
+  static void DeleteCore(std::unique_ptr<BlobDataSourceCore> core);
+
+  void WaitForCore(base::OnceClosure callback);
+
+  void ReadToDataPipeImpl(uint64_t offset,
+                          uint64_t length,
+                          mojo::ScopedDataPipeProducerHandle producer_handle,
+                          CompletionCallback callback);
+
+  void SetCoreOnUIImpl(base::WeakPtr<BlobDataSourceCore> weak_core,
+                       std::unique_ptr<BlobDataSourceCore> core);
+
+  // Used to call BlobDataSourceCore's method on the IO thread.
+  base::WeakPtr<BlobDataSourceCore> weak_core_;
+  // Owned by |this|. Must be deleted on the IO thread.
+  std::unique_ptr<BlobDataSourceCore> core_;
+  // Used to wait SetCoreOnUI() to be called.
+  std::vector<base::OnceClosure> pending_get_core_tasks_;
+
+  base::WeakPtrFactory<BundledExchangesBlobDataSource> weak_factory_{this};
+
+  DISALLOW_COPY_AND_ASSIGN(BundledExchangesBlobDataSource);
+};
+
+}  // namespace content
+
+#endif  // CONTENT_BROWSER_WEB_PACKAGE_BUNDLED_EXCHANGES_BLOB_DATA_SOURCE_H_

content/browser/web_package/bundled_exchanges_blob_data_source_unittest.cc

+// Copyright 2019 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 "content/browser/web_package/bundled_exchanges_blob_data_source.h"
+
+#include "base/files/scoped_temp_dir.h"
+#include "base/run_loop.h"
+#include "base/task/post_task.h"
+#include "content/public/test/browser_task_environment.h"
+#include "mojo/public/cpp/system/data_pipe_utils.h"
+#include "storage/browser/blob/blob_storage_context.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace content {
+
+namespace {
+
+constexpr size_t kTestBlobStorageMaxBytesDataItemSize = 13;
+constexpr size_t kTestBlobStorageMaxBlobMemorySize = 500;
+constexpr uint64_t kTestBlobStorageMinFileSizeBytes = 32;
+constexpr uint64_t kTestBlobStorageMaxFileSizeBytes = 100;
+constexpr uint64_t kTestBlobStorageMaxDiskSpace = 1000;
+
+}  // namespace
+
+class BundledExchangesBlobDataSourceTest : public testing::Test {
+ protected:
+  BundledExchangesBlobDataSourceTest() = default;
+  ~BundledExchangesBlobDataSourceTest() override = default;
+  void SetUp() override {
+    ASSERT_TRUE(data_dir_.CreateUniqueTempDir());
+    context_ = std::make_unique<storage::BlobStorageContext>(
+        data_dir_.GetPath(),
+        base::CreateTaskRunner({base::ThreadPool(), base::MayBlock()}));
+    storage::BlobStorageLimits limits;
+    limits.max_ipc_memory_size = kTestBlobStorageMaxBytesDataItemSize;
+    limits.max_shared_memory_size = kTestBlobStorageMaxBytesDataItemSize;
+    limits.max_bytes_data_item_size = kTestBlobStorageMaxBytesDataItemSize;
+    limits.max_blob_in_memory_space = kTestBlobStorageMaxBlobMemorySize;
+    limits.min_page_file_size = kTestBlobStorageMinFileSizeBytes;
+    limits.max_file_size = kTestBlobStorageMaxFileSizeBytes;
+    limits.desired_max_disk_space = kTestBlobStorageMaxDiskSpace;
+    limits.effective_max_disk_space = kTestBlobStorageMaxDiskSpace;
+    context_->set_limits_for_testing(limits);
+  }
+
+  std::unique_ptr<BundledExchangesBlobDataSource> CreateTestDataSource(
+      const std::string& test_data,
+      mojo::Remote<data_decoder::mojom::BundleDataSource>* remote_source,
+      base::Optional<int64_t> content_length = base::nullopt) {
+    mojo::ScopedDataPipeProducerHandle producer;
+    mojo::ScopedDataPipeConsumerHandle consumer;
+    CHECK_EQ(MOJO_RESULT_OK,
+             mojo::CreateDataPipe(nullptr, &producer, &consumer));
+    mojo::BlockingCopyFromString(test_data, producer);
+    producer.reset();
+
+    return std::make_unique<BundledExchangesBlobDataSource>(
+        content_length ? *content_length : test_data.size(),
+        std::move(consumer), nullptr, ContextGetter(),
+        remote_source->BindNewPipeAndPassReceiver());
+  }
+  std::unique_ptr<storage::BlobStorageContext> context_;
+
+ private:
+  BrowserContext::BlobContextGetter ContextGetter() {
+    return base::BindRepeating(
+        [](base::WeakPtr<storage::BlobStorageContext> weak_context) {
+          return weak_context;
+        },
+        context_->AsWeakPtr());
+  }
+
+  base::ScopedTempDir data_dir_;
+  BrowserTaskEnvironment task_environment_;
+  DISALLOW_COPY_AND_ASSIGN(BundledExchangesBlobDataSourceTest);
+};
+
+TEST_F(BundledExchangesBlobDataSourceTest, GetSize) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source);
+
+  base::RunLoop run_loop;
+  uint64_t size_result = 0;
+  remote_source->GetSize(base::BindOnce(
+      [](base::OnceClosure closure, uint64_t* size_result, uint64_t size) {
+        *size_result = size;
+        std::move(closure).Run();
+      },
+      run_loop.QuitClosure(), &size_result));
+  run_loop.Run();
+  EXPECT_EQ(kData.size(), size_result);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, Read) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+  remote_source->Read(
+      1, 3,
+      base::BindOnce(
+          [](base::OnceClosure closure,
+             base::Optional<std::vector<uint8_t>>* read_result,
+             const base::Optional<std::vector<uint8_t>>& result) {
+            *read_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_result));
+  run_loop.Run();
+  ASSERT_TRUE(read_result);
+  ASSERT_EQ(3u, read_result->size());
+  EXPECT_EQ('e', (*read_result)[0]);
+  EXPECT_EQ('s', (*read_result)[1]);
+  EXPECT_EQ('t', (*read_result)[2]);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, Read_OutOfRangeError) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+  remote_source->Read(
+      1, 100,
+      base::BindOnce(
+          [](base::OnceClosure closure,
+             base::Optional<std::vector<uint8_t>>* read_result,
+             const base::Optional<std::vector<uint8_t>>& result) {
+            *read_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_result));
+  run_loop.Run();
+  EXPECT_FALSE(read_result);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, Read_InvalidContentLengthError) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source, kData.size() + 1);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+  remote_source->Read(
+      1, 3,
+      base::BindOnce(
+          [](base::OnceClosure closure,
+             base::Optional<std::vector<uint8_t>>* read_result,
+             const base::Optional<std::vector<uint8_t>>& result) {
+            *read_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_result));
+  run_loop.Run();
+  EXPECT_FALSE(read_result);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, Read_NoStorage) {
+  std::string content = "Test Data";
+  // Make the content larger than the disk space.
+  content.resize(kTestBlobStorageMaxDiskSpace + 1, ' ');
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(content, &remote_source);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+  remote_source->Read(
+      1, 100,
+      base::BindOnce(
+          [](base::OnceClosure closure,
+             base::Optional<std::vector<uint8_t>>* read_result,
+             const base::Optional<std::vector<uint8_t>>& result) {
+            *read_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_result));
+  run_loop.Run();
+  EXPECT_FALSE(read_result);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, ReadToDataPipe) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+
+  mojo::ScopedDataPipeProducerHandle producer;
+  mojo::ScopedDataPipeConsumerHandle consumer;
+  CHECK_EQ(MOJO_RESULT_OK, mojo::CreateDataPipe(nullptr, &producer, &consumer));
+
+  net::Error read_response_body_result = net::ERR_FAILED;
+  source->ReadToDataPipe(
+      1, 3, std::move(producer),
+      base::BindOnce(
+          [](base::OnceClosure closure, net::Error* read_response_body_result,
+             net::Error result) {
+            *read_response_body_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_response_body_result));
+  run_loop.Run();
+  EXPECT_EQ(net::OK, read_response_body_result);
+
+  std::string result_string;
+  mojo::BlockingCopyToString(std::move(consumer), &result_string);
+  EXPECT_EQ(3u, result_string.size());
+  EXPECT_EQ(kData.substr(1, 3), result_string);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, ReadToDataPipe_OutOfRangeError) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+
+  mojo::ScopedDataPipeProducerHandle producer;
+  mojo::ScopedDataPipeConsumerHandle consumer;
+  CHECK_EQ(MOJO_RESULT_OK, mojo::CreateDataPipe(nullptr, &producer, &consumer));
+
+  net::Error read_response_body_result = net::ERR_FAILED;
+  source->ReadToDataPipe(
+      1, 100, std::move(producer),
+      base::BindOnce(
+          [](base::OnceClosure closure, net::Error* read_response_body_result,
+             net::Error result) {
+            *read_response_body_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_response_body_result));
+  run_loop.Run();
+  EXPECT_EQ(net::ERR_REQUEST_RANGE_NOT_SATISFIABLE, read_response_body_result);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest,
+       ReadToDataPipe_InvalidContentLengthError) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source, kData.size() + 1);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+
+  mojo::ScopedDataPipeProducerHandle producer;
+  mojo::ScopedDataPipeConsumerHandle consumer;
+  CHECK_EQ(MOJO_RESULT_OK, mojo::CreateDataPipe(nullptr, &producer, &consumer));
+
+  net::Error read_response_body_result = net::OK;
+  source->ReadToDataPipe(
+      1, 3, std::move(producer),
+      base::BindOnce(
+          [](base::OnceClosure closure, net::Error* read_response_body_result,
+             net::Error result) {
+            *read_response_body_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_response_body_result));
+  run_loop.Run();
+  EXPECT_EQ(net::ERR_FAILED, read_response_body_result);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, ReadToDataPipe_NoStorage) {
+  std::string content = "Test Data";
+  // Make the content larger than the disk space.
+  content.resize(kTestBlobStorageMaxDiskSpace + 1, ' ');
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(content, &remote_source);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+
+  mojo::ScopedDataPipeProducerHandle producer;
+  mojo::ScopedDataPipeConsumerHandle consumer;
+  CHECK_EQ(MOJO_RESULT_OK, mojo::CreateDataPipe(nullptr, &producer, &consumer));
+
+  net::Error read_response_body_result = net::OK;
+  source->ReadToDataPipe(
+      1, 3, std::move(producer),
+      base::BindOnce(
+          [](base::OnceClosure closure, net::Error* read_response_body_result,
+             net::Error result) {
+            *read_response_body_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_response_body_result));
+  run_loop.Run();
+  EXPECT_EQ(net::ERR_FAILED, read_response_body_result);
+}
+
+TEST_F(BundledExchangesBlobDataSourceTest, ReadToDataPipe_Destructed) {
+  const std::string kData = "Test Data";
+  mojo::Remote<data_decoder::mojom::BundleDataSource> remote_source;
+  auto source = CreateTestDataSource(kData, &remote_source);
+
+  base::RunLoop run_loop;
+  base::Optional<std::vector<uint8_t>> read_result;
+
+  mojo::ScopedDataPipeProducerHandle producer;
+  mojo::ScopedDataPipeConsumerHandle consumer;
+  CHECK_EQ(MOJO_RESULT_OK, mojo::CreateDataPipe(nullptr, &producer, &consumer));
+
+  net::Error read_response_body_result = net::OK;
+  source->ReadToDataPipe(
+      1, 3, std::move(producer),
+      base::BindOnce(
+          [](base::OnceClosure closure, net::Error* read_response_body_result,
+             net::Error result) {
+            *read_response_body_result = result;
+            std::move(closure).Run();
+          },
+          run_loop.QuitClosure(), &read_response_body_result));
+  source.reset();
+  run_loop.Run();
+  EXPECT_EQ(net::ERR_FAILED, read_response_body_result);
+}
+
+}  // namespace content

content/test/BUILD.gn

@@ -1822,6 +1822,7 @@ test("content_unittests") {
     "../browser/web_contents/web_contents_view_mac_unittest.mm",
     "../browser/web_contents/web_drag_dest_mac_unittest.mm",
     "../browser/web_contents/web_drag_source_mac_unittest.mm",
+    "../browser/web_package/bundled_exchanges_blob_data_source_unittest.cc",
     "../browser/web_package/bundled_exchanges_reader_unittest.cc",
     "../browser/web_package/bundled_exchanges_url_loader_factory_unittest.cc",
     "../browser/web_package/bundled_exchanges_utils_unittest.cc",