Implement ServiceWorkerVersion::SendMessage() (for dispatching events etc)

content/browser/service_worker/service_worker_version.cc

Index: content/browser/service_worker/service_worker_version.cc
diff --git a/content/browser/service_worker/service_worker_version.cc b/content/browser/service_worker/service_worker_version.cc
index 6389d75d6eaea9aedb8df9777a3d5e2bd410cadd..8fa9a16dd09d8b713277ea09eec703334d0850b3 100644
--- a/content/browser/service_worker/service_worker_version.cc
+++ b/content/browser/service_worker/service_worker_version.cc
@@ -13,10 +13,14 @@
 
 namespace content {
 
+typedef ServiceWorkerVersion::StatusCallback StatusCallback;
+typedef ServiceWorkerVersion::MessageCallback MessageCallback;
+
 namespace {
 
 void RunSoon(const base::Closure& callback) {
-  base::MessageLoop::current()->PostTask(FROM_HERE, callback);
+  if (!callback.is_null())
+    base::MessageLoop::current()->PostTask(FROM_HERE, callback);
 }
 
 template <typename CallbackArray, typename Arg>
@@ -26,6 +30,33 @@ void RunCallbacks(const CallbackArray& callbacks, const Arg& arg) {
     (*i).Run(arg);
 }
 
+// A callback adapter to start a |task| after StartWorker.
+void RunTaskAfterStartWorker(
+    base::WeakPtr<ServiceWorkerVersion> version,
+    const StatusCallback& error_callback,
+    const base::Closure& task,
+    ServiceWorkerStatusCode status) {
+  if (status != SERVICE_WORKER_OK) {
+    if (!error_callback.is_null())
+      error_callback.Run(status);
+    return;
+  }
+  if (version->status() != ServiceWorkerVersion::RUNNING) {
+    // We've tried to start the worker (and it has succeeded), but
+    // it looks it's not running yet.
+    NOTREACHED() << "The worker's not running after successful StartWorker";
+    if (!error_callback.is_null())
+      error_callback.Run(SERVICE_WORKER_ERROR_START_WORKER_FAILED);
+    return;
+  }
+  task.Run();
+}
+
+void RunEmptyMessageCallback(const MessageCallback& callback,
+                             ServiceWorkerStatusCode status) {
+  callback.Run(status, IPC::Message());
+}
+
 }  // namespace
 
 ServiceWorkerVersion::ServiceWorkerVersion(
@@ -34,7 +65,8 @@ ServiceWorkerVersion::ServiceWorkerVersion(
     int64 version_id)
     : version_id_(version_id),
       is_shutdown_(false),
-      registration_(registration) {
+      registration_(registration),
+      weak_factory_(this) {
   if (worker_registry) {
     embedded_worker_ = worker_registry->CreateWorker();
     embedded_worker_->AddObserver(this);
@@ -93,12 +125,55 @@ void ServiceWorkerVersion::StopWorker(const StatusCallback& callback) {
   stop_callbacks_.push_back(callback);
 }
 
+void ServiceWorkerVersion::SendMessage(
+    const IPC::Message& message, const StatusCallback& callback) {
+  DCHECK(!is_shutdown_);
+  DCHECK(embedded_worker_);
+  if (status() != RUNNING) {
+    // Schedule calling this method after starting the worker.
+    StartWorker(base::Bind(&RunTaskAfterStartWorker,
+                           weak_factory_.GetWeakPtr(), callback,
+                           base::Bind(&self::SendMessage,
+                                      weak_factory_.GetWeakPtr(),
+                                      message, callback)));
+    return;
+  }
+
+  ServiceWorkerStatusCode status = embedded_worker_->SendMessage(-1, message);
+  RunSoon(base::Bind(callback, status));
+}
+
+void ServiceWorkerVersion::SendMessageAndRegisterCallback(
+    const IPC::Message& message, const MessageCallback& callback) {
+  DCHECK(!is_shutdown_);
+  DCHECK(embedded_worker_);
+  if (status() != RUNNING) {
+    // Schedule calling this method after starting the worker.
+    StartWorker(base::Bind(&RunTaskAfterStartWorker,
+                           weak_factory_.GetWeakPtr(),
+                           base::Bind(&RunEmptyMessageCallback, callback),
+                           base::Bind(&self::SendMessageAndRegisterCallback,
+                                      weak_factory_.GetWeakPtr(),
+                                      message, callback)));
+    return;
+  }
+
+  int request_id = message_callbacks_.Add(new MessageCallback(callback));
+  ServiceWorkerStatusCode status =
+      embedded_worker_->SendMessage(request_id, message);
+  if (status != SERVICE_WORKER_OK) {
+    message_callbacks_.Remove(request_id);
+    RunSoon(base::Bind(callback, status, IPC::Message()));
+    return;
+  }
+}
+
 bool ServiceWorkerVersion::DispatchFetchEvent(
     const ServiceWorkerFetchRequest& request) {
   if (status() != RUNNING)
     return false;
   return embedded_worker_->SendMessage(
-      ServiceWorkerMsg_FetchEvent(request)) == SERVICE_WORKER_OK;
+      -1, ServiceWorkerMsg_FetchEvent(request)) == SERVICE_WORKER_OK;
 }
 
 void ServiceWorkerVersion::AddProcessToWorker(int process_id) {
@@ -123,14 +198,29 @@ void ServiceWorkerVersion::OnStopped() {
   RunCallbacks(stop_callbacks_, SERVICE_WORKER_OK);
   stop_callbacks_.clear();
 
-  // If there're any callbacks that were waiting start let them know it's
-  // failed.
+  // Let all start callbacks fail.
   RunCallbacks(start_callbacks_, SERVICE_WORKER_ERROR_START_WORKER_FAILED);
   start_callbacks_.clear();
+
+  // Let all message callbacks fail.

[alecflett, 2014/02/04 20:53:24]

lets add a TODO that indicates that in the future we'll likely re-send the
messages. (i.e. requeue them for the next call to StartWorker)

This also makes me wonder if there are certain messages that we shouldn't
re-send? But lets cross that bridge when we get to it.

[kinuko, 2014/02/04 22:50:24]

It totally makes sense to resend install on another restart, but for onfetch
just waiting for another call to StartWorker while letting the controllee
documents wait forever feels wrong.  If situation could differ by message types
we shouldn't probably add a generic mechanism here, but the user of this method
should implement that?

+  IDMap<MessageCallback, IDMapOwnPointer>::iterator iter(&message_callbacks_);
+  while (!iter.IsAtEnd()) {
+    iter.GetCurrentValue()->Run(SERVICE_WORKER_ERROR_ABORT, IPC::Message());
+    iter.Advance();
+  }
+  message_callbacks_.Clear();
 }
 
-void ServiceWorkerVersion::OnMessageReceived(const IPC::Message& message) {
-  NOTREACHED();
+void ServiceWorkerVersion::OnMessageReceived(
+    int request_id, const IPC::Message& message) {
+  MessageCallback* callback = message_callbacks_.Lookup(request_id);
+  if (callback) {
+    callback->Run(SERVICE_WORKER_OK, message);
+    message_callbacks_.Remove(request_id);

[alecflett, 2014/02/04 20:53:24]

This is going to be a little tricky - it's going to be possible to get multiple
responses back from an outgoing message. For fetch, this is an error, I think
it's catch-able on the renderer side:

addEventListener("fetch", function(e) { e.respondWith(fetch(e.request))});
addEventListener("fetch", function(e) { e.respondWith(fetch(e.request))});

(the 2nd event handler should fail when calling e.respondWith())

But at the same time, the install event could be waiting on multiple things:

addEventListener("install", function(e) { e.waitFor(caches.get('foo').add(url1);
});
addEventListener("install", function(e) { e.waitFor(caches.get('foo').add(url2);
});

I'm honestly not quite sure what the right thing to do here is - it seems like
the callback implementation itself needs to be the one to clear the request. We
could add 'this' (i.e. the ServiceWorkerVersion instance) as a parameter to
MessageCallback, and then MessageCallback's implementation could call BACK into
the version to clear it?

That said, I think we should keep what you have but add a TODO to let the
callback clear the request id.

[kinuko, 2014/02/04 22:17:12]

I'm planning to wait ('join') all these promises in the child process side, so
this should just get one response.

Or will there be events we need to wait for in the browser side?  (I'm afriad
it'll make things way more complex)

+    return;
+  }
+  NOTREACHED() << "Got unexpected message: " << request_id
+               << " " << message.type();
 }
 
 }  // namespace content