Initial implementation of stashed message ports, content side

content/browser/service_worker/stashed_port_manager.cc

+// Copyright 2015 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/stashed_port_manager.h"
+
+#include "content/browser/message_port_message_filter.h"
+#include "content/browser/message_port_service.h"
+#include "content/browser/service_worker/service_worker_context_wrapper.h"
+#include "content/public/browser/browser_thread.h"
+
+namespace content {
+
+struct StashedPortManager::StashedPort {
+  StashedPort();
+  ~StashedPort();
+
+  int64 service_worker_registration_id;
+  GURL service_worker_origin;
+
+  int message_port_id;
+  base::string16 name;
+
+  // The |route_id| to the port in the process this port currently lives in, or
+  // MSG_ROUTING_NONE if the port isn't currently associated with any running
+  // ServiceWorkerVersion.
+  int route_id;
+  // The running ServiceWorkerVersion this port is currently associated with.
+  // Set to null if the port does not currently exist in a running worker.
+  ServiceWorkerVersion* service_worker;
+};
+
+StashedPortManager::StashedPort::StashedPort() {
+}

[kinuko, 2015/05/19 09:01:45]

nit: please have one empty line between methods

[Marijn Kruisselbrink, 2015/05/19 17:40:18]

Done.

+StashedPortManager::StashedPort::~StashedPort() {
+}
+
+StashedPortManager::StashedPortManager(
+    const scoped_refptr<ServiceWorkerContextWrapper>& service_worker_context)
+    : service_worker_context_(service_worker_context) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+}
+
+StashedPortManager::~StashedPortManager() {
+}
+
+void StashedPortManager::Init() {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  BrowserThread::PostTask(BrowserThread::IO, FROM_HERE,
+                          base::Bind(&StashedPortManager::InitOnIO, this));
+}
+
+void StashedPortManager::Shutdown() {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  BrowserThread::PostTask(BrowserThread::IO, FROM_HERE,
+                          base::Bind(&StashedPortManager::ShutdownOnIO, this));
+}
+
+void StashedPortManager::InitOnIO() {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  // TODO(mek): Restore ports between service workers from storage.
+}
+
+void StashedPortManager::ShutdownOnIO() {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  for (ServiceWorkerVersion* worker : observing_service_workers_) {
+    worker->RemoveListener(this);
+  }
+  observing_service_workers_.clear();
+  MessagePortService::GetInstance()->OnMessagePortDelegateClosing(this);
+}
+
+void StashedPortManager::AddPort(ServiceWorkerVersion* service_worker,
+                                 int message_port_id,
+                                 const base::string16& name) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  DCHECK(ports_.find(message_port_id) == ports_.end());
+
+  StashedPort& port = ports_[message_port_id];
+  port.service_worker_registration_id = service_worker->registration_id();
+  port.service_worker_origin = service_worker->scope().GetOrigin();
+  port.message_port_id = message_port_id;
+  port.name = name;
+
+  // If service worker is currently running, get its current route id and
+  // start observing the worker.
+  if (service_worker->running_status() == ServiceWorkerVersion::RUNNING) {
+    MessagePortService::GetInstance()->GetMessagePortInfo(
+        message_port_id, nullptr, &port.route_id);
+    port.service_worker = service_worker;
+    if (observing_service_workers_.insert(service_worker).second) {
+      // Service Worker wasn't already being observed
+      service_worker->AddListener(this);
+    }
+  } else {
+    port.route_id = MSG_ROUTING_NONE;
+    port.service_worker = nullptr;
+  }
+
+  MessagePortService::GetInstance()->UpdateMessagePort(message_port_id, this,
+                                                       message_port_id);
+
+  // If service worker is not currently running, all messages to this port
+  // should be held in the browser process.
+  if (port.route_id == MSG_ROUTING_NONE)
+    MessagePortService::GetInstance()->HoldMessages(message_port_id);
+}
+
+void StashedPortManager::SendMessage(
+    int message_port_id,
+    const MessagePortMessage& message,
+    const std::vector<TransferredMessagePort>& sent_message_ports) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  DCHECK(ports_.find(message_port_id) != ports_.end());
+
+  const StashedPort& port = ports_[message_port_id];
+  DCHECK(port.service_worker && port.route_id != MSG_ROUTING_NONE);
+  port.service_worker->embedded_worker()
+      ->message_port_message_filter()
+      ->SendMessage(port.route_id, message, sent_message_ports);
+}
+
+void StashedPortManager::MessageWasHeld(int message_port_id) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  DCHECK(ports_.find(message_port_id) != ports_.end());
+
+  // Messages are queued in the browser process when the stashed port isn't
+  // currently available in any running renderer processes. Try to transfer
+  // the port to a (potentially newly started) service worker to enable
+  // message delivery.
+  const StashedPort& port = ports_[message_port_id];
+  service_worker_context_->FindRegistrationForId(
+      port.service_worker_registration_id, port.service_worker_origin,
+      base::Bind(&StashedPortManager::TransferMessagePort, this,
+                 message_port_id));
+}
+
+void StashedPortManager::SendMessagesAreQueued(int message_port_id) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  DCHECK(ports_.find(message_port_id) != ports_.end());
+  // TODO(mek): Support transfering a stashed message port to a different
+  // process.
+}
+
+void StashedPortManager::OnRunningStateChanged(
+    ServiceWorkerVersion* service_worker) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  // Listener is only added to service workers whose running state is RUNNING,
+  // so running state here should only ever be able to be STOPPING or STOPPED.
+  DCHECK(service_worker->running_status() == ServiceWorkerVersion::STOPPING ||
+         service_worker->running_status() == ServiceWorkerVersion::STOPPED);
+
+  // Hold messages for all ports associated with the no longer running worker.
+  for (auto& port : ports_) {
+    if (port.second.service_worker != service_worker)
+      continue;
+    port.second.route_id = MSG_ROUTING_NONE;
+    port.second.service_worker = nullptr;
+    MessagePortService::GetInstance()->HoldMessages(port.first);
+  }
+  observing_service_workers_.erase(service_worker);

