ServiceWorker: Use mojo's data pipe for respondWith(stream)

content/browser/service_worker/service_worker_url_request_job_unittest.cc

Index: content/browser/service_worker/service_worker_url_request_job_unittest.cc
diff --git a/content/browser/service_worker/service_worker_url_request_job_unittest.cc b/content/browser/service_worker/service_worker_url_request_job_unittest.cc
index 58e105862eb6b330e7ec7a7eeeb99e3421b2e80b..98961298598dc8fcb89ac02827f35bf0bd28ee21 100644
--- a/content/browser/service_worker/service_worker_url_request_job_unittest.cc
+++ b/content/browser/service_worker/service_worker_url_request_job_unittest.cc
@@ -28,9 +28,6 @@
 #include "content/browser/service_worker/service_worker_response_info.h"
 #include "content/browser/service_worker/service_worker_test_utils.h"
 #include "content/browser/service_worker/service_worker_version.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_impl.h"
 #include "content/common/service_worker/service_worker_messages.h"
 #include "content/common/service_worker/service_worker_status_code.h"
@@ -402,26 +399,26 @@ class ProviderDeleteHelper : public EmbeddedWorkerTestHelper {
   ~ProviderDeleteHelper() override {}
 
  protected:
-  void OnFetchEvent(int embedded_worker_id,
-                    int fetch_event_id,
-                    const ServiceWorkerFetchRequest& request,
-                    mojom::FetchEventPreloadHandlePtr preload_handle,
-                    const FetchCallback& callback) override {
+  void OnFetchEvent(
+      int /* embedded_worker_id */,
+      int /* fetch_event_id */,
+      const ServiceWorkerFetchRequest& /* request */,
+      mojom::FetchEventPreloadHandlePtr /* preload_handle */,
+      mojom::ServiceWorkerFetchResponseCallbackPtr response_callback,
+      const FetchCallback& finish_callback) override {
     context()->RemoveProviderHost(mock_render_process_id(), kProviderID);
-    SimulateSend(new ServiceWorkerHostMsg_FetchEventResponse(
-        embedded_worker_id, fetch_event_id,
-        SERVICE_WORKER_FETCH_EVENT_RESULT_RESPONSE,
+    response_callback->OnResponse(
         ServiceWorkerResponse(
             base::MakeUnique<std::vector<GURL>>(), 200, "OK",
             blink::kWebServiceWorkerResponseTypeDefault,
             base::MakeUnique<ServiceWorkerHeaderMap>(), std::string(), 0,
-            GURL(), blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
+            blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
             false /* response_is_in_cache_storage */,
             std::string() /* response_cache_storage_cache_name */,
             base::MakeUnique<
                 ServiceWorkerHeaderList>() /* cors_exposed_header_names */),
-        base::Time::Now()));
-    callback.Run(SERVICE_WORKER_OK, base::Time::Now());
+        base::Time::Now());
+    finish_callback.Run(SERVICE_WORKER_OK, base::Time::Now());
   }
 
  private:
@@ -488,25 +485,25 @@ class BlobResponder : public EmbeddedWorkerTestHelper {
   ~BlobResponder() override {}
 
  protected:
-  void OnFetchEvent(int embedded_worker_id,
-                    int fetch_event_id,
-                    const ServiceWorkerFetchRequest& request,
-                    mojom::FetchEventPreloadHandlePtr preload_handle,
-                    const FetchCallback& callback) override {
-    SimulateSend(new ServiceWorkerHostMsg_FetchEventResponse(
-        embedded_worker_id, fetch_event_id,
-        SERVICE_WORKER_FETCH_EVENT_RESULT_RESPONSE,
+  void OnFetchEvent(
+      int /* embedded_worker_id */,
+      int /* fetch_event_id */,
+      const ServiceWorkerFetchRequest& /* request */,
+      mojom::FetchEventPreloadHandlePtr /* preload_handle */,
+      mojom::ServiceWorkerFetchResponseCallbackPtr response_callback,
+      const FetchCallback& finish_callback) override {
+    response_callback->OnResponse(
         ServiceWorkerResponse(
             base::MakeUnique<std::vector<GURL>>(), 200, "OK",
             blink::kWebServiceWorkerResponseTypeDefault, MakeHeaders(),
-            blob_uuid_, blob_size_, GURL(),
+            blob_uuid_, blob_size_,
             blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
             false /* response_is_in_cache_storage */,
             std::string() /* response_cache_storage_cache_name */,
             base::MakeUnique<
                 ServiceWorkerHeaderList>() /* cors_exposed_header_names */),
-        base::Time::Now()));
-    callback.Run(SERVICE_WORKER_OK, base::Time::Now());
+        base::Time::Now());
+    finish_callback.Run(SERVICE_WORKER_OK, base::Time::Now());
   }
 
   std::string blob_uuid_;
@@ -569,45 +566,51 @@ TEST_F(ServiceWorkerURLRequestJobTest, NonExistentBlobUUIDResponse) {
 // Responds to fetch events with a stream.
 class StreamResponder : public EmbeddedWorkerTestHelper {
  public:
-  explicit StreamResponder(const GURL& stream_url)
-      : EmbeddedWorkerTestHelper(base::FilePath()), stream_url_(stream_url) {}
+  explicit StreamResponder(
+      blink::mojom::ServiceWorkerStreamCallbackRequest callback_request,
+      mojo::ScopedDataPipeConsumerHandle consumer_handle)
+      : EmbeddedWorkerTestHelper(base::FilePath()) {
+    EXPECT_TRUE(stream_handle_.is_null());
+    stream_handle_ = blink::mojom::ServiceWorkerStreamHandle::New();
+    stream_handle_->callback_request = std::move(callback_request);
+    stream_handle_->stream = std::move(consumer_handle);
+  }
   ~StreamResponder() override {}
 
  protected:
-  void OnFetchEvent(int embedded_worker_id,
-                    int fetch_event_id,
-                    const ServiceWorkerFetchRequest& request,
-                    mojom::FetchEventPreloadHandlePtr preload_handle,
-                    const FetchCallback& callback) override {
-    SimulateSend(new ServiceWorkerHostMsg_FetchEventResponse(
-        embedded_worker_id, fetch_event_id,
-        SERVICE_WORKER_FETCH_EVENT_RESULT_RESPONSE,
+  void OnFetchEvent(
+      int /* embedded_worker_id */,
+      int /* fetch_event_id */,
+      const ServiceWorkerFetchRequest& /* request */,
+      mojom::FetchEventPreloadHandlePtr /* preload_handle */,
+      mojom::ServiceWorkerFetchResponseCallbackPtr response_callback,
+      const FetchCallback& finish_callback) override {
+    ASSERT_FALSE(stream_handle_.is_null());
+    response_callback->OnResponseStream(
         ServiceWorkerResponse(
             base::MakeUnique<std::vector<GURL>>(), 200, "OK",
             blink::kWebServiceWorkerResponseTypeDefault, MakeHeaders(), "", 0,
-            stream_url_, blink::kWebServiceWorkerResponseErrorUnknown,
-            base::Time(), false /* response_is_in_cache_storage */,
+            blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
+            false /* response_is_in_cache_storage */,
             std::string() /* response_cache_storage_cache_name */,
             base::MakeUnique<
                 ServiceWorkerHeaderList>() /* cors_exposed_header_names */),
-        base::Time::Now()));
-    callback.Run(SERVICE_WORKER_OK, base::Time::Now());
+        std::move(stream_handle_), base::Time::Now());
+    finish_callback.Run(SERVICE_WORKER_OK, base::Time::Now());
   }
 
-  const GURL stream_url_;
+  blink::mojom::ServiceWorkerStreamHandlePtr stream_handle_;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(StreamResponder);
 };
 
 TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse) {
-  const GURL stream_url("blob://stream");
-  StreamContext* stream_context =
-      GetStreamContextForResourceContext(
-          browser_context_->GetResourceContext());
-  scoped_refptr<Stream> stream =
-      new Stream(stream_context->registry(), nullptr, stream_url);
-  SetUpWithHelper(new StreamResponder(stream_url));
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  SetUpWithHelper(new StreamResponder(mojo::MakeRequest(&stream_callback),
+                                      std::move(data_pipe.consumer_handle)));
+
   version_->SetStatus(ServiceWorkerVersion::ACTIVATED);
   request_ = url_request_context_.CreateRequest(
       GURL("https://example.com/foo.html"), net::DEFAULT_PRIORITY,
@@ -619,9 +622,15 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse) {
   expected_response.reserve((sizeof(kTestData) - 1) * 1024);
   for (int i = 0; i < 1024; ++i) {
     expected_response += kTestData;
-    stream->AddData(kTestData, sizeof(kTestData) - 1);
+    uint32_t written_bytes = sizeof(kTestData) - 1;
+    MojoResult result =
+        mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                           &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+    EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
   }
-  stream->Finalize(net::OK);
+  stream_callback->OnCompleted();
+  data_pipe.producer_handle.reset();
 
   EXPECT_FALSE(HasWork());
   base::RunLoop().RunUntilIdle();
@@ -649,12 +658,11 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse) {
   EXPECT_FALSE(HasWork());
 }
 
-TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_DelayedRegistration) {
-  const GURL stream_url("blob://stream");
-  StreamContext* stream_context =
-      GetStreamContextForResourceContext(
-          browser_context_->GetResourceContext());
-  SetUpWithHelper(new StreamResponder(stream_url));
+TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_ConsecutiveRead) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  SetUpWithHelper(new StreamResponder(mojo::MakeRequest(&stream_callback),
+                                      std::move(data_pipe.consumer_handle)));
 
   version_->SetStatus(ServiceWorkerVersion::ACTIVATED);
   request_ = url_request_context_.CreateRequest(
@@ -662,20 +670,21 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_DelayedRegistration) {
       &url_request_delegate_);
   request_->set_method("GET");
   request_->Start();
-
-  scoped_refptr<Stream> stream =
-      new Stream(stream_context->registry(), nullptr, stream_url);
   std::string expected_response;
   expected_response.reserve((sizeof(kTestData) - 1) * 1024);
   for (int i = 0; i < 1024; ++i) {
     expected_response += kTestData;
-    stream->AddData(kTestData, sizeof(kTestData) - 1);
+    uint32_t written_bytes = sizeof(kTestData) - 1;
+    MojoResult result =
+        mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                           &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+    EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
+    base::RunLoop().RunUntilIdle();
+    EXPECT_EQ(expected_response, url_request_delegate_.data_received());
   }
-  stream->Finalize(net::OK);
-
-  EXPECT_FALSE(HasWork());
+  stream_callback->OnCompleted();
   base::RunLoop().RunUntilIdle();
-  EXPECT_TRUE(HasWork());
   EXPECT_TRUE(request_->status().is_success());
   EXPECT_EQ(200,
             request_->response_headers()->response_code());
@@ -694,26 +703,67 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_DelayedRegistration) {
             info->response_type_via_service_worker());
   EXPECT_FALSE(info->service_worker_start_time().is_null());
   EXPECT_FALSE(info->service_worker_ready_time().is_null());
+}
 
-  request_.reset();
+TEST_F(ServiceWorkerURLRequestJobTest, StreamResponseAndCancel) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  SetUpWithHelper(new StreamResponder(mojo::MakeRequest(&stream_callback),
+                                      std::move(data_pipe.consumer_handle)));
+
+  version_->SetStatus(ServiceWorkerVersion::ACTIVATED);
+  request_ = url_request_context_.CreateRequest(
+      GURL("https://example.com/foo.html"), net::DEFAULT_PRIORITY,
+      &url_request_delegate_);
+  request_->set_method("GET");
+  request_->Start();
   EXPECT_FALSE(HasWork());
-}
+  base::RunLoop().RunUntilIdle();
+  EXPECT_TRUE(HasWork());
 
-TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_QuickFinalize) {
-  const GURL stream_url("blob://stream");
-  StreamContext* stream_context =
-      GetStreamContextForResourceContext(
-          browser_context_->GetResourceContext());
-  scoped_refptr<Stream> stream =
-      new Stream(stream_context->registry(), nullptr, stream_url);
-  std::string expected_response;
-  expected_response.reserve((sizeof(kTestData) - 1) * 1024);
-  for (int i = 0; i < 1024; ++i) {
-    expected_response += kTestData;
-    stream->AddData(kTestData, sizeof(kTestData) - 1);
+  for (int i = 0; i < 512; ++i) {
+    uint32_t written_bytes = sizeof(kTestData) - 1;
+    MojoResult result =
+        mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                           &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+    EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
   }
-  stream->Finalize(net::OK);
-  SetUpWithHelper(new StreamResponder(stream_url));
+  EXPECT_TRUE(data_pipe.producer_handle.is_valid());
+  request_->Cancel();
+  EXPECT_FALSE(HasWork());
+
+  // Fail to write the data pipe because it's already canceled.
+  uint32_t written_bytes = sizeof(kTestData) - 1;
+  MojoResult result =
+      mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                         &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+  ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, result);
+
+  stream_callback->OnAborted();
+
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(data_pipe.consumer_handle.is_valid());
+  EXPECT_FALSE(request_->status().is_success());
+
+  EXPECT_EQ(0, times_prepare_to_restart_invoked_);
+  ServiceWorkerResponseInfo* info =
+      ServiceWorkerResponseInfo::ForRequest(request_.get());
+  ASSERT_TRUE(info);
+  EXPECT_TRUE(info->was_fetched_via_service_worker());
+  EXPECT_FALSE(info->was_fallback_required());
+  EXPECT_EQ(0u, info->url_list_via_service_worker().size());
+  EXPECT_EQ(blink::kWebServiceWorkerResponseTypeDefault,
+            info->response_type_via_service_worker());
+  EXPECT_FALSE(info->service_worker_start_time().is_null());
+  EXPECT_FALSE(info->service_worker_ready_time().is_null());
+}
+
+TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_Abort) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  SetUpWithHelper(new StreamResponder(mojo::MakeRequest(&stream_callback),
+                                      std::move(data_pipe.consumer_handle)));
 
   version_->SetStatus(ServiceWorkerVersion::ACTIVATED);
   request_ = url_request_context_.CreateRequest(
@@ -721,14 +771,29 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_QuickFinalize) {
       &url_request_delegate_);
   request_->set_method("GET");
   request_->Start();
