Enable message ports for workers.

TEST=included ui test
Review URL: http://codereview.chromium.org/160576

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@22653 0039d316-1c4b-4281-b951-d872f2087c98

4 files changed, 92 insertions, 11 deletions

diff --git a/chrome/browser/worker_host/message_port_dispatcher.cc b/chrome/browser/worker_host/message_port_dispatcher.cc
index 9804e86..4d371de 100644
--- a/chrome/browser/worker_host/message_port_dispatcher.cc
+++ b/chrome/browser/worker_host/message_port_dispatcher.cc
@@ -58,6 +58,22 @@ bool MessagePortDispatcher::OnMessageReceived(
   return handled;
 }
 
+void MessagePortDispatcher::UpdateMessagePort(
+    int message_port_id,
+    IPC::Message::Sender* sender,
+    int routing_id,
+    CallbackWithReturnValue<int>::Type* next_routing_id) {
+  if (!message_ports_.count(message_port_id)) {
+    NOTREACHED();
+    return;
+  }
+
+  MessagePort& port = message_ports_[message_port_id];
+  port.sender = sender;
+  port.route_id = routing_id;
+  port.next_routing_id = next_routing_id;
+}
+
 bool MessagePortDispatcher::Send(IPC::Message* message) {
   return sender_->Send(message);
 }
@@ -84,7 +100,6 @@ void MessagePortDispatcher::OnDestroy(int message_port_id) {
   }
 
   DCHECK(message_ports_[message_port_id].queued_messages.empty());
-  delete message_ports_[message_port_id].next_routing_id;
   message_ports_.erase(message_port_id);
 }
 
diff --git a/chrome/browser/worker_host/message_port_dispatcher.h b/chrome/browser/worker_host/message_port_dispatcher.h
index d443b07..ae2536b 100644
--- a/chrome/browser/worker_host/message_port_dispatcher.h
+++ b/chrome/browser/worker_host/message_port_dispatcher.h
@@ -28,6 +28,12 @@ class MessagePortDispatcher : public NotificationObserver {
                          CallbackWithReturnValue<int>::Type* next_routing_id,
                          bool* message_was_ok);
 
+  void UpdateMessagePort(
+      int message_port_id,
+      IPC::Message::Sender* sender,
+      int routing_id,
+      CallbackWithReturnValue<int>::Type* next_routing_id);
+
   bool Send(IPC::Message* message);
 
  private:
diff --git a/chrome/browser/worker_host/worker_process_host.cc b/chrome/browser/worker_host/worker_process_host.cc
index ed3fd9f..09c9aa4 100644
--- a/chrome/browser/worker_host/worker_process_host.cc
+++ b/chrome/browser/worker_host/worker_process_host.cc
@@ -18,6 +18,7 @@
 #include "chrome/browser/browser_process.h"
 #include "chrome/browser/child_process_security_policy.h"
 #include "chrome/browser/renderer_host/render_view_host.h"
+#include "chrome/browser/renderer_host/resource_message_filter.h"
 #include "chrome/browser/worker_host/message_port_dispatcher.h"
 #include "chrome/browser/worker_host/worker_service.h"
 #include "chrome/common/chrome_switches.h"
