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.
+class BlobURLLoader : public mojom::URLLoader {

[dmurph, 2017/05/25 21:37:10]

Can you write threading expectations? I think we're constructed on the UI
thread, and then operated on by the IO thread.

[Marijn Kruisselbrink, 2017/05/25 22:12:22]

I'm pretty sure it's created on and fully runs on the IO thread (see the
DCHECK_CURRENTLY_ON in CreateLoaderAndStart)

[jam, 2017/05/26 04:46:47]

Correct, I added a dcheck as well.

+ 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) {
+    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();

[Marijn Kruisselbrink, 2017/05/25 22:12:22]

nit: BrowserThread::FILE is deprecated, and you probably should be using
base::CreateSequencedTaskRunnerWithTraits({base::MayBlock(),
base::TaskPriority::BACKGROUND}) or something similar. Although on the other
hand I can see how having minimal changes with the existing code (which is using
BrowserThread::FILE) might be nice too, especially since I'm not sure which
tasks BlobReader actually expects to be sequenced with one another. So yeah, I
guess just using the same sequenced task runner as the old code for now is
definitely the safest thing to do.

[jam, 2017/05/26 04:46:47]

Right, I wanted the same behavior. They can both be changed at the same time.

+    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);

[dmurph, 2017/05/25 21:37:10]

This part seems to make sense to me - I don't know mojo setup well though.

[jam, 2017/05/26 04:46:47]

This is pretty much copied from content/common/url_loader_impl.cc. What I was
hinting at in the cl message is that I'll combine these into a helper class
after this lands. It's too much boilerplate.

+    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.
+      // TODO: How should this be communicated to our client?

[dmurph, 2017/05/25 21:37:10]

NotifyComplete(error)? Or are you unsure what the error code would be?

Can you put UMAs on all of the error possibilities here? I'm thinking
specifically about mojo-related errors (not http request errors). But that could
be in a follow up cl, and I could do that.

[jam, 2017/05/26 04:46:46]

Done
Do we really need it on all codepaths? i.e. the URLRequestJob doesn't do this,
and in general URLRequestJob implementations don't.

+      writable_handle_watcher_.Cancel();
+      response_body_stream_.reset();
+      DeleteIfNeeded();
+      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;
+    // url_request_->Read(buf.get(), static_cast<int>(num_bytes), &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());
+        DeleteIfNeeded();

[dmurph, 2017/05/25 21:37:10]

Redundant line - NotifyCompleted calls this. Or, maybe you need to destroy the
handles and pending write?

[jam, 2017/05/26 04:46:46]

Done.

+        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 {
+          NotifyCompleted(net::OK);
+          writable_handle_watcher_.Cancel();
+          pending_write_->Complete(0);
+          pending_write_ = nullptr;  // This closes the data pipe.
+          DeleteIfNeeded();
+          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_));

[Marijn Kruisselbrink, 2017/05/25 22:12:22]

nit: mojo::ScopedHandleBase is not documented as clearing itself on being moved
from, so even if it might currently always do so it is possibly not safe to rely
on that? Of course it does seem reasonable for ScopedHandleBase to behave
similar to std::unique_ptr in this case, so maybe the correct fix is to just
document that behavior?

[jam, 2017/05/26 04:46:47]

The header says
"// Scoper for the actual handle types defined further below. It's move-only,
// like the C++11 |unique_ptr|."

https://cs.chromium.org/chromium/src/mojo/public/cpp/system/handle.h?rcl=8a5e...

is that what you're looking for?

+    }
+    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.

[dmurph, 2017/05/25 21:37:10]

I'm imagining this would:
1. Close the stream
2. CallNotifyComplete(error)

[jam, 2017/05/26 04:46:47]

Yep (this TODO is copied from the other implementation). Do you want me to
address the TODOs now or is later ok?

+    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);

[dmurph, 2017/05/25 21:37:10]

I double checked that the logic/contents for this class matches
blob_url_request_job.

Are you able to take the test for blob_url_request_job and apply it to this new
handler? Those are pretty fully-featured tests, and with those I would be really
confident that the code here is correct.

[jam, 2017/05/26 04:46:46]

In general, I prefer higher level tests so that they don't have to change when
the implementation does. Are the browser tests and layout tests not sufficient?
If not, I'd prefer to find ways of adding tests there, so that they can run on
both implementations.

+};
+
+}  // 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