Make sure message ports that are transferred to a serviceworker end up in the right process.

This still has one remaining issue in that there are some situation where MessagePortService in the browser process might unqueue these transferred message ports before the service worker has been started. That will be fixed in a separate CL.

https://codereview.chromium.org/730543005/ has a test for this fix.

BUG=432678

Review URL: https://codereview.chromium.org/733703002

Cr-Commit-Position: refs/heads/master@{#305074}

content/browser/service_worker/embedded_worker_instance.cc

@@ -340,6 +340,11 @@ void EmbeddedWorkerInstance::OnReportConsoleMessage(
           source_identifier, message_level, message, line_number, source_url));
 }
 
+MessagePortMessageFilter* EmbeddedWorkerInstance::message_port_message_filter()
+    const {
+  return registry_->MessagePortMessageFilterForProcess(process_id_);
+}
+
 void EmbeddedWorkerInstance::AddListener(Listener* listener) {
   listener_list_.AddObserver(listener);
 }

content/browser/service_worker/embedded_worker_instance.h

@@ -30,6 +30,7 @@ class Message;
 namespace content {
 
 class EmbeddedWorkerRegistry;
+class MessagePortMessageFilter;
 class ServiceWorkerContextCore;
 struct ServiceWorkerFetchRequest;
 
@@ -98,6 +99,7 @@ class CONTENT_EXPORT EmbeddedWorkerInstance {
   int worker_devtools_agent_route_id() const {
     return worker_devtools_agent_route_id_;
   }
+  MessagePortMessageFilter* message_port_message_filter() const;
 
   void AddListener(Listener* listener);
   void RemoveListener(Listener* listener);

content/browser/service_worker/embedded_worker_registry.cc

@@ -182,13 +182,18 @@ void EmbeddedWorkerRegistry::OnReportConsoleMessage(
 }
 
 void EmbeddedWorkerRegistry::AddChildProcessSender(
-    int process_id, IPC::Sender* sender) {
+    int process_id,
+    IPC::Sender* sender,
+    MessagePortMessageFilter* message_port_message_filter) {
   process_sender_map_[process_id] = sender;
+  process_message_port_message_filter_map_[process_id] =
+      message_port_message_filter;
   DCHECK(!ContainsKey(worker_process_map_, process_id));
 }
 
 void EmbeddedWorkerRegistry::RemoveChildProcessSender(int process_id) {
   process_sender_map_.erase(process_id);
+  process_message_port_message_filter_map_.erase(process_id);
   std::map<int, std::set<int> >::iterator found =
       worker_process_map_.find(process_id);
   if (found != worker_process_map_.end()) {
@@ -220,6 +225,11 @@ bool EmbeddedWorkerRegistry::CanHandle(int embedded_worker_id) const {
   return true;
 }
 
+MessagePortMessageFilter*
+EmbeddedWorkerRegistry::MessagePortMessageFilterForProcess(int process_id) {
+  return process_message_port_message_filter_map_[process_id];
+}
+
 EmbeddedWorkerRegistry::EmbeddedWorkerRegistry(
     const base::WeakPtr<ServiceWorkerContextCore>& context,
     int initial_embedded_worker_id)

content/browser/service_worker/embedded_worker_registry.h

@@ -28,6 +28,7 @@ class Sender;
 namespace content {
 
 class EmbeddedWorkerInstance;
+class MessagePortMessageFilter;
 class ServiceWorkerContextCore;
 
 // Acts as a thin stub between MessageFilter and each EmbeddedWorkerInstance,
@@ -91,7 +92,10 @@ class CONTENT_EXPORT EmbeddedWorkerRegistry
                               const GURL& source_url);
 
   // Keeps a map from process_id to sender information.
-  void AddChildProcessSender(int process_id, IPC::Sender* sender);
+  void AddChildProcessSender(
+      int process_id,
+      IPC::Sender* sender,
+      MessagePortMessageFilter* message_port_message_filter);
   void RemoveChildProcessSender(int process_id);
 
   // Returns an embedded worker instance for given |embedded_worker_id|.
@@ -100,12 +104,16 @@ class CONTENT_EXPORT EmbeddedWorkerRegistry
   // Returns true if |embedded_worker_id| is managed by this registry.
   bool CanHandle(int embedded_worker_id) const;
 
+  MessagePortMessageFilter* MessagePortMessageFilterForProcess(int process_id);
+
  private:
   friend class base::RefCounted<EmbeddedWorkerRegistry>;
   friend class EmbeddedWorkerInstance;
 
   typedef std::map<int, EmbeddedWorkerInstance*> WorkerInstanceMap;
   typedef std::map<int, IPC::Sender*> ProcessToSenderMap;
+  typedef std::map<int, MessagePortMessageFilter*>
+      ProcessToMessagePortMessageFilterMap;
 
   EmbeddedWorkerRegistry(
       const base::WeakPtr<ServiceWorkerContextCore>& context,
@@ -122,6 +130,7 @@ class CONTENT_EXPORT EmbeddedWorkerRegistry
 
   WorkerInstanceMap worker_map_;
   ProcessToSenderMap process_sender_map_;
+  ProcessToMessagePortMessageFilterMap process_message_port_message_filter_map_;
 
   // Map from process_id to embedded_worker_id.
   // This map only contains starting and running workers.

content/browser/service_worker/embedded_worker_test_helper.cc

@@ -33,7 +33,7 @@ EmbeddedWorkerTestHelper::EmbeddedWorkerTestHelper(int mock_render_process_id)
                          NULL,
                          NULL);
   wrapper_->process_manager()->SetProcessIdForTest(mock_render_process_id);
-  registry()->AddChildProcessSender(mock_render_process_id, this);
+  registry()->AddChildProcessSender(mock_render_process_id, this, nullptr);
 }
 
 EmbeddedWorkerTestHelper::~EmbeddedWorkerTestHelper() {
@@ -44,7 +44,7 @@ EmbeddedWorkerTestHelper::~EmbeddedWorkerTestHelper() {
 void EmbeddedWorkerTestHelper::SimulateAddProcessToPattern(
     const GURL& pattern,
     int process_id) {
-  registry()->AddChildProcessSender(process_id, this);
+  registry()->AddChildProcessSender(process_id, this, nullptr);
   wrapper_->process_manager()->AddProcessReferenceToPattern(
       pattern, process_id);
 }

content/browser/service_worker/service_worker_dispatcher_host.cc

@@ -122,7 +122,7 @@ void ServiceWorkerDispatcherHost::Init(
 
   context_wrapper_ = context_wrapper;
   GetContext()->embedded_worker_registry()->AddChildProcessSender(
-      render_process_id_, this);
+      render_process_id_, this, message_port_message_filter_);
 }
 
 void ServiceWorkerDispatcherHost::OnFilterAdded(IPC::Sender* sender) {
@@ -470,13 +470,8 @@ void ServiceWorkerDispatcherHost::OnPostMessageToWorker(
     return;
   }
 
-  std::vector<int> new_routing_ids;
-  message_port_message_filter_->UpdateMessagePortsWithNewRoutes(
-      sent_message_port_ids, &new_routing_ids);
-  handle->version()->SendMessage(
-      ServiceWorkerMsg_MessageToWorker(message,
-                                       sent_message_port_ids,
-                                       new_routing_ids),
+  handle->version()->DispatchMessageEvent(
+      message, sent_message_port_ids,
       base::Bind(&ServiceWorkerUtils::NoOpStatusCallback));
 }
 

content/browser/service_worker/service_worker_version.cc

@@ -7,6 +7,7 @@
 #include "base/command_line.h"
 #include "base/stl_util.h"
 #include "base/strings/string16.h"
+#include "content/browser/message_port_message_filter.h"
 #include "content/browser/service_worker/embedded_worker_instance.h"
 #include "content/browser/service_worker/embedded_worker_registry.h"
 #include "content/browser/service_worker/service_worker_context_core.h"
@@ -247,6 +248,30 @@ void ServiceWorkerVersion::SendMessage(
   RunSoon(base::Bind(callback, status));
 }
 
+void ServiceWorkerVersion::DispatchMessageEvent(
+    const base::string16& message,
+    const std::vector<int>& sent_message_port_ids,
+    const StatusCallback& callback) {
+  if (running_status() != RUNNING) {
+    // Schedule calling this method after starting the worker.
+    StartWorker(base::Bind(
+        &RunTaskAfterStartWorker, weak_factory_.GetWeakPtr(), callback,
+        base::Bind(&self::DispatchMessageEvent, weak_factory_.GetWeakPtr(),
+                   message, sent_message_port_ids, callback)));
+    return;
+  }
+
+  MessagePortMessageFilter* filter =
+      embedded_worker_->message_port_message_filter();
+  std::vector<int> new_routing_ids;
+  filter->UpdateMessagePortsWithNewRoutes(sent_message_port_ids,
+                                          &new_routing_ids);
+  ServiceWorkerStatusCode status =
+      embedded_worker_->SendMessage(ServiceWorkerMsg_MessageToWorker(
+          message, sent_message_port_ids, new_routing_ids));
+  RunSoon(base::Bind(callback, status));
+}
+
 void ServiceWorkerVersion::DispatchInstallEvent(
     int active_version_id,
     const StatusCallback& callback) {

content/browser/service_worker/service_worker_version.h

@@ -155,6 +155,11 @@ class CONTENT_EXPORT ServiceWorkerVersion
   // message is successfully sent or not.
   void SendMessage(const IPC::Message& message, const StatusCallback& callback);
 
+  // Sends a message event to the associated embedded worker.
+  void DispatchMessageEvent(const base::string16& message,
+                            const std::vector<int>& sent_message_port_ids,
+                            const StatusCallback& callback);
+
   // Sends install event to the associated embedded worker and asynchronously
   // calls |callback| when it errors out or it gets a response from the worker
   // to notify install completion.