[kinuko, 2015/05/15 06:59:18]

Is it guaranteed that OnRunningStateChanged is always fired before
ServiceWorkerVersion goes away?

[Marijn Kruisselbrink, 2015/05/18 22:19:36]

Actually I'm not sure. I misread a comment in ~ServiceWorkerVersion
"EmbeddedWorker's dtor sends StopWorker if it's still running." to indicate that
this would always be the case, but looking at this more carefully the comment
doesn't mean what I thought it meant.
I'm not sure under what circumstances a ServiceWorkerVersion instance for a
still running service worker would get deleted though, but if that happens it
seems OnRunningStateChanged won't get fired (and there doesn't seem to be any
other way to be notified of something like that happening either).

[kinuko, 2015/05/19 09:01:45]

My guess is it's not guaranteed, but falken@ would probably know better.

I think we can change the current code to make it guaranteed though, it looks
easy to do so.

[Marijn Kruisselbrink, 2015/05/19 17:40:19]

That's also my conclusion, yes.
Do you want me to file a bug for that?

+  service_worker->RemoveListener(this);
+}
+
+void StashedPortManager::TransferMessagePort(
+    int message_port_id,
+    ServiceWorkerStatusCode service_worker_status,
+    const scoped_refptr<ServiceWorkerRegistration>&
+        service_worker_registration) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  auto it = ports_.find(message_port_id);
+  if (it == ports_.end()) {
+    // Port no longer owned by StashedPortManager, no need to do anything else.
+    return;
+  }
+  const StashedPort& port = it->second;
+
+  if (service_worker_status != SERVICE_WORKER_OK) {
+    // TODO(mek): SW no longer exists, somehow handle this.
+    return;
+  }
+
+  // TODO(mek): Figure out and spec correct logic for determining which version
+  // of a service worker a port should be associated with.
+  scoped_refptr<ServiceWorkerVersion> version =
+      service_worker_registration->active_version();
+  if (!version)
+    version = service_worker_registration->waiting_version();
+  if (!version)
+    version = service_worker_registration->installing_version();
+  if (!version) {
+    // TODO(mek): Do something when no version is found.
+    return;
+  }
+
+  std::vector<TransferredMessagePort> ports(1);
+  std::vector<base::string16> port_names(1);
+  ports[0].id = port.message_port_id;
+  port_names[0] = port.name;
+  version->SendStashedMessagePorts(
+      ports, port_names, base::Bind(&StashedPortManager::OnPortsTransferred,
+                                    this, version, ports));
+}
+
+void StashedPortManager::OnPortsTransferred(
+    const scoped_refptr<ServiceWorkerVersion>& service_worker,
+    const std::vector<TransferredMessagePort>& ports,
+    ServiceWorkerStatusCode service_worker_status,
+    const std::vector<int>& route_ids) {
+  DCHECK_CURRENTLY_ON(BrowserThread::IO);
+  if (service_worker_status != SERVICE_WORKER_OK) {
+    // TODO(mek): Handle failure.
+    return;
+  }
+
+  if (service_worker->running_status() != ServiceWorkerVersion::RUNNING) {
+    // TODO(mek): Figure out what to do in this case.
+    return;
+  }
+
+  // Port was transfered to a service worker, start observing the worker so
+  // messages can be held when the worker stops running.
+  if (observing_service_workers_.insert(service_worker.get()).second) {
+    // Service Worker wasn't already being observed
+    service_worker->AddListener(this);
+  }
+
+  // Update ports with the new route ids and service worker version.
+  DCHECK(ports.size() == route_ids.size());
+  for (size_t i = 0; i < ports.size(); ++i) {
+    DCHECK(ports_.find(ports[i].id) != ports_.end());
+    StashedPort& port = ports_[ports[i].id];
+    port.route_id = route_ids[i];
+    port.service_worker = service_worker.get();
+  }
+}
+
+}  // namespace content