Update navigator.services API to use the new services.onconnect event [2/3].

content/browser/service_worker/service_worker_version.cc

 // Copyright 2013 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_version.h"
 
 #include <map>
 #include <string>
 
 #include "base/command_line.h"
@@ skipping 147 matching lines @@
     const std::vector<TransferredMessagePort>& sent_message_ports,
     const ServiceWorkerVersion::StatusCallback& callback,
     ServiceWorkerStatusCode status) {
   // Transfering the message ports failed, so destroy the ports.
   for (const TransferredMessagePort& port : sent_message_ports) {
     MessagePortService::GetInstance()->ClosePort(port.id);
   }
   callback.Run(status);
 }
 
-void RunErrorCrossOriginConnectCallback(
-    const ServiceWorkerVersion::CrossOriginConnectCallback& callback,
+void RunErrorServicePortConnectCallback(
+    const ServiceWorkerVersion::ServicePortConnectCallback& callback,
     ServiceWorkerStatusCode status) {
-  callback.Run(status, false /* accept_connection */);
+  callback.Run(status, false /* accept_connection */, base::string16(),
+               base::string16());
 }
 
 void RunErrorSendStashedPortsCallback(
     const ServiceWorkerVersion::SendStashedPortsCallback& callback,
     ServiceWorkerStatusCode status) {
   callback.Run(status, std::vector<int>());
 }
 
 using WindowOpenedCallback = base::Callback<void(int, int)>;
 
@@ skipping 705 matching lines @@
       AddRequest(callback, &geofencing_callbacks_, REQUEST_GEOFENCING);
   ServiceWorkerStatusCode status =
       embedded_worker_->SendMessage(ServiceWorkerMsg_GeofencingEvent(
           request_id, event_type, region_id, region));
   if (status != SERVICE_WORKER_OK) {
     geofencing_callbacks_.Remove(request_id);
     RunSoon(base::Bind(callback, status));
   }
 }
 
-void ServiceWorkerVersion::DispatchCrossOriginConnectEvent(
-    const CrossOriginConnectCallback& callback,
-    const NavigatorConnectClient& client) {
+void ServiceWorkerVersion::DispatchServicePortConnectEvent(
+    const ServicePortConnectCallback& callback,
+    const GURL& target_url,
+    const GURL& origin,
+    int port_id) {
   DCHECK_EQ(ACTIVATED, status()) << status();
 
   if (!base::CommandLine::ForCurrentProcess()->HasSwitch(
           switches::kEnableExperimentalWebPlatformFeatures)) {
-    callback.Run(SERVICE_WORKER_ERROR_ABORT, false);
+    callback.Run(SERVICE_WORKER_ERROR_ABORT, false, base::string16(),
+                 base::string16());
     return;
   }
 
   if (running_status() != RUNNING) {
     // Schedule calling this method after starting the worker.
     StartWorker(
         base::Bind(&RunTaskAfterStartWorker, weak_factory_.GetWeakPtr(),
-                   base::Bind(&RunErrorCrossOriginConnectCallback, callback),
-                   base::Bind(&self::DispatchCrossOriginConnectEvent,
-                              weak_factory_.GetWeakPtr(), callback, client)));
+                   base::Bind(&RunErrorServicePortConnectCallback, callback),
+                   base::Bind(&self::DispatchServicePortConnectEvent,
+                              weak_factory_.GetWeakPtr(), callback, target_url,
+                              origin, port_id)));
     return;
   }
 