+
+  std::string expected_response;
+  expected_response.reserve((sizeof(kTestData) - 1) * 1024);
+  for (int i = 0; i < 1024; ++i) {
+    expected_response += kTestData;
+    uint32_t written_bytes = sizeof(kTestData) - 1;
+    MojoResult result =
+        mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                           &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+    EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
+  }
+  stream_callback->OnAborted();
+  data_pipe.producer_handle.reset();
+
   EXPECT_FALSE(HasWork());
   base::RunLoop().RunUntilIdle();
   EXPECT_TRUE(HasWork());
-  EXPECT_TRUE(request_->status().is_success());
-  EXPECT_EQ(200,
-            request_->response_headers()->response_code());
-  EXPECT_EQ("OK",
-            request_->response_headers()->GetStatusText());
+  EXPECT_FALSE(request_->status().is_success());
+  net::HttpResponseHeaders* headers = request_->response_headers();
+  EXPECT_EQ(200, headers->response_code());
+  EXPECT_EQ("OK", headers->GetStatusText());
+  CheckHeaders(headers);
   EXPECT_EQ(expected_response, url_request_delegate_.data_received());
 
   EXPECT_EQ(0, times_prepare_to_restart_invoked_);
@@ -747,33 +812,45 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_QuickFinalize) {
   EXPECT_FALSE(HasWork());
 }
 
-TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_Flush) {
-  const GURL stream_url("blob://stream");
-  StreamContext* stream_context =
-      GetStreamContextForResourceContext(
-          browser_context_->GetResourceContext());
-  scoped_refptr<Stream> stream =
-      new Stream(stream_context->registry(), nullptr, stream_url);
-  SetUpWithHelper(new StreamResponder(stream_url));
+TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_AbortBeforeData) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  SetUpWithHelper(new StreamResponder(mojo::MakeRequest(&stream_callback),
+                                      std::move(data_pipe.consumer_handle)));
 
   version_->SetStatus(ServiceWorkerVersion::ACTIVATED);
+
   request_ = url_request_context_.CreateRequest(
       GURL("https://example.com/foo.html"), net::DEFAULT_PRIORITY,
       &url_request_delegate_);
   request_->set_method("GET");
   request_->Start();
+  base::RunLoop().RunUntilIdle();
+  EXPECT_TRUE(request_->status().is_io_pending());
+
+  stream_callback->OnAborted();
+  base::RunLoop().RunUntilIdle();
+  EXPECT_TRUE(request_->status().is_io_pending());
+
   std::string expected_response;
   expected_response.reserve((sizeof(kTestData) - 1) * 1024);
   for (int i = 0; i < 1024; ++i) {
     expected_response += kTestData;
-    stream->AddData(kTestData, sizeof(kTestData) - 1);
-    stream->Flush();
+    uint32_t written_bytes = sizeof(kTestData) - 1;
+    MojoResult result =
+        mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                           &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+    EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
     base::RunLoop().RunUntilIdle();
     EXPECT_EQ(expected_response, url_request_delegate_.data_received());
   }
