Add support for reading blobs when using the network service.

content/browser/blob_storage/blob_url_loader_factory.cc

+// Copyright 2017 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/blob_storage/blob_url_loader_factory.h"
+
+#include <stddef.h>
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/memory/weak_ptr.h"
+#include "base/threading/thread_task_runner_handle.h"
+#include "content/browser/blob_storage/chrome_blob_storage_context.h"
+#include "content/browser/storage_partition_impl.h"
+#include "content/common/net_adapters.h"
+#include "content/common/url_loader.mojom.h"
+#include "content/public/browser/browser_thread.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "net/base/io_buffer.h"
+#include "net/http/http_byte_range.h"
+#include "net/http/http_request_headers.h"
+#include "net/http/http_response_headers.h"
+#include "net/http/http_status_code.h"
+#include "net/http/http_util.h"
+#include "storage/browser/blob/blob_data_handle.h"
+#include "storage/browser/blob/blob_reader.h"
+#include "storage/browser/blob/blob_storage_context.h"
+#include "storage/browser/blob/blob_url_request_job.h"
+#include "storage/browser/fileapi/file_system_context.h"
+
+namespace content {
+
+namespace {
+constexpr size_t kDefaultAllocationSize = 512 * 1024;
+
+// Note: some of this code is duplicated from storage::BlobURLRequestJob.

[kinuko, 2017/05/26 09:48:12]

It feels we could share some more implementation with it..

[jam, 2017/05/26 14:14:53]

There are some scattered lines of code that are shared, but they're closely
tired to URLRequestJob/IOBuffer. I looked into sharing them, but then I'd have
to define an interface to abstract these jobs. So given that this code has
hardly changed in the last year, and once PlzNavigate/MojoLoading ship we can
use this as the one code path, I thought it's better to not add extra
abstractions here that I would have to undo later.

[kinuko, 2017/05/29 04:27:15]

I wondered we might be able to have some static helpers like you did for
headers, but might not be worth doing. (The code around has been stable for
sometime now)

+class BlobURLLoader : public mojom::URLLoader {
+ public:
+  BlobURLLoader(mojom::URLLoaderAssociatedRequest url_loader_request,
+                const ResourceRequest& request,
+                mojom::URLLoaderClientPtr client,
+                storage::BlobStorageContext* blob_storage_context,
+                storage::FileSystemContext* file_system_context)
+      : binding_(this, std::move(url_loader_request)),
+        request_(request),
+        client_(std::move(client)),
+        byte_range_set_(false),
+        writable_handle_watcher_(FROM_HERE,
+                                 mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+        peer_closed_handle_watcher_(FROM_HERE,
+                                    mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+        weak_factory_(this) {
+    DCHECK_CURRENTLY_ON(BrowserThread::IO);
+    blob_handle_ = blob_storage_context->GetBlobDataFromPublicURL(request.url);
+
+    // PostTask since it might destruct.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE,
+        base::Bind(&BlobURLLoader::Start, weak_factory_.GetWeakPtr(), request,
+                   make_scoped_refptr(file_system_context)));
+  }
+
+  void Start(const ResourceRequest& request,
+             scoped_refptr<storage::FileSystemContext> file_system_context) {
+    if (!blob_handle_) {
+      NotifyCompleted(net::ERR_FILE_NOT_FOUND);
+      return;
+    }
+
+    base::SequencedTaskRunner* file_task_runner =
+        BrowserThread::GetTaskRunnerForThread(BrowserThread::FILE).get();
+    blob_reader_ =
+        blob_handle_->CreateReader(file_system_context.get(), file_task_runner);
+
+    // We only support GET request per the spec.
+    if (request.method != "GET") {
+      NotifyCompleted(net::ERR_METHOD_NOT_SUPPORTED);
+      return;
+    }
+
+    if (blob_reader_->net_error()) {
+      NotifyCompleted(blob_reader_->net_error());
+      return;
+    }
+
+    net::HttpRequestHeaders request_headers;
+    request_headers.AddHeadersFromString(request.headers);
+    std::string range_header;
+    if (request_headers.GetHeader(net::HttpRequestHeaders::kRange,
+                                  &range_header)) {
+      // We only care about "Range" header here.
+      std::vector<net::HttpByteRange> ranges;
+      if (net::HttpUtil::ParseRangeHeader(range_header, &ranges)) {
+        if (ranges.size() == 1) {
+          byte_range_set_ = true;
+          byte_range_ = ranges[0];
+        } else {
+          // We don't support multiple range requests in one single URL request,
+          // because we need to do multipart encoding here.
+          // TODO(jianli): Support multipart byte range requests.
+          NotifyCompleted(net::ERR_REQUEST_RANGE_NOT_SATISFIABLE);
+        }
+      }
+    }
+
+    storage::BlobReader::Status size_status =
+        blob_reader_->CalculateSize(base::Bind(&BlobURLLoader::DidCalculateSize,
+                                               weak_factory_.GetWeakPtr()));
+    switch (size_status) {
+      case storage::BlobReader::Status::NET_ERROR:
+        NotifyCompleted(blob_reader_->net_error());
+        return;
+      case storage::BlobReader::Status::IO_PENDING:
+        return;
+      case storage::BlobReader::Status::DONE:
+        DidCalculateSize(net::OK);
+        return;
+    }
+  }
+
+  ~BlobURLLoader() override {}
+
+ private:
+  // mojom::URLLoader implementation:
+  void FollowRedirect() override { NOTREACHED(); }
+
+  void SetPriority(net::RequestPriority priority,
+                   int32_t intra_priority_value) override {
+    NOTREACHED();
+  }
+
+  void NotifyCompleted(int error_code) {
+    ResourceRequestCompletionStatus request_complete_data;
+    request_complete_data.error_code = error_code;
+    client_->OnComplete(request_complete_data);
+
+    DeleteIfNeeded();
+  }
+
+  void DidCalculateSize(int result) {
+    if (result != net::OK) {
+      NotifyCompleted(result);
+      return;
+    }
+
+    // Apply the range requirement.
+    if (!byte_range_.ComputeBounds(blob_reader_->total_size())) {
+      NotifyCompleted(net::ERR_REQUEST_RANGE_NOT_SATISFIABLE);
+      return;
+    }
+
+    DCHECK_LE(byte_range_.first_byte_position(),
+              byte_range_.last_byte_position() + 1);
+    uint64_t length =
+        base::checked_cast<uint64_t>(byte_range_.last_byte_position() -
+                                     byte_range_.first_byte_position() + 1);
+
+    if (byte_range_set_)
+      blob_reader_->SetReadRange(byte_range_.first_byte_position(), length);
+
+    net::HttpStatusCode status_code = net::HTTP_OK;
+    if (byte_range_set_ && byte_range_.IsValid()) {
+      status_code = net::HTTP_PARTIAL_CONTENT;
+    } else {
+      // TODO(horo): When the requester doesn't need the side data
+      // (ex:FileReader) we should skip reading the side data.
+      if (blob_reader_->has_side_data() &&
+          blob_reader_->ReadSideData(base::Bind(&BlobURLLoader::DidReadMetadata,
+                                                weak_factory_.GetWeakPtr())) ==
+              storage::BlobReader::Status::IO_PENDING) {
+        return;
+      }
+    }
+
+    HeadersCompleted(status_code);
+  }
+
+  void DidReadMetadata(storage::BlobReader::Status result) {
+    if (result != storage::BlobReader::Status::DONE) {
+      NotifyCompleted(blob_reader_->net_error());
+      return;
+    }
+    HeadersCompleted(net::HTTP_OK);
+  }
+
+  void HeadersCompleted(net::HttpStatusCode status_code) {
+    ResourceResponseHead response;
+    response.content_length = 0;
+    response.headers = storage::BlobURLRequestJob::GenerateHeaders(
+        status_code, blob_handle_.get(), blob_reader_.get(), &byte_range_,
+        &response.content_length);
+
+    std::string mime_type;
+    response.headers->GetMimeType(&mime_type);
+    // Match logic in StreamURLRequestJob::HeadersCompleted.
+    if (mime_type.empty())
+      mime_type = "text/plain";
+    response.mime_type = mime_type;
+
+    // TODO(jam): some of this code can be shared with
+    // content/network/url_loader_impl.h
+    client_->OnReceiveResponse(response, base::nullopt, nullptr);
+
+    net::IOBufferWithSize* metadata = blob_reader_->side_data();
+    if (metadata) {
+      const uint8_t* data = reinterpret_cast<const uint8_t*>(metadata->data());
+      client_->OnReceiveCachedMetadata(
+          std::vector<uint8_t>(data, data + metadata->size()));
+    }
+
+    mojo::DataPipe data_pipe(kDefaultAllocationSize);
+    response_body_stream_ = std::move(data_pipe.producer_handle);
+    response_body_consumer_handle_ = std::move(data_pipe.consumer_handle);
+    peer_closed_handle_watcher_.Watch(
+        response_body_stream_.get(), MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+        base::Bind(&BlobURLLoader::OnResponseBodyStreamClosed,
+                   base::Unretained(this)));
+    peer_closed_handle_watcher_.ArmOrNotify();
+
+    writable_handle_watcher_.Watch(
+        response_body_stream_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+        base::Bind(&BlobURLLoader::OnResponseBodyStreamReady,
+                   base::Unretained(this)));
+
+    // Start reading...
+    ReadMore();
+  }
+
+  void ReadMore() {
+    DCHECK(!pending_write_.get());
+
+    uint32_t num_bytes;
+    // TODO: we should use the abstractions in MojoAsyncResourceHandler.
+    MojoResult result = NetToMojoPendingBuffer::BeginWrite(
+        &response_body_stream_, &pending_write_, &num_bytes);
+    if (result == MOJO_RESULT_SHOULD_WAIT) {
+      // The pipe is full. We need to wait for it to have more space.
+      writable_handle_watcher_.ArmOrNotify();
+      return;
+    } else if (result != MOJO_RESULT_OK) {
+      // The response body stream is in a bad state. Bail.
+      writable_handle_watcher_.Cancel();
+      response_body_stream_.reset();
+      NotifyCompleted(net::ERR_UNEXPECTED);
+      DeleteIfNeeded();

[kinuko, 2017/05/26 09:48:12]

nit: not needed

[jam, 2017/05/26 14:14:53]

Done.

+      return;
+    }
+
+    CHECK_GT(static_cast<uint32_t>(std::numeric_limits<int>::max()), num_bytes);
+    scoped_refptr<net::IOBuffer> buf(
+        new NetToMojoIOBuffer(pending_write_.get()));
+    int bytes_read;
+    storage::BlobReader::Status read_status = blob_reader_->Read(
+        buf.get(), static_cast<int>(num_bytes), &bytes_read,
+        base::Bind(&BlobURLLoader::DidRead, weak_factory_.GetWeakPtr(), false));
+    switch (read_status) {
+      case storage::BlobReader::Status::NET_ERROR:
+        NotifyCompleted(blob_reader_->net_error());
+        return;
+      case storage::BlobReader::Status::IO_PENDING:
+        // Wait for DidRead.
+        return;
+      case storage::BlobReader::Status::DONE:
+        if (bytes_read > 0) {
+          DidRead(true, bytes_read);
+        } else {
+          writable_handle_watcher_.Cancel();
+          pending_write_->Complete(0);
+          pending_write_ = nullptr;  // This closes the data pipe.
+          NotifyCompleted(net::OK);
+          return;
+        }
+    }
+  }
+
+  void DidRead(bool completed_synchronously, int num_bytes) {
+    if (response_body_consumer_handle_.is_valid()) {
+      // Send the data pipe on the first OnReadCompleted call.
+      client_->OnStartLoadingResponseBody(
+          std::move(response_body_consumer_handle_));
+    }
+    response_body_stream_ = pending_write_->Complete(num_bytes);
+    pending_write_ = nullptr;
+    if (completed_synchronously) {
+      base::ThreadTaskRunnerHandle::Get()->PostTask(
+          FROM_HERE,
+          base::Bind(&BlobURLLoader::ReadMore, weak_factory_.GetWeakPtr()));
+    } else {
+      ReadMore();
+    }
+  }
+
+  void OnResponseBodyStreamClosed(MojoResult result) {
+    response_body_stream_.reset();
+    pending_write_ = nullptr;
+    DeleteIfNeeded();
+  }
+
+  void OnResponseBodyStreamReady(MojoResult result) {
+    // TODO: Handle a bad |result| value.
+    DCHECK_EQ(result, MOJO_RESULT_OK);
+    ReadMore();
+  }
+
+  void DeleteIfNeeded() {
+    bool has_data_pipe =
+        pending_write_.get() || response_body_stream_.is_valid();
+    if (!has_data_pipe)
+      delete this;
+  }
+
+  mojo::AssociatedBinding<mojom::URLLoader> binding_;
+  ResourceRequest request_;
+  mojom::URLLoaderClientPtr client_;
+
+  bool byte_range_set_;
+  net::HttpByteRange byte_range_;
+
+  std::unique_ptr<storage::BlobDataHandle> blob_handle_;
+  std::unique_ptr<storage::BlobReader> blob_reader_;
+
+  // TODO(jam): share with URLLoaderImpl
+  mojo::ScopedDataPipeProducerHandle response_body_stream_;
+  mojo::ScopedDataPipeConsumerHandle response_body_consumer_handle_;
+  scoped_refptr<NetToMojoPendingBuffer> pending_write_;
+  mojo::SimpleWatcher writable_handle_watcher_;
+  mojo::SimpleWatcher peer_closed_handle_watcher_;
+
+  base::WeakPtrFactory<BlobURLLoader> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(BlobURLLoader);
+};
+
+}  // namespace
+
+BlobURLLoaderFactory::BlobURLLoaderFactory(
+    StoragePartitionImpl* storage_partition)
+    : blob_storage_context_(ChromeBlobStorageContext::GetFor(
+          storage_partition->browser_context())),
+      file_system_context_(storage_partition->GetFileSystemContext()) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+}
+
+mojom::URLLoaderFactoryPtr BlobURLLoaderFactory::CreateFactory() {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  mojom::URLLoaderFactoryPtr factory;
+  mojom::URLLoaderFactoryRequest request = mojo::MakeRequest(&factory);
+  BrowserThread::PostTask(BrowserThread::IO, FROM_HERE,
+                          base::BindOnce(&BlobURLLoaderFactory::BindOnIO, this,
+                                         std::move(request)));
+
+  return factory;
+}
+
+BlobURLLoaderFactory::~BlobURLLoaderFactory() {}
+
+void BlobURLLoaderFactory::BindOnIO(mojom::URLLoaderFactoryRequest request) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+
+  loader_factory_bindings_.AddBinding(this, std::move(request));
+}
+
+void BlobURLLoaderFactory::CreateLoaderAndStart(
+    mojom::URLLoaderAssociatedRequest loader,
+    int32_t routing_id,
+    int32_t request_id,
+    uint32_t options,
+    const ResourceRequest& request,
+    mojom::URLLoaderClientPtr client) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  new BlobURLLoader(std::move(loader), request, std::move(client),
+                    blob_storage_context_->context(),
+                    file_system_context_.get());
+}
+
+void BlobURLLoaderFactory::SyncLoad(int32_t routing_id,
+                                    int32_t request_id,
+                                    const ResourceRequest& request,
+                                    SyncLoadCallback callback) {
+  NOTREACHED();
+}
+
+}  // namespace content