content/child/webmessageportchannel_impl.cc

   1 // Copyright (c) 2011 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "content/child/webmessageportchannel_impl.h"
   6
   7 #include "base/bind.h"
   8 #include "base/values.h"
   9 #include "content/child/child_process.h"
  10 #include "content/child/child_thread_impl.h"
  11 #include "content/common/message_port_messages.h"
  12 #include "content/public/child/v8_value_converter.h"
  13 #include "third_party/WebKit/public/platform/WebMessagePortChannelClient.h"
  14 #include "third_party/WebKit/public/platform/WebString.h"
  15 #include "third_party/WebKit/public/web/WebSerializedScriptValue.h"
  16 #include "v8/include/v8.h"
  17
  18 using blink::WebMessagePortChannel;
  19 using blink::WebMessagePortChannelArray;
  20 using blink::WebMessagePortChannelClient;
  21 using blink::WebString;
  22
  23 namespace content {
  24
  25 WebMessagePortChannelImpl::WebMessagePortChannelImpl(
  26     const scoped_refptr<base::SingleThreadTaskRunner>& main_thread_task_runner)
  27     : client_(NULL),
  28       route_id_(MSG_ROUTING_NONE),
  29       message_port_id_(MSG_ROUTING_NONE),
  30       send_messages_as_values_(false),
  31       main_thread_task_runner_(main_thread_task_runner) {
  32   AddRef();
  33   Init();
  34 }
  35
  36 WebMessagePortChannelImpl::WebMessagePortChannelImpl(
  37     int route_id,
  38     const TransferredMessagePort& port,
  39     const scoped_refptr<base::SingleThreadTaskRunner>& main_thread_task_runner)
  40     : client_(NULL),
  41       route_id_(route_id),
  42       message_port_id_(port.id),
  43       send_messages_as_values_(port.send_messages_as_values),
  44       main_thread_task_runner_(main_thread_task_runner) {
  45   AddRef();
  46   Init();
  47 }
  48
  49 WebMessagePortChannelImpl::~WebMessagePortChannelImpl() {
  50   // If we have any queued messages with attached ports, manually destroy them.
  51   while (!message_queue_.empty()) {
  52     const WebMessagePortChannelArray& channel_array =
  53         message_queue_.front().ports;
  54     for (size_t i = 0; i < channel_array.size(); i++) {
  55       channel_array[i]->destroy();
  56     }
  57     message_queue_.pop();
  58   }
  59
  60   if (message_port_id_ != MSG_ROUTING_NONE)
  61     Send(new MessagePortHostMsg_DestroyMessagePort(message_port_id_));
  62
  63   if (route_id_ != MSG_ROUTING_NONE)
  64     ChildThreadImpl::current()->GetRouter()->RemoveRoute(route_id_);
  65 }
  66
  67 // static
  68 void WebMessagePortChannelImpl::CreatePair(
  69     const scoped_refptr<base::SingleThreadTaskRunner>& main_thread_task_runner,
  70     blink::WebMessagePortChannel** channel1,
  71     blink::WebMessagePortChannel** channel2) {
  72   WebMessagePortChannelImpl* impl1 =
  73       new WebMessagePortChannelImpl(main_thread_task_runner);
  74   WebMessagePortChannelImpl* impl2 =
  75       new WebMessagePortChannelImpl(main_thread_task_runner);
  76
  77   impl1->Entangle(impl2);
  78   impl2->Entangle(impl1);
  79
  80   *channel1 = impl1;
  81   *channel2 = impl2;
  82 }
  83
  84 // static
  85 std::vector<TransferredMessagePort>
  86 WebMessagePortChannelImpl::ExtractMessagePortIDs(
  87     scoped_ptr<WebMessagePortChannelArray> channels) {
  88   std::vector<TransferredMessagePort> message_ports;
  89   if (channels)
  90     message_ports = ExtractMessagePortIDs(*channels);
  91   return message_ports;
  92 }
  93
  94 // static
  95 std::vector<TransferredMessagePort>
  96 WebMessagePortChannelImpl::ExtractMessagePortIDs(
  97     const WebMessagePortChannelArray& channels) {
  98   std::vector<TransferredMessagePort> message_ports(channels.size());
  99   for (size_t i = 0; i < channels.size(); ++i) {
 100     WebMessagePortChannelImpl* webchannel =
 101         static_cast<WebMessagePortChannelImpl*>(channels[i]);
 102     // The message port ids might not be set up yet if this channel
 103     // wasn't created on the main thread.
 104     DCHECK(webchannel->main_thread_task_runner_->BelongsToCurrentThread());
 105     message_ports[i].id = webchannel->message_port_id();
 106     message_ports[i].send_messages_as_values =
 107         webchannel->send_messages_as_values_;
 108     webchannel->QueueMessages();
 109     DCHECK(message_ports[i].id != MSG_ROUTING_NONE);
 110   }
 111   return message_ports;
 112 }
 113
 114 // static
 115 std::vector<TransferredMessagePort>
 116 WebMessagePortChannelImpl::ExtractMessagePortIDsWithoutQueueing(
 117     scoped_ptr<WebMessagePortChannelArray> channels) {
 118   if (!channels)
 119     return std::vector<TransferredMessagePort>();
 120
 121   std::vector<TransferredMessagePort> message_ports(channels->size());
 122   for (size_t i = 0; i < channels->size(); ++i) {
 123     WebMessagePortChannelImpl* webchannel =
 124         static_cast<WebMessagePortChannelImpl*>((*channels)[i]);
 125     // The message port ids might not be set up yet if this channel
 126     // wasn't created on the main thread.
 127     DCHECK(webchannel->main_thread_task_runner_->BelongsToCurrentThread());
 128     message_ports[i].id = webchannel->message_port_id();
 129     message_ports[i].send_messages_as_values =
 130         webchannel->send_messages_as_values_;
 131     // Don't queue messages, but do increase the child processes ref-count to
 132     // ensure this child process stays alive long enough to receive all
 133     // in-flight messages.
 134     ChildProcess::current()->AddRefProcess();
 135     DCHECK(message_ports[i].id != MSG_ROUTING_NONE);
 136   }
 137   return message_ports;
 138 }
 139
 140 // static
 141 WebMessagePortChannelArray WebMessagePortChannelImpl::CreatePorts(
 142     const std::vector<TransferredMessagePort>& message_ports,
 143     const std::vector<int>& new_routing_ids,
 144     const scoped_refptr<base::SingleThreadTaskRunner>&
 145         main_thread_task_runner) {
 146   DCHECK_EQ(message_ports.size(), new_routing_ids.size());
 147   WebMessagePortChannelArray channels(message_ports.size());
 148   for (size_t i = 0; i < message_ports.size() && i < new_routing_ids.size();
 149        ++i) {
 150     channels[i] = new WebMessagePortChannelImpl(
 151         new_routing_ids[i], message_ports[i],
 152         main_thread_task_runner);
 153   }
 154   return channels;
 155 }
 156
 157 void WebMessagePortChannelImpl::setClient(WebMessagePortChannelClient* client) {
 158   // Must lock here since client_ is called on the main thread.
 159   base::AutoLock auto_lock(lock_);
 160   client_ = client;
 161 }
 162
 163 void WebMessagePortChannelImpl::destroy() {
 164   setClient(NULL);
 165
 166   // Release the object on the main thread, since the destructor might want to
 167   // send an IPC, and that has to happen on the main thread.
 168   main_thread_task_runner_->ReleaseSoon(FROM_HERE, this);
 169 }
 170
 171 void WebMessagePortChannelImpl::postMessage(
 172     const WebString& message_as_string,
 173     WebMessagePortChannelArray* channels_ptr) {
 174   MessagePortMessage message(message_as_string);
 175   scoped_ptr<WebMessagePortChannelArray> channels(channels_ptr);
 176   if (send_messages_as_values_) {
 177     blink::WebSerializedScriptValue serialized_value =
 178         blink::WebSerializedScriptValue::fromString(message_as_string);
 179     v8::Local<v8::Value> v8_value = serialized_value.deserialize();
 180     scoped_ptr<V8ValueConverter> converter(V8ValueConverter::create());
 181     converter->SetDateAllowed(true);
 182     converter->SetRegExpAllowed(true);
 183     scoped_ptr<base::Value> message_as_value(converter->FromV8Value(
 184         v8_value, v8::Isolate::GetCurrent()->GetCurrentContext()));
 185     message = MessagePortMessage(message_as_value.Pass());
 186   }
 187   if (!main_thread_task_runner_->BelongsToCurrentThread()) {
 188     main_thread_task_runner_->PostTask(
 189         FROM_HERE, base::Bind(&WebMessagePortChannelImpl::PostMessage, this,
 190                               message, base::Passed(channels.Pass())));
 191   } else {
 192     PostMessage(message, channels.Pass());
 193   }
 194 }
 195
 196 void WebMessagePortChannelImpl::PostMessage(
 197     const MessagePortMessage& message,
 198     scoped_ptr<WebMessagePortChannelArray> channels) {
 199   IPC::Message* msg = new MessagePortHostMsg_PostMessage(
 200       message_port_id_, message, ExtractMessagePortIDs(channels.Pass()));
 201   Send(msg);
 202 }
 203
 204 bool WebMessagePortChannelImpl::tryGetMessage(
 205     WebString* message,
 206     WebMessagePortChannelArray& channels) {
 207   base::AutoLock auto_lock(lock_);
 208   if (message_queue_.empty())
 209     return false;
 210
 211   const MessagePortMessage& data = message_queue_.front().message;
 212   DCHECK(data.is_string() != data.is_value());
 213   if (data.is_value()) {
 214     v8::HandleScope handle_scope(client_->scriptIsolate());
 215     v8::Context::Scope context_scope(
 216         client_->scriptContextForMessageConversion());
 217     scoped_ptr<V8ValueConverter> converter(V8ValueConverter::create());
 218     converter->SetDateAllowed(true);
 219     converter->SetRegExpAllowed(true);
 220     v8::Local<v8::Value> v8_value = converter->ToV8Value(
 221         data.as_value(), client_->scriptContextForMessageConversion());
 222     blink::WebSerializedScriptValue serialized_value =
 223         blink::WebSerializedScriptValue::serialize(v8_value);
 224     *message = serialized_value.toString();
 225   } else {
 226     *message = message_queue_.front().message.message_as_string;
 227   }
 228   channels = message_queue_.front().ports;
 229   message_queue_.pop();
 230   return true;
 231 }
 232
 233 void WebMessagePortChannelImpl::Init() {
 234   if (!main_thread_task_runner_->BelongsToCurrentThread()) {
 235     main_thread_task_runner_->PostTask(
 236         FROM_HERE, base::Bind(&WebMessagePortChannelImpl::Init, this));
 237     return;
 238   }
 239
 240   if (route_id_ == MSG_ROUTING_NONE) {
 241     DCHECK(message_port_id_ == MSG_ROUTING_NONE);
 242     Send(new MessagePortHostMsg_CreateMessagePort(
 243         &route_id_, &message_port_id_));
 244   } else if (message_port_id_ != MSG_ROUTING_NONE) {
 245     Send(new MessagePortHostMsg_ReleaseMessages(message_port_id_));
 246   }
 247
 248   ChildThreadImpl::current()->GetRouter()->AddRoute(route_id_, this);
 249 }
 250
 251 void WebMessagePortChannelImpl::Entangle(
 252     scoped_refptr<WebMessagePortChannelImpl> channel) {
 253   // The message port ids might not be set up yet, if this channel wasn't
 254   // created on the main thread.  So need to wait until we're on the main thread
 255   // before getting the other message port id.
 256   if (!main_thread_task_runner_->BelongsToCurrentThread()) {
 257     main_thread_task_runner_->PostTask(
 258         FROM_HERE,
 259         base::Bind(&WebMessagePortChannelImpl::Entangle, this, channel));
 260     return;
 261   }
 262
 263   Send(new MessagePortHostMsg_Entangle(
 264       message_port_id_, channel->message_port_id()));
 265 }
 266
 267 void WebMessagePortChannelImpl::QueueMessages() {
 268   if (!main_thread_task_runner_->BelongsToCurrentThread()) {
 269     main_thread_task_runner_->PostTask(
 270         FROM_HERE, base::Bind(&WebMessagePortChannelImpl::QueueMessages, this));
 271     return;
 272   }
 273   // This message port is being sent elsewhere (perhaps to another process).
 274   // The new endpoint needs to receive the queued messages, including ones that
 275   // could still be in-flight.  So we tell the browser to queue messages, and it
 276   // sends us an ack, whose receipt we know means that no more messages are
 277   // in-flight.  We then send the queued messages to the browser, which prepends
 278   // them to the ones it queued and it sends them to the new endpoint.
 279   Send(new MessagePortHostMsg_QueueMessages(message_port_id_));
 280
 281   // The process could potentially go away while we're still waiting for
 282   // in-flight messages.  Ensure it stays alive.
 283   ChildProcess::current()->AddRefProcess();
 284 }
 285
 286 void WebMessagePortChannelImpl::Send(IPC::Message* message) {
 287   if (!main_thread_task_runner_->BelongsToCurrentThread()) {
 288     DCHECK(!message->is_sync());
 289     main_thread_task_runner_->PostTask(
 290         FROM_HERE,
 291         base::Bind(&WebMessagePortChannelImpl::Send, this, message));
 292     return;
 293   }
 294
 295   ChildThreadImpl::current()->GetRouter()->Send(message);
 296 }
 297
 298 bool WebMessagePortChannelImpl::OnMessageReceived(const IPC::Message& message) {
 299   bool handled = true;
 300   IPC_BEGIN_MESSAGE_MAP(WebMessagePortChannelImpl, message)
 301     IPC_MESSAGE_HANDLER(MessagePortMsg_Message, OnMessage)
 302     IPC_MESSAGE_HANDLER(MessagePortMsg_MessagesQueued, OnMessagesQueued)
 303     IPC_MESSAGE_UNHANDLED(handled = false)
 304   IPC_END_MESSAGE_MAP()
 305   return handled;
 306 }
 307
 308 void WebMessagePortChannelImpl::OnMessage(
 309     const MessagePortMessage& message,
 310     const std::vector<TransferredMessagePort>& sent_message_ports,
 311     const std::vector<int>& new_routing_ids) {
 312   base::AutoLock auto_lock(lock_);
 313   Message msg;
 314   msg.message = message;
 315   msg.ports = CreatePorts(sent_message_ports, new_routing_ids,
 316                           main_thread_task_runner_.get());
 317
 318   bool was_empty = message_queue_.empty();
 319   message_queue_.push(msg);
 320   if (client_ && was_empty)
 321     client_->messageAvailable();
 322 }
 323
 324 void WebMessagePortChannelImpl::OnMessagesQueued() {
 325   std::vector<QueuedMessage> queued_messages;
 326
 327   {
 328     base::AutoLock auto_lock(lock_);
 329     queued_messages.reserve(message_queue_.size());
 330     while (!message_queue_.empty()) {
 331       MessagePortMessage message = message_queue_.front().message;
 332       std::vector<TransferredMessagePort> ports =
 333           ExtractMessagePortIDs(message_queue_.front().ports);
 334       queued_messages.push_back(std::make_pair(message, ports));
 335       message_queue_.pop();
 336     }
 337   }
 338
 339   Send(new MessagePortHostMsg_SendQueuedMessages(
 340       message_port_id_, queued_messages));
 341
 342   message_port_id_ = MSG_ROUTING_NONE;
 343
 344   Release();
 345   ChildProcess::current()->ReleaseProcess();
 346 }
 347
 348 WebMessagePortChannelImpl::Message::Message() {}
 349
 350 WebMessagePortChannelImpl::Message::~Message() {}
 351
 352 }  // namespace content