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

content/renderer/service_worker/service_worker_context_client.cc

 // Copyright 2015 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/renderer/service_worker/service_worker_context_client.h"
 
+#include <map>
 #include <memory>
 #include <utility>
 
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/memory/ptr_util.h"
 #include "base/metrics/histogram_macros.h"
 #include "base/strings/utf_string_conversions.h"
 #include "base/threading/thread_checker.h"
 #include "base/threading/thread_local.h"
@@ skipping 86 matching lines @@
     // Service workers are only available in secure contexts, so all requests
     // are initiated in a secure context.
     extra_data->set_initiated_in_secure_context(true);
     request.SetExtraData(extra_data.release());
   }
 
  private:
   std::unique_ptr<ServiceWorkerNetworkProvider> provider_;
 };
 
+class StreamHandleListener
+    : public blink::WebServiceWorkerStreamHandle::Listener {
+ public:
+  StreamHandleListener(
+      blink::mojom::ServiceWorkerStreamCallbackPtr callback_ptr)
+      : callback_ptr_(std::move(callback_ptr)) {}
+
+  void OnAborted() override { callback_ptr_->OnAborted(); }
+  void OnCompleted() override { callback_ptr_->OnCompleted(); }
+
+ private:
+  blink::mojom::ServiceWorkerStreamCallbackPtr callback_ptr_;
+};
+
 ServiceWorkerStatusCode EventResultToStatus(
     blink::WebServiceWorkerEventResult result) {
   switch (result) {
     case blink::kWebServiceWorkerEventResultCompleted:
       return SERVICE_WORKER_OK;
     case blink::kWebServiceWorkerEventResultRejected:
       return SERVICE_WORKER_ERROR_EVENT_WAITUNTIL_REJECTED;
   }
   NOTREACHED() << "Got invalid result: " << result;
   return SERVICE_WORKER_ERROR_FAILED;
@@ skipping 89 matching lines @@
   web_response->SetStatusText(blink::WebString::FromUTF8(response.status_text));
   web_response->SetResponseType(response.response_type);
   for (const auto& pair : response.headers) {
     web_response->SetHeader(blink::WebString::FromUTF8(pair.first),
                             blink::WebString::FromUTF8(pair.second));
   }
   if (!response.blob_uuid.empty()) {
     web_response->SetBlob(blink::WebString::FromASCII(response.blob_uuid),
                           response.blob_size);
   }
-  web_response->SetStreamURL(blink::WebURL(response.stream_url));
   web_response->SetError(response.error);
   web_response->SetResponseTime(response.response_time.ToInternalValue());
   if (response.is_in_cache_storage) {
     web_response->SetCacheStorageCacheName(
         blink::WebString::FromUTF8(response.cache_storage_cache_name));
   }
 
   std::vector<blink::WebString> cors_exposed_header_names;
   for (const auto& name : response.cors_exposed_header_names)
     cors_exposed_header_names.push_back(blink::WebString::FromUTF8(name));
 
   web_response->SetCorsExposedHeaderNames(
       blink::WebVector<blink::WebString>(cors_exposed_header_names));
 }
 
 // Use this template in willDestroyWorkerContext to abort all the pending
 // events callbacks.
 template <typename T>
 void AbortPendingEventCallbacks(T& callbacks) {
   for (typename T::iterator it(&callbacks); !it.IsAtEnd(); it.Advance()) {
     it.GetCurrentValue()->Run(SERVICE_WORKER_ERROR_ABORT, base::Time::Now());
   }
 }
 
