[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..a61328204f89da630c2ddb1781b9a3842fb5b3d3 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,10 @@ 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),
+      stream_total_bytes_read_(0),
+      stream_max_range_(0),
       request_mode_(request_mode),
       credentials_mode_(credentials_mode),
       request_context_type_(request_context_type),
@@ -76,6 +85,17 @@ void ServiceWorkerURLRequestJob::Start() {
 
 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();
@@ -132,6 +152,39 @@ void ServiceWorkerURLRequestJob::SetExtraRequestHeaders(
 
 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;
@@ -205,6 +258,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 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;
@@ -438,6 +542,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 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 =