[WIP] Mojofy respondwith

content/browser/service_worker/service_worker_url_request_job.cc

 // Copyright 2014 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/service_worker/service_worker_url_request_job.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <map>
@@ skipping 23 matching lines @@
 #include "content/browser/service_worker/service_worker_response_info.h"
 #include "content/browser/service_worker/service_worker_stream_reader.h"
 #include "content/common/resource_request_body_impl.h"
 #include "content/common/service_worker/service_worker_types.h"
 #include "content/common/service_worker/service_worker_utils.h"
 #include "content/public/browser/blob_handle.h"
 #include "content/public/browser/browser_thread.h"
 #include "content/public/browser/resource_request_info.h"
 #include "content/public/browser/service_worker_context.h"
 #include "content/public/common/referrer.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
 #include "net/base/net_errors.h"
 #include "net/http/http_request_headers.h"
 #include "net/http/http_response_headers.h"
 #include "net/http/http_response_info.h"
 #include "net/http/http_util.h"
 #include "net/log/net_log.h"
 #include "net/log/net_log_event_type.h"
 #include "net/log/net_log_with_source.h"
 #include "storage/browser/blob/blob_data_builder.h"
 #include "storage/browser/blob/blob_data_handle.h"
@@ skipping 65 matching lines @@
     base::File::Info file_info;
     if (!base::GetFileInfo(path, &file_info) || file_info.is_directory)
       return std::vector<int64_t>();
     sizes.push_back(file_info.size);
   }
   return sizes;
 }
 
 }  // namespace
 