+template <typename Key, typename Callback>
+void AbortPendingEventCallbacks(std::map<Key, Callback>& callbacks) {
+  for (const auto& it : callbacks)
+    it.second.Run(SERVICE_WORKER_ERROR_ABORT, base::Time::Now());
+}
+
 }  // namespace
 
 // Holding data that needs to be bound to the worker context on the
 // worker thread.
 struct ServiceWorkerContextClient::WorkerContextData {
+  using SimpleEventCallback =
+      base::Callback<void(ServiceWorkerStatusCode, base::Time)>;
   using ClientsCallbacksMap =
       IDMap<std::unique_ptr<blink::WebServiceWorkerClientsCallbacks>>;
   using ClaimClientsCallbacksMap =
       IDMap<std::unique_ptr<blink::WebServiceWorkerClientsClaimCallbacks>>;
   using ClientCallbacksMap =
       IDMap<std::unique_ptr<blink::WebServiceWorkerClientCallbacks>>;
   using SkipWaitingCallbacksMap =
       IDMap<std::unique_ptr<blink::WebServiceWorkerSkipWaitingCallbacks>>;
   using ActivateEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchActivateEventCallback>>;
   using BackgroundFetchAbortEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchBackgroundFetchAbortEventCallback>>;
   using BackgroundFetchClickEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchBackgroundFetchClickEventCallback>>;
   using BackgroundFetchFailEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchBackgroundFetchFailEventCallback>>;
   using BackgroundFetchedEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchBackgroundFetchedEventCallback>>;
   using SyncEventCallbacksMap = IDMap<std::unique_ptr<const SyncCallback>>;
   using NotificationClickEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchNotificationClickEventCallback>>;
   using NotificationCloseEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchNotificationCloseEventCallback>>;
   using PushEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchPushEventCallback>>;
-  using FetchEventCallbacksMap = IDMap<std::unique_ptr<const FetchCallback>>;
   using ExtendableMessageEventCallbacksMap =
       IDMap<std::unique_ptr<const DispatchExtendableMessageEventCallback>>;
   using NavigationPreloadRequestsMap = IDMap<
       std::unique_ptr<ServiceWorkerContextClient::NavigationPreloadRequest>>;
 
   explicit WorkerContextData(ServiceWorkerContextClient* owner)
       : event_dispatcher_binding(owner),
         weak_factory(owner),
         proxy_weak_factory(owner->proxy_) {}
 
@@ skipping 46 matching lines @@
   // Pending callbacks for Notification Click Events.
   NotificationClickEventCallbacksMap notification_click_event_callbacks;
 
   // Pending callbacks for Notification Close Events.
   NotificationCloseEventCallbacksMap notification_close_event_callbacks;
 
   // Pending callbacks for Push Events.
   PushEventCallbacksMap push_event_callbacks;
 
   // Pending callbacks for Fetch Events.
-  FetchEventCallbacksMap fetch_event_callbacks;
+  std::map<int /* fetch_event_id */, SimpleEventCallback> fetch_event_callbacks;
+
+  // Pending callbacks for respondWith on each fetch event.
+  std::map<int /* fetch_event_id */,
+           mojom::ServiceWorkerFetchResponseCallbackPtr>
+      fetch_response_callbacks;
 
   // Pending callbacks for Extendable Message Events.
   ExtendableMessageEventCallbacksMap message_event_callbacks;
 
   // Pending navigation preload requests.
   NavigationPreloadRequestsMap preload_requests;
 
   base::ThreadChecker thread_checker;
   base::WeakPtrFactory<ServiceWorkerContextClient> weak_factory;
   base::WeakPtrFactory<blink::WebServiceWorkerContextProxy> proxy_weak_factory;
@@ skipping 317 matching lines @@
           context, service_worker_version_id_, script_url_);
 }
 
 void ServiceWorkerContextClient::WillDestroyWorkerContext(
     v8::Local<v8::Context> context) {
   // At this point WillStopCurrentWorkerThread is already called, so
   // worker_task_runner_->RunsTasksOnCurrentThread() returns false
   // (while we're still on the worker thread).
   proxy_ = NULL;
 
   // Aborts all the pending events callbacks.

[horo, 2017/04/10 09:24:38]

Is it OK not to abort all fetch_response_callbacks?

[shimazu, 2017/04/11 01:32:58]

Good point. There is no behavior change in my understanding, but we should know
clearly what will happen if it's aborted.
Currently no fetch response seems triggered when it's aborted, but it might have
to be handled a kind of error like a network error.
WDYT?

[horo, 2017/04/11 05:41:53]

Even if we don't call all fetch_response_callback,
ServiceWorkerVersion::OnStoppedInternal() aborts all pending request.
So it is OK for the pending requests in fetch_response_callbacks.

https://chromium.googlesource.com/chromium/src/+/25f662809b16e810def434f02226...

But after the worker started streaming, the current code can't handle the
abortion correctly.
I think if you change FetchDataLoaderAsDataPipe::OnStateChange() to call
DidFetchDataLoadedDataPipe() before Cancel(),
you can treat the MOJO_HANDLE_SIGNAL_PEER_CLOSED signal before OnCompleted() as
an error in ServiceWorkerDataPipeReader.

   AbortPendingEventCallbacks(context_->activate_event_callbacks);
   AbortPendingEventCallbacks(context_->background_fetch_abort_event_callbacks);
   AbortPendingEventCallbacks(context_->background_fetch_click_event_callbacks);
   AbortPendingEventCallbacks(context_->background_fetch_fail_event_callbacks);
   AbortPendingEventCallbacks(context_->background_fetched_event_callbacks);
   AbortPendingEventCallbacks(context_->sync_event_callbacks);
   AbortPendingEventCallbacks(context_->notification_click_event_callbacks);
   AbortPendingEventCallbacks(context_->notification_close_event_callbacks);
   AbortPendingEventCallbacks(context_->push_event_callbacks);
   AbortPendingEventCallbacks(context_->fetch_event_callbacks);
@@ skipping 158 matching lines @@
     blink::WebServiceWorkerEventResult result,
     double event_dispatch_time) {
   Send(new ServiceWorkerHostMsg_InstallEventFinished(
       GetRoutingID(), request_id, result, proxy_->HasFetchEventHandler(),
       base::Time::FromDoubleT(event_dispatch_time)));
 }
 
 void ServiceWorkerContextClient::RespondToFetchEvent(
     int fetch_event_id,
     double event_dispatch_time) {
-  Send(new ServiceWorkerHostMsg_FetchEventResponse(
-      GetRoutingID(), fetch_event_id,
-      SERVICE_WORKER_FETCH_EVENT_RESULT_FALLBACK, ServiceWorkerResponse(),
-      base::Time::FromDoubleT(event_dispatch_time)));
+  const mojom::ServiceWorkerFetchResponseCallbackPtr& response_callback =
+      context_->fetch_response_callbacks[fetch_event_id];
+  DCHECK(response_callback.is_bound());
+  response_callback->OnFallback(base::Time::FromDoubleT(event_dispatch_time));
+  context_->fetch_response_callbacks.erase(fetch_event_id);
 }
 