-  stream->Finalize(net::OK);
+
+  data_pipe.producer_handle.reset();
   base::RunLoop().RunUntilIdle();
-  EXPECT_TRUE(request_->status().is_success());
+
+  EXPECT_EQ(net::ERR_CONNECTION_RESET, request_->status().ToNetError());
+  EXPECT_EQ(net::ERR_CONNECTION_RESET, url_request_delegate_.request_status());
   EXPECT_EQ(200,
             request_->response_headers()->response_code());
   EXPECT_EQ("OK",
@@ -793,45 +870,36 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_Flush) {
   EXPECT_FALSE(info->service_worker_ready_time().is_null());
 }
 
-TEST_F(ServiceWorkerURLRequestJobTest, StreamResponseAndCancel) {
-  const GURL stream_url("blob://stream");
-  StreamContext* stream_context =
-      GetStreamContextForResourceContext(
-          browser_context_->GetResourceContext());
-  scoped_refptr<Stream> stream =
-      new Stream(stream_context->registry(), nullptr, stream_url);
-  ASSERT_EQ(stream.get(),
-            stream_context->registry()->GetStream(stream_url).get());
-  SetUpWithHelper(new StreamResponder(stream_url));
+TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_AbortAfterData) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  SetUpWithHelper(new StreamResponder(mojo::MakeRequest(&stream_callback),
+                                      std::move(data_pipe.consumer_handle)));
 
   version_->SetStatus(ServiceWorkerVersion::ACTIVATED);
