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