-void ServiceWorkerContextClient::RespondToFetchEvent(
+void ServiceWorkerContextClient::RespondToFetchEventWithResponse(
     int fetch_event_id,
     const blink::WebServiceWorkerResponse& web_response,
     double event_dispatch_time) {
-  Send(new ServiceWorkerHostMsg_FetchEventResponse(
-      GetRoutingID(), fetch_event_id,
-      SERVICE_WORKER_FETCH_EVENT_RESULT_RESPONSE,
-      GetServiceWorkerResponseFromWebResponse(web_response),
-      base::Time::FromDoubleT(event_dispatch_time)));
+  ServiceWorkerResponse response(
+      GetServiceWorkerResponseFromWebResponse(web_response));
+  const mojom::ServiceWorkerFetchResponseCallbackPtr& response_callback =
+      context_->fetch_response_callbacks[fetch_event_id];
+  if (response.blob_uuid.size()) {
+    // Send the legacy IPC due to the ordering issue between respondWith and
+    // blob.
+    // TODO(shimazu): mojofy this IPC after blob starts using sync IPC for
+    // creation or is mojofied: https://crbug.com/611935.
+    Send(new ServiceWorkerHostMsg_FetchEventResponse(
+        GetRoutingID(), fetch_event_id, response,
+        base::Time::FromDoubleT(event_dispatch_time)));
+  } else {
+    response_callback->OnResponse(response,
+                                  base::Time::FromDoubleT(event_dispatch_time));
+  }
+  context_->fetch_response_callbacks.erase(fetch_event_id);
+}
+
+void ServiceWorkerContextClient::RespondToFetchEventWithResponseStream(
+    int fetch_event_id,
+    const blink::WebServiceWorkerResponse& web_response,
+    blink::WebServiceWorkerStreamHandle* web_stream_handle,
+    double event_dispatch_time) {
+  ServiceWorkerResponse response(
+      GetServiceWorkerResponseFromWebResponse(web_response));
+  const mojom::ServiceWorkerFetchResponseCallbackPtr& response_callback =
+      context_->fetch_response_callbacks[fetch_event_id];
+  blink::mojom::ServiceWorkerStreamHandlePtr stream_handle =
+      blink::mojom::ServiceWorkerStreamHandle::New();
+  blink::mojom::ServiceWorkerStreamCallbackPtr callback_ptr;
+  stream_handle->callback_request = mojo::MakeRequest(&callback_ptr);
+  stream_handle->stream = web_stream_handle->DrainStreamDataPipe();
+
+  web_stream_handle->SetListener(
+      base::MakeUnique<StreamHandleListener>(std::move(callback_ptr)));
+
+  response_callback->OnResponseStream(
+      response, std::move(stream_handle),
+      base::Time::FromDoubleT(event_dispatch_time));
+  context_->fetch_response_callbacks.erase(fetch_event_id);
 }
 
 void ServiceWorkerContextClient::DidHandleFetchEvent(
     int fetch_event_id,
     blink::WebServiceWorkerEventResult result,
     double event_dispatch_time) {
-  const FetchCallback* callback =
-      context_->fetch_event_callbacks.Lookup(fetch_event_id);
+  const WorkerContextData::SimpleEventCallback& callback =
+      context_->fetch_event_callbacks[fetch_event_id];
   DCHECK(callback);
-  callback->Run(EventResultToStatus(result),
-                base::Time::FromDoubleT(event_dispatch_time));
-  context_->fetch_event_callbacks.Remove(fetch_event_id);
+  callback.Run(EventResultToStatus(result),
+               base::Time::FromDoubleT(event_dispatch_time));
+  context_->fetch_event_callbacks.erase(fetch_event_id);
 }
 
 void ServiceWorkerContextClient::DidHandleNotificationClickEvent(
     int request_id,
     blink::WebServiceWorkerEventResult result,
     double event_dispatch_time) {
   const DispatchNotificationClickEventCallback* callback =
       context_->notification_click_event_callbacks.Lookup(request_id);
   DCHECK(callback);
 
@@ skipping 329 matching lines @@
 void ServiceWorkerContextClient::OnInstallEvent(int request_id) {
   TRACE_EVENT0("ServiceWorker",
                "ServiceWorkerContextClient::OnInstallEvent");
   proxy_->DispatchInstallEvent(request_id);
 }
 
 void ServiceWorkerContextClient::DispatchFetchEvent(
     int fetch_event_id,
     const ServiceWorkerFetchRequest& request,
     mojom::FetchEventPreloadHandlePtr preload_handle,
+    mojom::ServiceWorkerFetchResponseCallbackPtr response_callback,
     const DispatchFetchEventCallback& callback) {
   std::unique_ptr<NavigationPreloadRequest> preload_request =
       preload_handle
           ? base::MakeUnique<NavigationPreloadRequest>(
                 fetch_event_id, request.url, std::move(preload_handle))
           : nullptr;
   const bool navigation_preload_sent = !!preload_request;
   TRACE_EVENT0("ServiceWorker",
                "ServiceWorkerContextClient::DispatchFetchEvent");
-  context_->fetch_event_callbacks.AddWithID(
-      base::MakeUnique<FetchCallback>(callback), fetch_event_id);
+  context_->fetch_response_callbacks.insert(
+      std::make_pair(fetch_event_id, std::move(response_callback)));
+  context_->fetch_event_callbacks.insert(
+      std::make_pair(fetch_event_id, std::move(callback)));
   if (preload_request) {
     context_->preload_requests.AddWithID(std::move(preload_request),
                                          fetch_event_id);
   }
 
   blink::WebServiceWorkerRequest web_request;
   ToWebServiceWorkerRequest(request, &web_request);
 
   if (request.fetch_type == ServiceWorkerFetchType::FOREIGN_FETCH) {
     proxy_->DispatchForeignFetchEvent(fetch_event_id, web_request);
@@ skipping 263 matching lines @@
 }
 
 base::WeakPtr<ServiceWorkerContextClient>
 ServiceWorkerContextClient::GetWeakPtr() {
   DCHECK(worker_task_runner_->RunsTasksOnCurrentThread());
   DCHECK(context_);
   return context_->weak_factory.GetWeakPtr();
 }
 
 }  // namespace content