[ServiceWorker] Make Stream support in ServiceWorkerURLRequestJob

content/browser/service_worker/service_worker_url_request_job.cc

Index: content/browser/service_worker/service_worker_url_request_job.cc
diff --git a/content/browser/service_worker/service_worker_url_request_job.cc b/content/browser/service_worker/service_worker_url_request_job.cc
index fcf377bb17ab4f8650f5d0a9552fc6dff6fa23d5..c7a98d9b6b6ad79611bf01b0ce9571a096e6182f 100644
--- a/content/browser/service_worker/service_worker_url_request_job.cc
+++ b/content/browser/service_worker/service_worker_url_request_job.cc
@@ -13,8 +13,12 @@
 #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"
@@ -37,6 +41,7 @@ ServiceWorkerURLRequestJob::ServiceWorkerURLRequestJob(
     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,
@@ -48,6 +53,8 @@ ServiceWorkerURLRequestJob::ServiceWorkerURLRequestJob(
       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),
@@ -76,6 +83,17 @@ void ServiceWorkerURLRequestJob::Start() {
 
 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();
@@ -132,11 +150,37 @@ void ServiceWorkerURLRequestJob::SetExtraRequestHeaders(
 
 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);
@@ -205,9 +249,57 @@ void ServiceWorkerURLRequestJob::OnReadCompleted(net::URLRequest* request,
     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();
@@ -260,7 +352,7 @@ void ServiceWorkerURLRequestJob::StartRequest() {
 
     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;
@@ -438,6 +530,29 @@ void ServiceWorkerURLRequestJob::DidDispatchFetchEvent(
   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 =