+
   request_ = url_request_context_.CreateRequest(
       GURL("https://example.com/foo.html"), net::DEFAULT_PRIORITY,
       &url_request_delegate_);
   request_->set_method("GET");
   request_->Start();
-  EXPECT_FALSE(HasWork());
   base::RunLoop().RunUntilIdle();
-  EXPECT_TRUE(HasWork());
+  EXPECT_TRUE(request_->status().is_io_pending());
 
-  std::string expected_response;
-  expected_response.reserve((sizeof(kTestData) - 1) * 1024);
-  for (int i = 0; i < 512; ++i) {
-    expected_response += kTestData;
-    stream->AddData(kTestData, sizeof(kTestData) - 1);
-  }
-  ASSERT_TRUE(stream_context->registry()->GetStream(stream_url).get());
-  request_->Cancel();
-  EXPECT_FALSE(HasWork());
-  ASSERT_FALSE(stream_context->registry()->GetStream(stream_url).get());
-  for (int i = 0; i < 512; ++i) {
-    expected_response += kTestData;
-    stream->AddData(kTestData, sizeof(kTestData) - 1);
-  }
-  stream->Finalize(net::OK);
+  data_pipe.producer_handle.reset();
+  base::RunLoop().RunUntilIdle();
+  EXPECT_TRUE(request_->status().is_io_pending());
 
+  stream_callback->OnAborted();
   base::RunLoop().RunUntilIdle();
+
+  EXPECT_FALSE(request_->status().is_io_pending());
   EXPECT_FALSE(request_->status().is_success());
+  EXPECT_EQ(net::ERR_CONNECTION_RESET, request_->status().ToNetError());
+  EXPECT_EQ(net::ERR_CONNECTION_RESET, url_request_delegate_.request_status());
+  EXPECT_EQ(200, request_->response_headers()->response_code());
+  EXPECT_EQ("OK", request_->response_headers()->GetStatusText());
+  EXPECT_EQ("", url_request_delegate_.data_received());
 
   EXPECT_EQ(0, times_prepare_to_restart_invoked_);
   ServiceWorkerResponseInfo* info =
@@ -846,13 +914,11 @@ TEST_F(ServiceWorkerURLRequestJobTest, StreamResponseAndCancel) {
   EXPECT_FALSE(info->service_worker_ready_time().is_null());
 }
 
