[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),
+      stream_total_bytes_read_(0),
+      stream_max_range_(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_.get()) {
+    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->NotifyReaderAbortedBeforeRegistration(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 35 matching lines @@
       !net::HttpUtil::ParseRangeHeader(range_header, &ranges)) {
     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) {

[nhiroki, 2014/11/27 05:03:25]

DCHECK_GE(buf_size, 0)?

[horo, 2014/11/27 06:52:49]

Done.

+  if (stream_.get()) {
+    DCHECK(buf);
+    DCHECK(bytes_read);

[nhiroki, 2014/11/27 05:03:25]

Can you move these checks into the beginning of this function because those are
also used at line 193.

[horo, 2014/11/27 06:52:49]

Done.

+    int to_read = buf_size;
+    if (stream_max_range_ && to_read) {
+      if (to_read + stream_total_bytes_read_ > stream_max_range_)
+        to_read = stream_max_range_ - stream_total_bytes_read_;
+
+      if (to_read <= 0) {

[nhiroki, 2014/11/27 05:03:25]

"to_read < 0" seems a bug because...

1) |buf_size| should be >= 0, and
2) |stream_max_range_| >= |stream_total_bytes_read_|

How about having DCHECK_GE(to_read, 0) and change the condition of the
if-statement like "if (!to_read)"?

[horo, 2014/11/27 06:52:49]

We don't need stream_total_bytes_read_ nor stream_max_range_.
Removed.

+        *bytes_read = 0;
+        return true;
+      }
+    }
+
+    switch (stream_->ReadRawData(buf, to_read, bytes_read)) {
+      case Stream::STREAM_HAS_DATA:

[nhiroki, 2014/11/27 05:03:25]

"DCHECK_GT(bytes_read, 0)" like line 276?

[horo, 2014/11/27 06:52:49]

Done.

+      case Stream::STREAM_COMPLETE:
+        stream_total_bytes_read_ += *bytes_read;
+        return true;
+      case Stream::STREAM_EMPTY:
+        stream_pending_buffer_ = buf;
+        stream_pending_buffer_size_ = to_read;
+        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;
@@ skipping 53 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 pending_buffer_ is empty, i.e. there's no ReadRawData()
+  // operation waiting for IO completion.
+  if (!stream_pending_buffer_.get())
+    return;
+
+  // pending_buffer_ is set to the IOBuffer instance provided to ReadRawData()
+  // by URLRequestJob.
+
+  int bytes_read;
+  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:
+      // Ensure this. Calling NotifyReadComplete call with 0 signals
+      // completion.
+      bytes_read = 0;

[nhiroki, 2014/11/27 05:03:25]

This looks odd... at line 298 |stream_total_bytes_read_| never reaches the
actual total bytes because the last byte size is always 0.

How about using another variable for the signal?

[horo, 2014/11/27 06:52:50]

I think checking bytes_read == 0 is enough.

+      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_ = NULL;

[nhiroki, 2014/11/27 05:03:25]

nullptr?

[horo, 2014/11/27 06:52:49]

Done.

+  stream_pending_buffer_size_ = 0;
+
+  stream_total_bytes_read_ += bytes_read;
+  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;
   if (range_response_info_)
     return range_response_info_.get();
   return http_response_info_.get();
 }
 
 void ServiceWorkerURLRequestJob::GetExtraResponseInfo(
     bool* was_fetched_via_service_worker,
@@ skipping 213 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 ServieWorker.
+      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