[ServiceWorker] Make Stream support in ServiceWorkerURLRequestJob

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 <map>
 #include <string>
 #include <vector>
 
 #include "base/bind.h"
 #include "base/guid.h"
 #include "base/profiler/scoped_tracker.h"
 #include "base/strings/stringprintf.h"
 #include "base/time/time.h"
+#include "content/browser/resource_context_impl.h"
 #include "content/browser/service_worker/service_worker_fetch_dispatcher.h"
 #include "content/browser/service_worker/service_worker_provider_host.h"
+#include "content/browser/streams/stream.h"
+#include "content/browser/streams/stream_context.h"
+#include "content/browser/streams/stream_registry.h"
 #include "content/common/resource_request_body.h"
 #include "content/common/service_worker/service_worker_types.h"
 #include "content/public/browser/blob_handle.h"
 #include "content/public/browser/resource_request_info.h"
 #include "content/public/browser/service_worker_context.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 "storage/browser/blob/blob_data_handle.h"
 #include "storage/browser/blob/blob_storage_context.h"
 #include "storage/browser/blob/blob_url_request_job_factory.h"
 #include "ui/base/page_transition_types.h"
 
 namespace content {
 
 ServiceWorkerURLRequestJob::ServiceWorkerURLRequestJob(
     net::URLRequest* request,
     net::NetworkDelegate* network_delegate,
     base::WeakPtr<ServiceWorkerProviderHost> provider_host,
     base::WeakPtr<storage::BlobStorageContext> blob_storage_context,
+    const ResourceContext* resource_context,
     FetchRequestMode request_mode,
     FetchCredentialsMode credentials_mode,
     RequestContextType request_context_type,
     RequestContextFrameType frame_type,
     scoped_refptr<ResourceRequestBody> body)
     : net::URLRequestJob(request, network_delegate),
       provider_host_(provider_host),
       response_type_(NOT_DETERMINED),
       is_started_(false),
       service_worker_response_type_(blink::WebServiceWorkerResponseTypeDefault),
       blob_storage_context_(blob_storage_context),
+      resource_context_(resource_context),
+      stream_pending_buffer_size_(0),
       request_mode_(request_mode),
       credentials_mode_(credentials_mode),
       request_context_type_(request_context_type),
       frame_type_(frame_type),
       fall_back_required_(false),
       body_(body),
       weak_factory_(this) {
 }
 
 void ServiceWorkerURLRequestJob::FallbackToNetwork() {
   DCHECK_EQ(NOT_DETERMINED, response_type_);
   response_type_ = FALLBACK_TO_NETWORK;
   MaybeStartRequest();
 }
 
 void ServiceWorkerURLRequestJob::ForwardToServiceWorker() {
   DCHECK_EQ(NOT_DETERMINED, response_type_);
   response_type_ = FORWARD_TO_SERVICE_WORKER;
   MaybeStartRequest();
 }
 
 void ServiceWorkerURLRequestJob::Start() {
   is_started_ = true;
   MaybeStartRequest();
 }
 
 void ServiceWorkerURLRequestJob::Kill() {
   net::URLRequestJob::Kill();
+  if (stream_) {
+    stream_->RemoveReadObserver(this);
+    stream_->Abort();
+    stream_ = nullptr;
+  }
+  if (!waiting_stream_url_.is_empty()) {
+    StreamRegistry* stream_registry =
+        GetStreamContextForResourceContext(resource_context_)->registry();
+    stream_registry->RemoveRegisterObserver(waiting_stream_url_);
+    stream_registry->AbortPendingStream(waiting_stream_url_);
+  }
   fetch_dispatcher_.reset();
   blob_request_.reset();
   weak_factory_.InvalidateWeakPtrs();
 }
 
 net::LoadState ServiceWorkerURLRequestJob::GetLoadState() const {
   // TODO(kinuko): refine this for better debug.
   return net::URLRequestJob::GetLoadState();
 }
 
@@ skipping 36 matching lines @@
     return;
   }
 
   // We don't support multiple range requests in one single URL request.
   if (ranges.size() == 1U)
     byte_range_ = ranges[0];
 }
 
 bool ServiceWorkerURLRequestJob::ReadRawData(
     net::IOBuffer* buf, int buf_size, int *bytes_read) {
+  DCHECK(buf);
+  DCHECK_GE(buf_size, 0);
+  DCHECK(bytes_read);
+  DCHECK(waiting_stream_url_.is_empty());
+  if (stream_.get()) {
+    switch (stream_->ReadRawData(buf, buf_size, bytes_read)) {
+      case Stream::STREAM_HAS_DATA:
+        DCHECK_GT(*bytes_read, 0);
+        return true;
+      case Stream::STREAM_COMPLETE:
+        DCHECK(!*bytes_read);
+        return true;
+      case Stream::STREAM_EMPTY:
+        stream_pending_buffer_ = buf;
+        stream_pending_buffer_size_ = buf_size;
+        SetStatus(net::URLRequestStatus(net::URLRequestStatus::IO_PENDING, 0));
+        return false;
+      case Stream::STREAM_ABORTED:
+        // Handle this as connection reset.
+        NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
+                                         net::ERR_CONNECTION_RESET));
+        return false;
+    }
+    NOTREACHED();
+    return false;
+  }
+
   if (!blob_request_) {
     *bytes_read = 0;
     return true;
   }