-TEST_F(ServiceWorkerURLRequestJobTest,
-       StreamResponse_DelayedRegistrationAndCancel) {
-  const GURL stream_url("blob://stream");
-  StreamContext* stream_context =
-      GetStreamContextForResourceContext(
-          browser_context_->GetResourceContext());
-  SetUpWithHelper(new StreamResponder(stream_url));
+TEST_F(ServiceWorkerURLRequestJobTest, StreamResponse_ConsecutiveReadAndAbort) {
+  blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
+  mojo::DataPipe data_pipe;
+  SetUpWithHelper(new StreamResponder(mojo::MakeRequest(&stream_callback),
+                                      std::move(data_pipe.consumer_handle)));
 
   version_->SetStatus(ServiceWorkerVersion::ACTIVATED);
   request_ = url_request_context_.CreateRequest(
@@ -860,25 +926,41 @@ TEST_F(ServiceWorkerURLRequestJobTest,
       &url_request_delegate_);
   request_->set_method("GET");
   request_->Start();
-  EXPECT_FALSE(HasWork());
-  base::RunLoop().RunUntilIdle();
-  EXPECT_TRUE(HasWork());
-  request_->Cancel();
-  EXPECT_FALSE(HasWork());
+  std::string expected_response;
+  expected_response.reserve((sizeof(kTestData) - 1) * 1024);
+  for (int i = 0; i < 512; ++i) {
+    expected_response += kTestData;
+    uint32_t written_bytes = sizeof(kTestData) - 1;
+    MojoResult result =
+        mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
+                           &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+    EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
 
-  scoped_refptr<Stream> stream =
-      new Stream(stream_context->registry(), nullptr, stream_url);
-  // The stream should not be registered to the stream registry.
-  ASSERT_FALSE(stream_context->registry()->GetStream(stream_url).get());
-  for (int i = 0; i < 1024; ++i)
-    stream->AddData(kTestData, sizeof(kTestData) - 1);
-  stream->Finalize(net::OK);
+    base::RunLoop().RunUntilIdle();
+
+    EXPECT_EQ(expected_response, url_request_delegate_.data_received());
+  }
+  stream_callback->OnAborted();
 
   base::RunLoop().RunUntilIdle();
-  EXPECT_FALSE(request_->status().is_success());
+
+  EXPECT_TRUE(request_->status().is_success());
+  EXPECT_EQ(200, request_->response_headers()->response_code());
+  EXPECT_EQ("OK", request_->response_headers()->GetStatusText());
+  EXPECT_EQ(expected_response, url_request_delegate_.data_received());
 
   EXPECT_EQ(0, times_prepare_to_restart_invoked_);
-  EXPECT_FALSE(ServiceWorkerResponseInfo::ForRequest(request_.get()));
+  ServiceWorkerResponseInfo* info =
+      ServiceWorkerResponseInfo::ForRequest(request_.get());
+  ASSERT_TRUE(info);
+  EXPECT_TRUE(info->was_fetched_via_service_worker());
+  EXPECT_FALSE(info->was_fallback_required());
+  EXPECT_EQ(0u, info->url_list_via_service_worker().size());
+  EXPECT_EQ(blink::kWebServiceWorkerResponseTypeDefault,
+            info->response_type_via_service_worker());
+  EXPECT_FALSE(info->service_worker_start_time().is_null());
+  EXPECT_FALSE(info->service_worker_ready_time().is_null());
 }
 
 // Helper to simulate failing to dispatch a fetch event to a worker.
@@ -888,13 +970,15 @@ class FailFetchHelper : public EmbeddedWorkerTestHelper {
   ~FailFetchHelper() override {}
 
  protected:
-  void OnFetchEvent(int embedded_worker_id,
-                    int fetch_event_id,
-                    const ServiceWorkerFetchRequest& request,
-                    mojom::FetchEventPreloadHandlePtr preload_handle,
-                    const FetchCallback& callback) override {
+  void OnFetchEvent(
+      int embedded_worker_id,
+      int /* fetch_event_id */,
+      const ServiceWorkerFetchRequest& /* request */,
+      mojom::FetchEventPreloadHandlePtr /* preload_handle */,
+      mojom::ServiceWorkerFetchResponseCallbackPtr /* response_callback */,
+      const FetchCallback& finish_callback) override {
     SimulateWorkerStopped(embedded_worker_id);
-    callback.Run(SERVICE_WORKER_ERROR_ABORT, base::Time::Now());
+    finish_callback.Run(SERVICE_WORKER_ERROR_ABORT, base::Time::Now());
   }
 
  private:
@@ -969,33 +1053,33 @@ class EarlyResponseHelper : public EmbeddedWorkerTestHelper {
   ~EarlyResponseHelper() override {}
 
   void FinishWaitUntil() {
-    callback_.Run(SERVICE_WORKER_OK, base::Time::Now());
+    finish_callback_.Run(SERVICE_WORKER_OK, base::Time::Now());
   }
 
  protected:
-  void OnFetchEvent(int embedded_worker_id,
-                    int fetch_event_id,
-                    const ServiceWorkerFetchRequest& request,
-                    mojom::FetchEventPreloadHandlePtr preload_handle,
-                    const FetchCallback& callback) override {
-    callback_ = callback;
-    SimulateSend(new ServiceWorkerHostMsg_FetchEventResponse(
-        embedded_worker_id, fetch_event_id,
-        SERVICE_WORKER_FETCH_EVENT_RESULT_RESPONSE,
+  void OnFetchEvent(
+      int /* embedded_worker_id */,
+      int /* fetch_event_id */,
+      const ServiceWorkerFetchRequest& /* request */,
+      mojom::FetchEventPreloadHandlePtr /* preload_handle */,
+      mojom::ServiceWorkerFetchResponseCallbackPtr response_callback,
+      const FetchCallback& finish_callback) override {
+    finish_callback_ = finish_callback;
+    response_callback->OnResponse(
         ServiceWorkerResponse(
             base::MakeUnique<std::vector<GURL>>(), 200, "OK",
             blink::kWebServiceWorkerResponseTypeDefault,
             base::MakeUnique<ServiceWorkerHeaderMap>(), std::string(), 0,
-            GURL(), blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
+            blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
             false /* response_is_in_cache_storage */,
             std::string() /* response_cache_storage_cache_name */,
             base::MakeUnique<
                 ServiceWorkerHeaderList>() /* cors_exposed_header_names */),
-        base::Time::Now()));
+        base::Time::Now());
   }
 
  private:
-  FetchCallback callback_;
+  FetchCallback finish_callback_;
   DISALLOW_COPY_AND_ASSIGN(EarlyResponseHelper);
 };
 
@@ -1035,37 +1119,39 @@ class DelayedResponseHelper : public EmbeddedWorkerTestHelper {
   ~DelayedResponseHelper() override {}
 
   void Respond() {
-    SimulateSend(new ServiceWorkerHostMsg_FetchEventResponse(
-        embedded_worker_id_, fetch_event_id_,
-        SERVICE_WORKER_FETCH_EVENT_RESULT_RESPONSE,
+    response_callback_->OnResponse(
         ServiceWorkerResponse(
             base::MakeUnique<std::vector<GURL>>(), 200, "OK",
             blink::kWebServiceWorkerResponseTypeDefault,
             base::MakeUnique<ServiceWorkerHeaderMap>(), std::string(), 0,
-            GURL(), blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
+            blink::kWebServiceWorkerResponseErrorUnknown, base::Time(),
             false /* response_is_in_cache_storage */,
             std::string() /* response_cache_storage_cache_name */,
             base::MakeUnique<
                 ServiceWorkerHeaderList>() /* cors_exposed_header_names */),
-        base::Time::Now()));
-    callback_.Run(SERVICE_WORKER_OK, base::Time::Now());
+        base::Time::Now());
+    finish_callback_.Run(SERVICE_WORKER_OK, base::Time::Now());
   }
 
  protected:
-  void OnFetchEvent(int embedded_worker_id,
-                    int fetch_event_id,
-                    const ServiceWorkerFetchRequest& request,
-                    mojom::FetchEventPreloadHandlePtr preload_handle,
-                    const FetchCallback& callback) override {
+  void OnFetchEvent(
+      int embedded_worker_id,
+      int fetch_event_id,
+      const ServiceWorkerFetchRequest& /* request */,
+      mojom::FetchEventPreloadHandlePtr /* preload_handle */,
+      mojom::ServiceWorkerFetchResponseCallbackPtr response_callback,
+      const FetchCallback& finish_callback) override {
     embedded_worker_id_ = embedded_worker_id;
     fetch_event_id_ = fetch_event_id;
-    callback_ = callback;
+    response_callback_ = std::move(response_callback);
+    finish_callback_ = finish_callback;
   }
 
  private:
   int embedded_worker_id_ = 0;
   int fetch_event_id_ = 0;
-  FetchCallback callback_;
+  mojom::ServiceWorkerFetchResponseCallbackPtr response_callback_;
+  FetchCallback finish_callback_;
   DISALLOW_COPY_AND_ASSIGN(DelayedResponseHelper);
 };