+class ServiceWorkerDataPipeReader {
+ public:
+  ServiceWorkerDataPipeReader(
+      ServiceWorkerURLRequestJob* owner,
+      scoped_refptr<ServiceWorkerVersion> streaming_version,
+      mojo::ScopedDataPipeConsumerHandle stream)
+      : owner_(owner),
+        streaming_version_(streaming_version),
+        handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+        stream_(std::move(stream)) {
+    TRACE_EVENT_ASYNC_BEGIN1("ServiceWorker", "ServiceWorkerDataPipeReader",
+                             this, "Url", owner->request()->url().spec());
+    streaming_version_->AddStreamingURLRequestJob(owner_);
+  }
+  ~ServiceWorkerDataPipeReader() {
+    // LOG(ERROR) << "~ServiceWorkerDataPipeReader";
+    if (streaming_version_) {
+      streaming_version_->RemoveStreamingURLRequestJob(owner_);
+      streaming_version_ = nullptr;
+    }
+    TRACE_EVENT_ASYNC_END0("ServiceWorker", "ServiceWorkerDataPipeReader",
+                           this);
+  }
+  void Start() {
+    TRACE_EVENT_ASYNC_STEP_INTO0("ServiceWorker", "ServiceWorkerDataPipeReader",
+                                 this, "Start");
+    // LOG(ERROR) << "ServiceWorkerDataPipeReader::Start";
+    handle_watcher_.Watch(
+        stream_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+        base::Bind(&ServiceWorkerDataPipeReader::OnHandleGotReadable,
+                   base::Unretained(this)));
+    handle_watcher_.ArmOrNotify();
+    owner_->OnResponseStarted();
+  }
+  void OnHandleGotReadable(MojoResult) {
+    TRACE_EVENT_ASYNC_STEP_INTO0("ServiceWorker", "ServiceWorkerDataPipeReader",
+                                 this, "OnHandleGotReadable");
+    // LOG(ERROR) << "OnHandleGotReadable";
+    if (!stream_pending_buffer_)
+      return;
+    uint32_t size_to_pass = stream_pending_buffer_size_;
+    MojoResult mojo_result =
+        mojo::ReadDataRaw(stream_.get(), stream_pending_buffer_->data(),
+                          &size_to_pass, MOJO_READ_DATA_FLAG_NONE);
+    if (mojo_result == MOJO_RESULT_SHOULD_WAIT) {
+      handle_watcher_.ArmOrNotify();
+      return;
+    }
+    stream_pending_buffer_ = nullptr;
+    stream_pending_buffer_size_ = 0;
+
+    switch (mojo_result) {
+      case MOJO_RESULT_OK:
+        owner_->OnReadRawDataComplete(size_to_pass);
+        return;
+      case MOJO_RESULT_FAILED_PRECONDITION:
+        handle_watcher_.Cancel();
+        owner_->RecordResult(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE);
+        owner_->OnReadRawDataComplete(0);
+        return;
+      case MOJO_RESULT_BUSY:
+      case MOJO_RESULT_RESOURCE_EXHAUSTED:
+        owner_->OnReadRawDataComplete(net::ERR_FAILED);
+        return;
+    }
+  }
+  int ReadRawData(net::IOBuffer* buf, int buf_size) {
+    TRACE_EVENT_ASYNC_STEP_INTO0("ServiceWorker", "ServiceWorkerDataPipeReader",
+                                 this, "ReadRawData");
+    uint32_t size_to_pass = buf_size;
+    MojoResult mojo_result = mojo::ReadDataRaw(
+        stream_.get(), buf->data(), &size_to_pass, MOJO_READ_DATA_FLAG_NONE);
+    switch (mojo_result) {
+      case MOJO_RESULT_OK:
+        return size_to_pass;
+      case MOJO_RESULT_FAILED_PRECONDITION:
+        handle_watcher_.Cancel();
+        owner_->RecordResult(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE);
+        return 0;
+      case MOJO_RESULT_BUSY:
+        return net::ERR_FAILED;
+      case MOJO_RESULT_SHOULD_WAIT:
+        stream_pending_buffer_ = buf;
+        stream_pending_buffer_size_ = buf_size;
+        handle_watcher_.ArmOrNotify();
+        return net::ERR_IO_PENDING;
+      case MOJO_RESULT_RESOURCE_EXHAUSTED:
+        return net::ERR_FAILED;
+    }
+    return net::ERR_FAILED;
+  }
+
+ private:
+  ServiceWorkerURLRequestJob* owner_;
+  scoped_refptr<ServiceWorkerVersion> streaming_version_;
+  scoped_refptr<net::IOBuffer> stream_pending_buffer_;
+  int stream_pending_buffer_size_ = 0;
+  mojo::SimpleWatcher handle_watcher_;
+  mojo::ScopedDataPipeConsumerHandle stream_;
+};
+
 // Sets the size on each DataElement in the request body that is a file with
 // unknown size. This ensures ServiceWorkerURLRequestJob::CreateRequestBodyBlob
 // can successfuly create a blob from the data elements, as files with unknown
 // sizes are not supported by the blob storage system.
 class ServiceWorkerURLRequestJob::FileSizeResolver {
  public:
   explicit FileSizeResolver(ServiceWorkerURLRequestJob* owner)
       : owner_(owner), weak_factory_(this) {
     TRACE_EVENT_ASYNC_BEGIN1("ServiceWorker", "FileSizeResolver", this, "URL",
                              owner_->request()->url().spec());
@@ skipping 115 matching lines @@
       fall_back_required_(false),
       body_(body),
       fetch_type_(fetch_type),
       timeout_(timeout),
       weak_factory_(this) {
   DCHECK(delegate_) << "ServiceWorkerURLRequestJob requires a delegate";
 }
 
 ServiceWorkerURLRequestJob::~ServiceWorkerURLRequestJob() {
   stream_reader_.reset();
+  data_pipe_reader_.reset();
   file_size_resolver_.reset();
 
   if (!ShouldRecordResult())
     return;
   ServiceWorkerMetrics::URLRequestJobResult result =
       ServiceWorkerMetrics::REQUEST_JOB_ERROR_KILLED;
   if (response_body_type_ == STREAM)
     result = ServiceWorkerMetrics::REQUEST_JOB_ERROR_KILLED_WITH_STREAM;
   else if (response_body_type_ == BLOB)
     result = ServiceWorkerMetrics::REQUEST_JOB_ERROR_KILLED_WITH_BLOB;
@@ skipping 31 matching lines @@
 }
 
 void ServiceWorkerURLRequestJob::Start() {
   is_started_ = true;
   MaybeStartRequest();
 }
 
 void ServiceWorkerURLRequestJob::Kill() {
   net::URLRequestJob::Kill();
   stream_reader_.reset();
+  data_pipe_reader_.reset();
   fetch_dispatcher_.reset();
   blob_reader_.reset();
   weak_factory_.InvalidateWeakPtrs();
 }
 
 net::LoadState ServiceWorkerURLRequestJob::GetLoadState() const {
   // TODO(kinuko): refine this for better debug.
   return net::URLRequestJob::GetLoadState();
 }
 
 bool ServiceWorkerURLRequestJob::GetCharset(std::string* charset) {
   if (!http_info())
     return false;
+  // LOG(ERROR) << "http_info()->headers " << http_info()->headers;
   return http_info()->headers->GetCharset(charset);
 }
 
 bool ServiceWorkerURLRequestJob::GetMimeType(std::string* mime_type) const {
   if (!http_info())
     return false;
   return http_info()->headers->GetMimeType(mime_type);
 }
 
 void ServiceWorkerURLRequestJob::GetResponseInfo(net::HttpResponseInfo* info) {
@@ skipping 23 matching lines @@
   if (ranges.size() == 1U)
     byte_range_ = ranges[0];
 }
 
 int ServiceWorkerURLRequestJob::ReadRawData(net::IOBuffer* buf, int buf_size) {
   DCHECK(buf);
   DCHECK_GE(buf_size, 0);
 
   if (stream_reader_)
     return stream_reader_->ReadRawData(buf, buf_size);
+  if (data_pipe_reader_)
+    return data_pipe_reader_->ReadRawData(buf, buf_size);
   if (blob_reader_)
     return blob_reader_->ReadRawData(buf, buf_size);
 
   return 0;
 }
 
 void ServiceWorkerURLRequestJob::OnResponseStarted() {
   if (response_time_.is_null())
     response_time_ = base::Time::Now();
   CommitResponseHeader();
@@ skipping 175 matching lines @@
   ServiceWorkerMetrics::RecordActivatedWorkerPreparationForMainFrame(
       worker_ready_time_ - request()->creation_time(), initial_worker_status_,
       worker_start_situation_, did_navigation_preload_);
   MaybeReportNavigationPreloadMetrics();
 }
 
 void ServiceWorkerURLRequestJob::DidDispatchFetchEvent(
     ServiceWorkerStatusCode status,
     ServiceWorkerFetchEventResult fetch_result,
     const ServiceWorkerResponse& response,
+    mojo::ScopedDataPipeConsumerHandle stream,
     const scoped_refptr<ServiceWorkerVersion>& version) {
   // Do not clear |fetch_dispatcher_| if it has dispatched a navigation preload
   // request to keep the mojom::URLLoader related objects in it, because the
   // preload response might still need to be streamed even after calling
   // respondWith().
   if (!did_navigation_preload_) {
     fetch_dispatcher_.reset();
   }
   ServiceWorkerMetrics::RecordFetchEventStatus(IsMainResourceLoad(), status);
 
@@ skipping 55 matching lines @@
   // Set up a request for reading the stream.
   if (response.stream_url.is_valid()) {
     DCHECK(response.blob_uuid.empty());
     SetResponseBodyType(STREAM);
     SetResponse(response);
     stream_reader_.reset(new ServiceWorkerStreamReader(this, version));
     stream_reader_->Start(response.stream_url);
     return;
   }
 
+  if (stream.is_valid()) {
+    // LOG(ERROR) << "response.status_code " << response.status_code;
+    // LOG(ERROR) << "stream.is_valid()";
+    SetResponseBodyType(STREAM);
+    SetResponse(response);
+    data_pipe_reader_.reset(
+        new ServiceWorkerDataPipeReader(this, version, std::move(stream)));
+    data_pipe_reader_->Start();
+    return;
+  }
+
   // Set up a request for reading the blob.
   if (!response.blob_uuid.empty() && blob_storage_context_) {
     SetResponseBodyType(BLOB);
     std::unique_ptr<storage::BlobDataHandle> blob_data_handle =
         blob_storage_context_->GetBlobDataFromUUID(response.blob_uuid);
     if (!blob_data_handle) {
       // The renderer gave us a bad blob UUID.
       RecordResult(ServiceWorkerMetrics::REQUEST_JOB_ERROR_BAD_BLOB);
       DeliverErrorResponse();
       return;
@@ skipping 257 matching lines @@
   DCHECK(!reported_navigation_preload_metrics_);
   reported_navigation_preload_metrics_ = true;
 
   ServiceWorkerMetrics::RecordNavigationPreloadResponse(
       worker_ready_time_ - worker_start_time_,
       navigation_preload_response_time_ - worker_start_time_,
       initial_worker_status_, worker_start_situation_);
 }
 
 }  // namespace content