-
   blob_request_->Read(buf, buf_size, bytes_read);
   net::URLRequestStatus status = blob_request_->status();
   SetStatus(status);
   if (status.is_io_pending())
     return false;
   return status.is_success();
 }
 
 void ServiceWorkerURLRequestJob::OnReceivedRedirect(
     net::URLRequest* request,
@@ skipping 48 matching lines @@
   SetStatus(request->status());
   if (!request->status().is_success()) {
     NotifyDone(request->status());
     return;
   }
   NotifyReadComplete(bytes_read);
   if (bytes_read == 0)
     NotifyDone(request->status());
 }
 
+void ServiceWorkerURLRequestJob::OnDataAvailable(Stream* stream) {
+  // Clear the IO_PENDING status.
+  SetStatus(net::URLRequestStatus());
+  // Do nothing if stream_pending_buffer_ is empty, i.e. there's no ReadRawData
+  // operation waiting for IO completion.
+  if (!stream_pending_buffer_.get())
+    return;
+
+  // stream_pending_buffer_ is set to the IOBuffer instance provided to
+  // ReadRawData() by URLRequestJob.
+
+  int bytes_read = 0;
+  switch (stream_->ReadRawData(
+      stream_pending_buffer_.get(), stream_pending_buffer_size_, &bytes_read)) {
+    case Stream::STREAM_HAS_DATA:
+      DCHECK_GT(bytes_read, 0);
+      break;
+    case Stream::STREAM_COMPLETE:
+      // Calling NotifyReadComplete with 0 signals completion.
+      DCHECK(!bytes_read);
+      break;
+    case Stream::STREAM_EMPTY:
+      NOTREACHED();
+      break;
+    case Stream::STREAM_ABORTED:
+      // Handle this as connection reset.
+      NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
+                                       net::ERR_CONNECTION_RESET));
+      break;
+  }
+
+  // Clear the buffers before notifying the read is complete, so that it is
+  // safe for the observer to read.
+  stream_pending_buffer_ = nullptr;
+  stream_pending_buffer_size_ = 0;
+  NotifyReadComplete(bytes_read);
+}
+
+void ServiceWorkerURLRequestJob::OnStreamRegistered(Stream* stream) {
+  StreamContext* stream_context =
+      GetStreamContextForResourceContext(resource_context_);
+  stream_context->registry()->RemoveRegisterObserver(waiting_stream_url_);
+  waiting_stream_url_ = GURL();
+  stream_ = stream;
+  stream_->SetReadObserver(this);
+  CommitResponseHeader();
+}
+
 const net::HttpResponseInfo* ServiceWorkerURLRequestJob::http_info() const {
   if (!http_response_info_)
-    return NULL;
+    return nullptr;
   if (range_response_info_)
     return range_response_info_.get();
   return http_response_info_.get();
 }
 
 void ServiceWorkerURLRequestJob::GetExtraResponseInfo(
     bool* was_fetched_via_service_worker,
     bool* was_fallback_required_by_service_worker,
     GURL* original_url_via_service_worker,
     blink::WebServiceWorkerResponseType* response_type_via_service_worker,
@@ skipping 32 matching lines @@
 }
 
 void ServiceWorkerURLRequestJob::StartRequest() {
   switch (response_type_) {
     case NOT_DETERMINED:
       NOTREACHED();
       return;
 
     case FALLBACK_TO_NETWORK:
       // Restart the request to create a new job. Our request handler will
-      // return NULL, and the default job (which will hit network) should be
+      // return nullptr, and the default job (which will hit network) should be
       // created.
       NotifyRestartRequired();
       return;
 
     case FORWARD_TO_SERVICE_WORKER:
       DCHECK(provider_host_ && provider_host_->active_version());
       DCHECK(!fetch_dispatcher_);
       // Send a fetch event to the ServiceWorker associated to the
       // provider_host.
       fetch_dispatcher_.reset(new ServiceWorkerFetchDispatcher(
@@ skipping 157 matching lines @@
   // Treat a response whose status is 0 as a Network Error.
   if (response.status_code == 0) {
     NotifyDone(
         net::URLRequestStatus(net::URLRequestStatus::FAILED, net::ERR_FAILED));
     return;
   }
 
   fetch_end_time_ = base::TimeTicks::Now();
   load_timing_info_.send_end = fetch_end_time_;
 
+  // Set up a request for reading the stream.
+  if (response.stream_url.is_valid()) {
+    DCHECK(response.blob_uuid.empty());
+    response_url_ = response.url;
+    service_worker_response_type_ = response.response_type;
+    CreateResponseHeader(
+        response.status_code, response.status_text, response.headers);
+    load_timing_info_.receive_headers_end = base::TimeTicks::Now();
+    StreamContext* stream_context =
+        GetStreamContextForResourceContext(resource_context_);
+    stream_ =
+        stream_context->registry()->GetStream(response.stream_url);
+    if (!stream_.get()) {
+      waiting_stream_url_ = response.stream_url;
+      // Wait for StreamHostMsg_StartBuilding message from the ServiceWorker.
+      stream_context->registry()->SetRegisterObserver(waiting_stream_url_,
+                                                      this);
+      return;
+    }
+    stream_->SetReadObserver(this);
+    CommitResponseHeader();
+    return;
+  }
   // Set up a request for reading the blob.
   if (!response.blob_uuid.empty() && blob_storage_context_) {
     scoped_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.
       DeliverErrorResponse();
       return;
     }
     blob_request_ = storage::BlobProtocolHandler::CreateBlobRequest(
@@ skipping 40 matching lines @@
 
 void ServiceWorkerURLRequestJob::DeliverErrorResponse() {
   // TODO(falken): Print an error to the console of the ServiceWorker and of
   // the requesting page.
   CreateResponseHeader(
       500, "Service Worker Response Error", ServiceWorkerHeaderMap());
   CommitResponseHeader();
 }
 
 }  // namespace content