@@ -62,8 +63,8 @@ class WorkerCrashTask : public Task {
 WorkerProcessHost::WorkerProcessHost(
     ResourceDispatcherHost* resource_dispatcher_host_)
     : ChildProcessHost(WORKER_PROCESS, resource_dispatcher_host_) {
-  next_route_id_ = NewCallbackWithReturnValue(
-      WorkerService::GetInstance(), &WorkerService::next_worker_route_id);
+  next_route_id_callback_.reset(NewCallbackWithReturnValue(
+      WorkerService::GetInstance(), &WorkerService::next_worker_route_id));
 }
 
 WorkerProcessHost::~WorkerProcessHost() {
@@ -148,9 +149,8 @@ bool WorkerProcessHost::FilterMessage(const IPC::Message& message,
   for (Instances::iterator i = instances_.begin(); i != instances_.end(); ++i) {
     if (i->sender_pid == sender_pid &&
         i->sender_route_id == message.routing_id()) {
-      IPC::Message* new_message = new IPC::Message(message);
-      new_message->set_routing_id(i->worker_route_id);
-      Send(new_message);
+      RelayMessage(
+          message, this, i->worker_route_id, next_route_id_callback_.get());
       return true;
     }
   }
@@ -167,7 +167,7 @@ URLRequestContext* WorkerProcessHost::GetRequestContext(
 void WorkerProcessHost::OnMessageReceived(const IPC::Message& message) {
   bool msg_is_ok = true;
   bool handled = MessagePortDispatcher::GetInstance()->OnMessageReceived(
-      message, this, next_route_id_, &msg_is_ok);
+      message, this, next_route_id_callback_.get(), &msg_is_ok);
 
   if (!handled) {
     handled = true;
@@ -192,9 +192,9 @@ void WorkerProcessHost::OnMessageReceived(const IPC::Message& message) {
 
   for (Instances::iterator i = instances_.begin(); i != instances_.end(); ++i) {
     if (i->worker_route_id == message.routing_id()) {
-      IPC::Message* new_message = new IPC::Message(message);
-      new_message->set_routing_id(i->sender_route_id);
-      i->sender->Send(new_message);
+      CallbackWithReturnValue<int>::Type* next_route_id =
+          GetNextRouteIdCallback(i->sender);
+      RelayMessage(message, i->sender, i->sender_route_id, next_route_id);
 
       if (message.type() == WorkerHostMsg_WorkerContextDestroyed::ID) {
         instances_.erase(i);
@@ -205,6 +205,55 @@ void WorkerProcessHost::OnMessageReceived(const IPC::Message& message) {
   }
 }
 
+CallbackWithReturnValue<int>::Type* WorkerProcessHost::GetNextRouteIdCallback(
+    IPC::Message::Sender* sender) {
+  // We don't keep callbacks for senders associated with workers, so figure out
+  // what kind of sender this is, and cast it to the correct class to get the
+  // callback.
+  for (ChildProcessHost::Iterator iter(ChildProcessInfo::WORKER_PROCESS);
+       !iter.Done(); ++iter) {
+    WorkerProcessHost* worker = static_cast<WorkerProcessHost*>(*iter);
+    if (static_cast<IPC::Message::Sender*>(worker) == sender)
+      return worker->next_route_id_callback_.get();
+  }
+
+  // Must be a ResourceMessageFilter.
+  return static_cast<ResourceMessageFilter*>(sender)->next_route_id_callback();
+}
+
+void WorkerProcessHost::RelayMessage(
+    const IPC::Message& message,
+    IPC::Message::Sender* sender,
+    int route_id,
+    CallbackWithReturnValue<int>::Type* next_route_id) {
+  IPC::Message* new_message;
+  if (message.type() == WorkerMsg_PostMessage::ID) {
+    // We want to send the receiver a routing id for the new channel, so
+    // crack the message first.
+    string16 msg;
+    int sent_message_port_id;
+    int new_routing_id;  // Ignore the bogus value from the sender.
+    if (!WorkerMsg_PostMessage::Read(
+            &message, &msg, &sent_message_port_id, &new_routing_id)) {
+      return;
+    }
+
+    if (sent_message_port_id != MSG_ROUTING_NONE) {
+      new_routing_id = next_route_id->Run();
+      MessagePortDispatcher::GetInstance()->UpdateMessagePort(
+          sent_message_port_id, sender, new_routing_id, next_route_id);
+    }
+
+    new_message = new WorkerMsg_PostMessage(
+        route_id, msg, sent_message_port_id, new_routing_id);
+  } else {
+    new_message = new IPC::Message(message);
+    new_message->set_routing_id(route_id);
+  }
+
+  sender->Send(new_message);
+}
+
 void WorkerProcessHost::SenderShutdown(IPC::Message::Sender* sender) {
   for (Instances::iterator i = instances_.begin(); i != instances_.end();) {
     if (i->sender == sender) {
diff --git a/chrome/browser/worker_host/worker_process_host.h b/chrome/browser/worker_host/worker_process_host.h
index 44bc676..648d731 100644
--- a/chrome/browser/worker_host/worker_process_host.h
+++ b/chrome/browser/worker_host/worker_process_host.h
@@ -57,6 +57,17 @@ class WorkerProcessHost : public ChildProcessHost {
   // Called when a message arrives from the worker process.
   void OnMessageReceived(const IPC::Message& message);
 
+  // Given a Sender, returns the callback that generates a new routing id.
+  static CallbackWithReturnValue<int>::Type* GetNextRouteIdCallback(
+      IPC::Message::Sender* sender);
+
+  // Relays a message to the given endpoint.  Takes care of parsing the message
+  // if it contains a message port and sending it a valid route id.
+  static void RelayMessage(const IPC::Message& message,
+                           IPC::Message::Sender* sender,
+                           int route_id,
+                           CallbackWithReturnValue<int>::Type* next_route_id);
+
   virtual bool CanShutdown() { return instances_.empty(); }
 
   // Updates the title shown in the task manager.
@@ -71,7 +82,7 @@ class WorkerProcessHost : public ChildProcessHost {
   Instances instances_;
 
   // A callback to create a routing id for the associated worker process.
-  CallbackWithReturnValue<int>::Type* next_route_id_;
+  scoped_ptr<CallbackWithReturnValue<int>::Type> next_route_id_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(WorkerProcessHost);
 };