-  int request_id = AddRequest(callback, &cross_origin_connect_callbacks_,
-                              REQUEST_CROSS_ORIGIN_CONNECT);
-  ServiceWorkerStatusCode status = embedded_worker_->SendMessage(
-      ServiceWorkerMsg_CrossOriginConnectEvent(request_id, client));
-  if (status != SERVICE_WORKER_OK) {
-    cross_origin_connect_callbacks_.Remove(request_id);
-    RunSoon(base::Bind(callback, status, false));
+  int request_id = AddRequest(callback, &service_port_connect_callbacks_,
+                              REQUEST_SERVICE_PORT_CONNECT);
+  if (!service_port_dispatcher_) {
+    embedded_worker_->GetServiceRegistry()->ConnectToRemoteService(
+        mojo::GetProxy(&service_port_dispatcher_));
+    service_port_dispatcher_.set_connection_error_handler(
+        base::Bind(&ServiceWorkerVersion::OnMojoConnectionError<
+                       ServicePortDispatcher, ServicePortConnectCallback>,
+                   weak_factory_.GetWeakPtr(), &service_port_dispatcher_,
+                   &service_port_connect_callbacks_,
+                   base::Bind(&RunErrorServicePortConnectCallback)));
   }
+  service_port_dispatcher_->Connect(

[michaeln, 2015/07/15 02:47:25]

More curiousity about where the "message pipe" fifo ordering is guaranteed. 
* between MojoApplicationHost and MojoApplication
* between each ServiceRegistryImpl  pair
* between each Service pair

I'd assume its between each Service pair, but just checking?

Also, rockot mentioned that a when you make a mojo call with a callback, no
other calls go thru until completion. The caller can issue more Connect() calls,
but mojo will sit on them until a prior connect call completes. (I hope thats
not right, but i think thats what i heard?)

 Connect(string target_url, string origin, int32 port_id)
       => (ServicePortConnectResult result, string name, string data);

The existing ipc code code delivers ServiceWorkerMsg_CrossOriginConnectEvent
msgs back to back giving the remote side a chance to process them concurrently
if possible. Do we lose that impl detail with this switch to mojo?

[Marijn Kruisselbrink, 2015/07/15 20:49:40]

Yes, I'm pretty sure it's between each Service pair.
I'd be surprised if that is true (and from
https://groups.google.com/a/chromium.org/d/msg/mojo-dev/7kNfPXtVLxM/IaCbIgCI2v4J
it appears that isn't true, although I might have phrased my question wrong),
although it wouldn't be too disastrous here, since with one service per worker
multiple calls will end up being serialized/queued somewhere anyway.

[michaeln, 2015/07/16 22:23:46]

Right, "there is one FIFO per interface"
 
Thnx for checking. Your question and the answer looks clear enough to me. Nice
that mojo will complain about neglecting to return when the idl (err mojom)
describes a return.

+      mojo::String::From(target_url), mojo::String::From(origin), port_id,
+      base::Bind(&ServiceWorkerVersion::OnServicePortConnectEventFinished,
+                 weak_factory_.GetWeakPtr(), request_id));
 }
 
 void ServiceWorkerVersion::DispatchCrossOriginMessageEvent(
     const NavigatorConnectClient& client,
     const base::string16& message,
     const std::vector<TransferredMessagePort>& sent_message_ports,
     const StatusCallback& callback) {
   // Unlike in the case of DispatchMessageEvent, here the caller is assumed to
   // have already put all the sent message ports on hold. So no need to do that
   // here again.
@@ skipping 247 matching lines @@
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_FetchEventFinished,
                         OnFetchEventFinished)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_SyncEventFinished,
                         OnSyncEventFinished)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_NotificationClickEventFinished,
                         OnNotificationClickEventFinished)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_PushEventFinished,
                         OnPushEventFinished)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_GeofencingEventFinished,
                         OnGeofencingEventFinished)
-    IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_CrossOriginConnectEventFinished,
-                        OnCrossOriginConnectEventFinished)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_OpenWindow,
                         OnOpenWindow)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_SetCachedMetadata,
                         OnSetCachedMetadata)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_ClearCachedMetadata,
                         OnClearCachedMetadata)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_PostMessageToClient,
                         OnPostMessageToClient)
     IPC_MESSAGE_HANDLER(ServiceWorkerHostMsg_FocusClient,
                         OnFocusClient)
@@ skipping 218 matching lines @@
   if (!callback) {
     NOTREACHED() << "Got unexpected message: " << request_id;
     return;
   }
 
   scoped_refptr<ServiceWorkerVersion> protect(this);
   callback->Run(SERVICE_WORKER_OK);
   RemoveCallbackAndStopIfRedundant(&geofencing_callbacks_, request_id);
 }
 
-void ServiceWorkerVersion::OnCrossOriginConnectEventFinished(
+void ServiceWorkerVersion::OnServicePortConnectEventFinished(
     int request_id,
-    bool accept_connection) {
+    ServicePortConnectResult result,
+    const mojo::String& name,
+    const mojo::String& data) {
   TRACE_EVENT1("ServiceWorker",
-               "ServiceWorkerVersion::OnCrossOriginConnectEventFinished",
+               "ServiceWorkerVersion::OnServicePortConnectEventFinished",
                "Request id", request_id);
-  CrossOriginConnectCallback* callback =
-      cross_origin_connect_callbacks_.Lookup(request_id);
+  ServicePortConnectCallback* callback =
+      service_port_connect_callbacks_.Lookup(request_id);
   if (!callback) {
     NOTREACHED() << "Got unexpected message: " << request_id;
     return;
   }
 
   scoped_refptr<ServiceWorkerVersion> protect(this);
-  callback->Run(SERVICE_WORKER_OK, accept_connection);
-  RemoveCallbackAndStopIfRedundant(&cross_origin_connect_callbacks_,
+  callback->Run(SERVICE_WORKER_OK, result == SERVICE_PORT_CONNECT_RESULT_ACCEPT,
+                name.To<base::string16>(), data.To<base::string16>());
+  RemoveCallbackAndStopIfRedundant(&service_port_connect_callbacks_,
                                    request_id);
 }
 
 void ServiceWorkerVersion::OnOpenWindow(int request_id, GURL url) {
   // Just abort if we are shutting down.
   if (!context_)
     return;
 
   if (!url.is_valid()) {
     DVLOG(1) << "Received unexpected invalid URL from renderer process.";
@@ skipping 479 matching lines @@
     return;
   }
 
   // TODO(falken): We may need to handle StopIfIdle failure and
   // forcibly fail pending callbacks so no one is stuck waiting
   // for the worker.
   embedded_worker_->StopIfIdle();
 }
 
 bool ServiceWorkerVersion::HasInflightRequests() const {
-  return
-    !activate_callbacks_.IsEmpty() ||
-    !install_callbacks_.IsEmpty() ||
-    !fetch_callbacks_.IsEmpty() ||
-    !sync_callbacks_.IsEmpty() ||
-    !notification_click_callbacks_.IsEmpty() ||
-    !push_callbacks_.IsEmpty() ||
-    !geofencing_callbacks_.IsEmpty() ||
-    !cross_origin_connect_callbacks_.IsEmpty() ||
-    !streaming_url_request_jobs_.empty();
+  return !activate_callbacks_.IsEmpty() || !install_callbacks_.IsEmpty() ||
+         !fetch_callbacks_.IsEmpty() || !sync_callbacks_.IsEmpty() ||
+         !notification_click_callbacks_.IsEmpty() ||
+         !push_callbacks_.IsEmpty() || !geofencing_callbacks_.IsEmpty() ||
+         !service_port_connect_callbacks_.IsEmpty() ||
+         !streaming_url_request_jobs_.empty();
 }
 
 void ServiceWorkerVersion::RecordStartWorkerResult(
     ServiceWorkerStatusCode status) {
   base::TimeTicks start_time = start_time_;
   ClearTick(&start_time_);
 
   ServiceWorkerMetrics::RecordStartWorkerStatus(status,
                                                 IsInstalled(prestart_status_));
 
@@ skipping 67 matching lines @@
                               SERVICE_WORKER_ERROR_TIMEOUT);
     case REQUEST_NOTIFICATION_CLICK:
       return RunIDMapCallback(&notification_click_callbacks_, info.id,
                               SERVICE_WORKER_ERROR_TIMEOUT);
     case REQUEST_PUSH:
       return RunIDMapCallback(&push_callbacks_, info.id,
                               SERVICE_WORKER_ERROR_TIMEOUT);
     case REQUEST_GEOFENCING:
       return RunIDMapCallback(&geofencing_callbacks_, info.id,
                               SERVICE_WORKER_ERROR_TIMEOUT);
-    case REQUEST_CROSS_ORIGIN_CONNECT:
-      return RunIDMapCallback(&cross_origin_connect_callbacks_, info.id,
+    case REQUEST_SERVICE_PORT_CONNECT:
+      return RunIDMapCallback(&service_port_connect_callbacks_, info.id,
                               SERVICE_WORKER_ERROR_TIMEOUT,
-                              false /* accept_connection */);
+                              false /* accept_connection */, base::string16(),
+                              base::string16());
   }
   NOTREACHED() << "Got unexpected request type: " << info.type;
   return false;
 }
 
 void ServiceWorkerVersion::SetAllRequestTimes(const base::TimeTicks& ticks) {
   std::queue<RequestInfo> new_requests;
   while (!requests_.empty()) {
     RequestInfo info = requests_.front();
     info.time = ticks;
@@ skipping 97 matching lines @@
   RunIDMapCallbacks(&install_callbacks_,
                     SERVICE_WORKER_ERROR_INSTALL_WORKER_FAILED);
   RunIDMapCallbacks(&fetch_callbacks_, SERVICE_WORKER_ERROR_FAILED,
                     SERVICE_WORKER_FETCH_EVENT_RESULT_FALLBACK,
                     ServiceWorkerResponse());
   RunIDMapCallbacks(&sync_callbacks_, SERVICE_WORKER_ERROR_FAILED);
   RunIDMapCallbacks(&notification_click_callbacks_,
                     SERVICE_WORKER_ERROR_FAILED);
   RunIDMapCallbacks(&push_callbacks_, SERVICE_WORKER_ERROR_FAILED);
   RunIDMapCallbacks(&geofencing_callbacks_, SERVICE_WORKER_ERROR_FAILED);
-  RunIDMapCallbacks(&cross_origin_connect_callbacks_,
-                    SERVICE_WORKER_ERROR_FAILED, false);
+
+  // Close all mojo services. This will also fire and clear all callbacks
+  // for messages that are still outstanding for those services.

[michaeln, 2015/07/15 02:47:26]

I'm curious. Prior to getting here, the embeddedworkerinstance will have reset
the registry from whence this dispatcher came. What affect does that registry
reset have on this dispatcher object?

[Marijn Kruisselbrink, 2015/07/15 20:49:40]

The registry reset has no effect on dispatcher objects. This is where I was
hoping mojo had some kind of way to make connections depend on the lifetime of
other connections to automatically clean up all connections to services when the
registry is closed, but that doesn't happen. Each channel is fully independent.

[michaeln, 2015/07/16 22:23:46]

Got it (or maybe starting to get it). In mojo terms, i guess "each message pipe
is fully independent" would be more accurate, and each instance of an
interfaceptr/interfaceimpl has a distinct pipe.

+  service_port_dispatcher_.reset();
 
   streaming_url_request_jobs_.clear();
 
   FOR_EACH_OBSERVER(Listener, listeners_, OnRunningStateChanged(this));
 
   if (should_restart)
     StartWorkerInternal();
 }
 
+template <typename Interface, typename Callback>
+void ServiceWorkerVersion::OnMojoConnectionError(
+    mojo::InterfacePtr<Interface>* interface,
+    IDMap<Callback, IDMapOwnPointer>* callbacks,
+    const base::Callback<void(const Callback&, ServiceWorkerStatusCode)>
+        callback) {
+  typename IDMap<Callback, IDMapOwnPointer>::iterator iter(callbacks);
+  while (!iter.IsAtEnd()) {
+    callback.Run(*iter.GetCurrentValue(), SERVICE_WORKER_ERROR_FAILED);
+    iter.Advance();
+  }
+  callbacks->Clear();
+}
+
 }  // namespace content