Implement ServiceWorkerVersion::SendMessage()

As a preparation to implement DispatchInstallEvent(), this implements
following two methods:
- SendMessage()
- SendMessageAndRegisterCallback()

Unlike my previous patches, this supports:
- Sending messages without explicitly starting the worker
- Sending multiple messages concurrently
- Receiving a response for each message

Now each message can be sent with a unique 'request_id', which can be
used to send back a response to a particular message.

BUG=313530
TEST=ServiceWorkerVersionTest.\*
R=alecflett@chromium.org

Review URL: https://codereview.chromium.org/139923005

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

13 files changed, 332 insertions, 45 deletions

diff --git a/content/browser/service_worker/embedded_worker_instance.cc b/content/browser/service_worker/embedded_worker_instance.cc
index 1ffdd77..eaa8b46 100644
--- a/content/browser/service_worker/embedded_worker_instance.cc
+++ b/content/browser/service_worker/embedded_worker_instance.cc
@@ -44,11 +44,13 @@ ServiceWorkerStatusCode EmbeddedWorkerInstance::Stop() {
 }
 
 ServiceWorkerStatusCode EmbeddedWorkerInstance::SendMessage(
+    int request_id,
     const IPC::Message& message) {
   DCHECK(status_ == RUNNING);
   return registry_->Send(process_id_,
                          new EmbeddedWorkerContextMsg_SendMessageToWorker(
-                             thread_id_, embedded_worker_id_, message));
+                             thread_id_, embedded_worker_id_,
+                             request_id, message));
 }
 
 void EmbeddedWorkerInstance::AddProcessReference(int process_id) {
@@ -95,8 +97,10 @@ void EmbeddedWorkerInstance::OnStopped() {
   FOR_EACH_OBSERVER(Observer, observer_list_, OnStopped());
 }
 
-void EmbeddedWorkerInstance::OnMessageReceived(const IPC::Message& message) {
-  FOR_EACH_OBSERVER(Observer, observer_list_, OnMessageReceived(message));
+void EmbeddedWorkerInstance::OnMessageReceived(int request_id,
+                                               const IPC::Message& message) {
+  FOR_EACH_OBSERVER(Observer, observer_list_,
+                    OnMessageReceived(request_id, message));
 }
 
 void EmbeddedWorkerInstance::AddObserver(Observer* observer) {
diff --git a/content/browser/service_worker/embedded_worker_instance.h b/content/browser/service_worker/embedded_worker_instance.h
index e55634c..29d4b49 100644
--- a/content/browser/service_worker/embedded_worker_instance.h
+++ b/content/browser/service_worker/embedded_worker_instance.h
@@ -44,7 +44,8 @@ class CONTENT_EXPORT EmbeddedWorkerInstance {
     virtual ~Observer() {}
     virtual void OnStarted() = 0;
     virtual void OnStopped() = 0;
-    virtual void OnMessageReceived(const IPC::Message& message) = 0;
+    virtual void OnMessageReceived(int request_id,
+                                   const IPC::Message& message) = 0;
   };
 
   ~EmbeddedWorkerInstance();
@@ -62,7 +63,11 @@ class CONTENT_EXPORT EmbeddedWorkerInstance {
 
   // Sends |message| to the embedded worker running in the child process.
   // It is invalid to call this while the worker is not in RUNNING status.
-  ServiceWorkerStatusCode SendMessage(const IPC::Message& message);
+  // |request_id| can be optionally used to establish 2-way request-response
+  // messaging (e.g. the receiver can send back a response using the same
+  // request_id).
+  ServiceWorkerStatusCode SendMessage(
+      int request_id, const IPC::Message& message);
 
   // Add or remove |process_id| to the internal process set where this
   // worker can be started.
@@ -102,7 +107,7 @@ class CONTENT_EXPORT EmbeddedWorkerInstance {
 
   // Called back from Registry when the worker instance sends message
   // to the browser (i.e. EmbeddedWorker observers).
-  void OnMessageReceived(const IPC::Message& message);
+  void OnMessageReceived(int request_id, const IPC::Message& message);
 
   // Chooses a process to start this worker and populate process_id_.
   // Returns false when no process is available.
diff --git a/content/browser/service_worker/embedded_worker_registry.cc b/content/browser/service_worker/embedded_worker_registry.cc
index 73cea86..17f69b2 100644
--- a/content/browser/service_worker/embedded_worker_registry.cc
+++ b/content/browser/service_worker/embedded_worker_registry.cc
@@ -68,7 +68,7 @@ void EmbeddedWorkerRegistry::OnWorkerStopped(
 }
 
 void EmbeddedWorkerRegistry::OnSendMessageToBrowser(
-    int embedded_worker_id, const IPC::Message& message) {
+    int embedded_worker_id, int request_id, const IPC::Message& message) {
   WorkerInstanceMap::iterator found = worker_map_.find(embedded_worker_id);
   if (found == worker_map_.end()) {
     LOG(ERROR) << "Worker " << embedded_worker_id << " not registered";
@@ -77,7 +77,7 @@ void EmbeddedWorkerRegistry::OnSendMessageToBrowser(
   // TODO(kinuko): Filter out unexpected messages here and uncomment below
   // when we actually define messages that are to be sent from child process
   // to the browser via this channel. (We don't have any yet)
-  // found->second->OnMessageReceived(message);
+  found->second->OnMessageReceived(request_id, message);
 }
 
 void EmbeddedWorkerRegistry::AddChildProcessSender(
diff --git a/content/browser/service_worker/embedded_worker_registry.h b/content/browser/service_worker/embedded_worker_registry.h
index f7c40cd..02a5358 100644
--- a/content/browser/service_worker/embedded_worker_registry.h
+++ b/content/browser/service_worker/embedded_worker_registry.h
@@ -54,6 +54,7 @@ class CONTENT_EXPORT EmbeddedWorkerRegistry
   void OnWorkerStarted(int process_id, int thread_id, int embedded_worker_id);
   void OnWorkerStopped(int process_id, int embedded_worker_id);
   void OnSendMessageToBrowser(int embedded_worker_id,
+                              int request_id,
                               const IPC::Message& message);
 
   // Keeps a map from process_id to sender information.
diff --git a/content/browser/service_worker/embedded_worker_test_helper.cc b/content/browser/service_worker/embedded_worker_test_helper.cc
index 267ba4f2..20b663a 100644
--- a/content/browser/service_worker/embedded_worker_test_helper.cc
+++ b/content/browser/service_worker/embedded_worker_test_helper.cc
@@ -79,6 +79,7 @@ void EmbeddedWorkerTestHelper::OnStopWorker(int embedded_worker_id) {
 void EmbeddedWorkerTestHelper::OnSendMessageToWorker(
     int thread_id,
     int embedded_worker_id,
+    int request_id,
     const IPC::Message& message) {
   // Do nothing on the message by default.
 }
@@ -101,8 +102,8 @@ void EmbeddedWorkerTestHelper::SimulateWorkerStopped(
 }
 
 void EmbeddedWorkerTestHelper::SimulateSendMessageToBrowser(
-    int embedded_worker_id, const IPC::Message& message) {
-  registry()->OnSendMessageToBrowser(embedded_worker_id, message);
+    int embedded_worker_id, int request_id, const IPC::Message& message) {
+  registry()->OnSendMessageToBrowser(embedded_worker_id, request_id, message);
 }
 
 void EmbeddedWorkerTestHelper::OnStartWorkerStub(
@@ -134,6 +135,7 @@ void EmbeddedWorkerTestHelper::OnStopWorkerStub(int embedded_worker_id) {
 void EmbeddedWorkerTestHelper::OnSendMessageToWorkerStub(
     int thread_id,
     int embedded_worker_id,
+    int request_id,
     const IPC::Message& message) {
   EmbeddedWorkerInstance* worker = registry()->GetWorker(embedded_worker_id);
   ASSERT_TRUE(worker != NULL);
@@ -144,6 +146,7 @@ void EmbeddedWorkerTestHelper::OnSendMessageToWorkerStub(
                  weak_factory_.GetWeakPtr(),
                  thread_id,
                  embedded_worker_id,
+                 request_id,
                  message));
 }
 
diff --git a/content/browser/service_worker/embedded_worker_test_helper.h b/content/browser/service_worker/embedded_worker_test_helper.h
index 5fafc5e..38675c8 100644
--- a/content/browser/service_worker/embedded_worker_test_helper.h
+++ b/content/browser/service_worker/embedded_worker_test_helper.h
@@ -70,6 +70,7 @@ class EmbeddedWorkerTestHelper : public IPC::Sender,
   virtual void OnStopWorker(int embedded_worker_id);
   virtual void OnSendMessageToWorker(int thread_id,
                                      int embedded_worker_id,
+                                     int request_id,
                                      const IPC::Message& message);
 
   // Call this to simulate sending WorkerStarted, WorkerStopped and
@@ -77,6 +78,7 @@ class EmbeddedWorkerTestHelper : public IPC::Sender,
   void SimulateWorkerStarted(int thread_id, int embedded_worker_id);
   void SimulateWorkerStopped(int embedded_worker_id);
   void SimulateSendMessageToBrowser(int embedded_worker_id,
+                                    int request_id,
                                     const IPC::Message& message);
 
  private:
@@ -86,6 +88,7 @@ class EmbeddedWorkerTestHelper : public IPC::Sender,
   void OnStopWorkerStub(int embedded_worker_id);
   void OnSendMessageToWorkerStub(int thread_id,
                                  int embedded_worker_id,
+                                 int request_id,
                                  const IPC::Message& message);
 
   EmbeddedWorkerRegistry* registry();
diff --git a/content/browser/service_worker/service_worker_browsertest.cc b/content/browser/service_worker/service_worker_browsertest.cc
index 57b108a..50c364c 100644
--- a/content/browser/service_worker/service_worker_browsertest.cc
+++ b/content/browser/service_worker/service_worker_browsertest.cc
@@ -155,7 +155,8 @@ class EmbeddedWorkerBrowserTest : public ServiceWorkerBrowserTest,
     last_worker_status_ = worker_->status();
     BrowserThread::PostTask(BrowserThread::UI, FROM_HERE, done_closure_);
   }
-  virtual void OnMessageReceived(const IPC::Message& message) OVERRIDE {
+  virtual void OnMessageReceived(
+      int request_id, const IPC::Message& message) OVERRIDE {
     NOTREACHED();
   }
 
diff --git a/content/browser/service_worker/service_worker_dispatcher_host.cc b/content/browser/service_worker/service_worker_dispatcher_host.cc
index 6b2d299..51ee49a 100644
--- a/content/browser/service_worker/service_worker_dispatcher_host.cc
+++ b/content/browser/service_worker/service_worker_dispatcher_host.cc
@@ -197,11 +197,12 @@ void ServiceWorkerDispatcherHost::OnWorkerStopped(int embedded_worker_id) {
 
 void ServiceWorkerDispatcherHost::OnSendMessageToBrowser(
     int embedded_worker_id,
+    int request_id,
     const IPC::Message& message) {
   if (!context_)
     return;
   context_->embedded_worker_registry()->OnSendMessageToBrowser(
-      embedded_worker_id, message);
+      embedded_worker_id, request_id, message);
 }
 
 void ServiceWorkerDispatcherHost::UnregistrationComplete(
diff --git a/content/browser/service_worker/service_worker_dispatcher_host.h b/content/browser/service_worker/service_worker_dispatcher_host.h
index c92ee11..471cfe0 100644
--- a/content/browser/service_worker/service_worker_dispatcher_host.h
+++ b/content/browser/service_worker/service_worker_dispatcher_host.h
@@ -50,6 +50,7 @@ class CONTENT_EXPORT ServiceWorkerDispatcherHost : public BrowserMessageFilter {
                        int embedded_worker_id);
   void OnWorkerStopped(int embedded_worker_id);
   void OnSendMessageToBrowser(int embedded_worker_id,
+                              int request_id,
                               const IPC::Message& message);
 
   // Callbacks from ServiceWorkerContextCore
diff --git a/content/browser/service_worker/service_worker_test_utils.h b/content/browser/service_worker/service_worker_test_utils.h
index 177bfdb..fb64f8f 100644
--- a/content/browser/service_worker/service_worker_test_utils.h
+++ b/content/browser/service_worker/service_worker_test_utils.h
@@ -26,6 +26,13 @@ CreateReceiver(BrowserThread::ID run_quit_thread,
   return base::Bind(&ReceiveResult<Arg>, run_quit_thread, quit, out);
 }
 
+template <typename Arg> base::Callback<void(Arg)>
+CreateReceiverOnCurrentThread(Arg* out) {
+  BrowserThread::ID id;
+  BrowserThread::GetCurrentThreadIdentifier(&id);
+  return base::Bind(&ReceiveResult<Arg>, id, base::Closure(), out);
+}
+
 }  // namespace content
 
 #endif  // CONTENT_BROWSER_SERVICE_WORKER_SERVICE_WORKER_TEST_UTILS_H_
diff --git a/content/browser/service_worker/service_worker_version.cc b/content/browser/service_worker/service_worker_version.cc
index 6389d75..8fe826d 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,30 @@ 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.
+  // TODO(kinuko): Consider if we want to add queue+resend mechanism here.
+  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);
+    return;
+  }
+  NOTREACHED() << "Got unexpected message: " << request_id
+               << " " << message.type();
 }
 
 }  // namespace content
diff --git a/content/browser/service_worker/service_worker_version.h b/content/browser/service_worker/service_worker_version.h
index efd5b91..b1daad82 100644
--- a/content/browser/service_worker/service_worker_version.h
+++ b/content/browser/service_worker/service_worker_version.h
@@ -10,6 +10,7 @@
 #include "base/basictypes.h"
 #include "base/callback.h"
 #include "base/gtest_prod_util.h"
+#include "base/id_map.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
 #include "content/browser/service_worker/embedded_worker_instance.h"
@@ -62,6 +63,8 @@ class CONTENT_EXPORT ServiceWorkerVersion
       public EmbeddedWorkerInstance::Observer {
  public:
   typedef base::Callback<void(ServiceWorkerStatusCode)> StatusCallback;
+  typedef base::Callback<void(ServiceWorkerStatusCode,
+                              const IPC::Message& message)> MessageCallback;
 
   enum Status {
     STOPPED = EmbeddedWorkerInstance::STOPPED,
@@ -92,6 +95,24 @@ class CONTENT_EXPORT ServiceWorkerVersion
   // This returns OK (success) if the worker is already stopped.
   void StopWorker(const StatusCallback& callback);
 
+  // Sends an IPC message to the worker.
+  // If the worker is not running this first tries to start it by
+  // calling StartWorker internally.
+  // |callback| can be null if the sender does not need to know if the
+  // message is successfully sent or not.
+  // (If the sender expects the receiver to respond please use
+  // SendMessageAndRegisterCallback instead)
+  void SendMessage(const IPC::Message& message, const StatusCallback& callback);
+
+  // Sends an IPC message to the worker and registers |callback| to
+  // be notified when a response message is received.
+  // The |callback| will be also fired with an error code if the worker
+  // is unexpectedly (being) stopped.
+  // If the worker is not running this first tries to start it by
+  // calling StartWorker internally.
+  void SendMessageAndRegisterCallback(const IPC::Message& message,
+                                      const MessageCallback& callback);
+
   // Sends fetch event to the associated embedded worker.
   // This immediately returns false if the worker is not running
   // or sending a message to the child process fails.
@@ -109,9 +130,11 @@ class CONTENT_EXPORT ServiceWorkerVersion
   // EmbeddedWorkerInstance::Observer overrides:
   virtual void OnStarted() OVERRIDE;
   virtual void OnStopped() OVERRIDE;
-  virtual void OnMessageReceived(const IPC::Message& message) OVERRIDE;
+  virtual void OnMessageReceived(int request_id,
+                                 const IPC::Message& message) OVERRIDE;
 
  private:
+  typedef ServiceWorkerVersion self;
   friend class base::RefCounted<ServiceWorkerVersion>;
 
   virtual ~ServiceWorkerVersion();
@@ -122,9 +145,14 @@ class CONTENT_EXPORT ServiceWorkerVersion
   scoped_refptr<ServiceWorkerRegistration> registration_;
   scoped_ptr<EmbeddedWorkerInstance> embedded_worker_;
 
+  // Pending callbacks.
   std::vector<StatusCallback> start_callbacks_;
   std::vector<StatusCallback> stop_callbacks_;
 
+  IDMap<MessageCallback, IDMapOwnPointer> message_callbacks_;
+
+  base::WeakPtrFactory<ServiceWorkerVersion> weak_factory_;
+
   DISALLOW_COPY_AND_ASSIGN(ServiceWorkerVersion);
 };
 
diff --git a/content/browser/service_worker/service_worker_version_unittest.cc b/content/browser/service_worker/service_worker_version_unittest.cc
index d2984ca..24eed82 100644
--- a/content/browser/service_worker/service_worker_version_unittest.cc
+++ b/content/browser/service_worker/service_worker_version_unittest.cc
@@ -13,8 +13,75 @@
 #include "content/public/test/test_browser_thread_bundle.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
+// IPC messages for testing ---------------------------------------------------
+
+#define IPC_MESSAGE_IMPL
+#include "ipc/ipc_message_macros.h"
+
+#define IPC_MESSAGE_START TestMsgStart
+
+IPC_MESSAGE_CONTROL0(TestMsg_Message);
+IPC_MESSAGE_CONTROL1(TestMsg_Request, int);
+IPC_MESSAGE_CONTROL1(TestMsg_Response, int);
+
+// ---------------------------------------------------------------------------
+
 namespace content {
 
+namespace {
+
+class MessageReceiver : public EmbeddedWorkerTestHelper {
+ public:
+  MessageReceiver(ServiceWorkerContextCore* context)
+      : EmbeddedWorkerTestHelper(context),
+        current_embedded_worker_id_(0),
+        current_request_id_(0) {}
+  virtual ~MessageReceiver() {}
+
+  virtual void OnSendMessageToWorker(int thread_id,
+                                     int embedded_worker_id,
+                                     int request_id,
+                                     const IPC::Message& message) OVERRIDE {
+    current_embedded_worker_id_ = embedded_worker_id;
+    current_request_id_ = request_id;
+    bool handled = true;
+    IPC_BEGIN_MESSAGE_MAP(MessageReceiver, message)
+      IPC_MESSAGE_HANDLER(TestMsg_Message, OnMessage)
+      IPC_MESSAGE_HANDLER(TestMsg_Request, OnRequest)
+      IPC_MESSAGE_UNHANDLED(handled = false)
+    IPC_END_MESSAGE_MAP()
+    ASSERT_TRUE(handled);
+  }
+
+ private:
+  void OnMessage() {
+    // Do nothing.
+  }
+
+  void OnRequest(int value) {
+    // Double the given value and send back the response.
+    SimulateSendMessageToBrowser(current_embedded_worker_id_,
+                                 current_request_id_,
+                                 TestMsg_Response(value * 2));
+  }
+
+  int current_embedded_worker_id_;
+  int current_request_id_;
+  DISALLOW_COPY_AND_ASSIGN(MessageReceiver);
+};
+
+void ReceiveResponse(ServiceWorkerStatusCode* status_out,
+                     int* value_out,
+                     ServiceWorkerStatusCode status,
+                     const IPC::Message& message) {
+  Tuple1<int> param;
+  ASSERT_TRUE(TestMsg_Response::Read(&message, &param));
+  *status_out = status;
+  *value_out = param.a;
+}
+
+}  // namespace
+
 class ServiceWorkerVersionTest : public testing::Test {
  protected:
   ServiceWorkerVersionTest()
@@ -22,16 +89,27 @@ class ServiceWorkerVersionTest : public testing::Test {
 
   virtual void SetUp() OVERRIDE {
     context_.reset(new ServiceWorkerContextCore(base::FilePath(), NULL));
-    helper_.reset(new EmbeddedWorkerTestHelper(context_.get()));
+    helper_.reset(new MessageReceiver(context_.get()));
 
     registration_ = new ServiceWorkerRegistration(
         GURL("http://www.example.com/*"),
         GURL("http://www.example.com/service_worker.js"),
         1L);
+    version_ = new ServiceWorkerVersion(
+        registration_,
+        embedded_worker_registry(),
+        1L);
+
+    // Simulate adding one process to the worker.
+    int embedded_worker_id = version_->embedded_worker()->embedded_worker_id();
+    helper_->SimulateAddProcess(embedded_worker_id, 1);
   }
 
   virtual void TearDown() OVERRIDE {
+    version_->Shutdown();
+    version_ = 0;
     registration_->Shutdown();
+    registration_ = 0;
     helper_.reset();
     context_.reset();
   }
@@ -44,36 +122,24 @@ class ServiceWorkerVersionTest : public testing::Test {
   scoped_ptr<ServiceWorkerContextCore> context_;
   scoped_ptr<EmbeddedWorkerTestHelper> helper_;
   scoped_refptr<ServiceWorkerRegistration> registration_;
+  scoped_refptr<ServiceWorkerVersion> version_;
   DISALLOW_COPY_AND_ASSIGN(ServiceWorkerVersionTest);
 };
 
 TEST_F(ServiceWorkerVersionTest, ConcurrentStartAndStop) {
-  const int64 version_id = 1L;
-
-  scoped_refptr<ServiceWorkerVersion> version = new ServiceWorkerVersion(
-      registration_,
-      embedded_worker_registry(),
-      version_id);
-  int embedded_worker_id = version->embedded_worker()->embedded_worker_id();
-
-  // Simulate adding one process to the worker.
-  helper_->SimulateAddProcess(embedded_worker_id, 1);
-
-  BrowserThread::ID current = BrowserThread::IO;
-
   // Call StartWorker() multiple times.
   ServiceWorkerStatusCode status1 = SERVICE_WORKER_ERROR_FAILED;
   ServiceWorkerStatusCode status2 = SERVICE_WORKER_ERROR_FAILED;
   ServiceWorkerStatusCode status3 = SERVICE_WORKER_ERROR_FAILED;
-  version->StartWorker(CreateReceiver(current, base::Closure(), &status1));
-  version->StartWorker(CreateReceiver(current, base::Closure(), &status2));
+  version_->StartWorker(CreateReceiverOnCurrentThread(&status1));
+  version_->StartWorker(CreateReceiverOnCurrentThread(&status2));
 
-  EXPECT_EQ(ServiceWorkerVersion::STARTING, version->status());
+  EXPECT_EQ(ServiceWorkerVersion::STARTING, version_->status());
   base::RunLoop().RunUntilIdle();
-  EXPECT_EQ(ServiceWorkerVersion::RUNNING, version->status());
+  EXPECT_EQ(ServiceWorkerVersion::RUNNING, version_->status());
 
   // Call StartWorker() after it's started.
-  version->StartWorker(CreateReceiver(current, base::Closure(), &status3));
+  version_->StartWorker(CreateReceiverOnCurrentThread(&status3));
   base::RunLoop().RunUntilIdle();
 
   // All should just succeed.
@@ -85,22 +151,98 @@ TEST_F(ServiceWorkerVersionTest, ConcurrentStartAndStop) {
   status1 = SERVICE_WORKER_ERROR_FAILED;
   status2 = SERVICE_WORKER_ERROR_FAILED;
   status3 = SERVICE_WORKER_ERROR_FAILED;
-  version->StopWorker(CreateReceiver(current, base::Closure(), &status1));
-  version->StopWorker(CreateReceiver(current, base::Closure(), &status2));
+  version_->StopWorker(CreateReceiverOnCurrentThread(&status1));
+  version_->StopWorker(CreateReceiverOnCurrentThread(&status2));
 
   // Also try calling StartWorker while StopWorker is in queue.
-  version->StartWorker(CreateReceiver(current, base::Closure(), &status3));
+  version_->StartWorker(CreateReceiverOnCurrentThread(&status3));
 
-  EXPECT_EQ(ServiceWorkerVersion::STOPPING, version->status());
+  EXPECT_EQ(ServiceWorkerVersion::STOPPING, version_->status());
   base::RunLoop().RunUntilIdle();
-  EXPECT_EQ(ServiceWorkerVersion::STOPPED, version->status());
+  EXPECT_EQ(ServiceWorkerVersion::STOPPED, version_->status());
 
   // All StopWorker should just succeed, while StartWorker fails.
   EXPECT_EQ(SERVICE_WORKER_OK, status1);
   EXPECT_EQ(SERVICE_WORKER_OK, status2);
   EXPECT_EQ(SERVICE_WORKER_ERROR_START_WORKER_FAILED, status3);
+}
 
-  version->Shutdown();
+TEST_F(ServiceWorkerVersionTest, SendMessage) {
+  EXPECT_EQ(ServiceWorkerVersion::STOPPED, version_->status());
+
+  // Send a message without starting the worker.
+  ServiceWorkerStatusCode status = SERVICE_WORKER_ERROR_FAILED;
+  version_->SendMessage(TestMsg_Message(),
+                        CreateReceiverOnCurrentThread(&status));
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(SERVICE_WORKER_OK, status);
+
+  // The worker should be now started.
+  EXPECT_EQ(ServiceWorkerVersion::RUNNING, version_->status());
+
+  // Stop the worker, and then send the message immediately.
+  ServiceWorkerStatusCode msg_status = SERVICE_WORKER_ERROR_FAILED;
+  ServiceWorkerStatusCode stop_status = SERVICE_WORKER_ERROR_FAILED;
+  version_->StopWorker(CreateReceiverOnCurrentThread(&stop_status));
+  version_->SendMessage(TestMsg_Message(),
+                       CreateReceiverOnCurrentThread(&msg_status));
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(SERVICE_WORKER_OK, stop_status);
+
+  // SendMessage should return START_WORKER_FAILED error since it tried to
+  // start a worker while it was stopping.
+  EXPECT_EQ(SERVICE_WORKER_ERROR_START_WORKER_FAILED, msg_status);
+}
+
+TEST_F(ServiceWorkerVersionTest, ReSendMessageAfterStop) {
+  EXPECT_EQ(ServiceWorkerVersion::STOPPED, version_->status());
+
+  // Start the worker.
+  ServiceWorkerStatusCode start_status = SERVICE_WORKER_ERROR_FAILED;
+  version_->StartWorker(CreateReceiverOnCurrentThread(&start_status));
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(SERVICE_WORKER_OK, start_status);
+  EXPECT_EQ(ServiceWorkerVersion::RUNNING, version_->status());
+
+  // Stop the worker, and then send the message immediately.
+  ServiceWorkerStatusCode msg_status = SERVICE_WORKER_ERROR_FAILED;
+  ServiceWorkerStatusCode stop_status = SERVICE_WORKER_ERROR_FAILED;
+  version_->StopWorker(CreateReceiverOnCurrentThread(&stop_status));
+  version_->SendMessage(TestMsg_Message(),
+                       CreateReceiverOnCurrentThread(&msg_status));
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(SERVICE_WORKER_OK, stop_status);
+
+  // SendMessage should return START_WORKER_FAILED error since it tried to
+  // start a worker while it was stopping.
+  EXPECT_EQ(SERVICE_WORKER_ERROR_START_WORKER_FAILED, msg_status);
+
+  // Resend the message, which should succeed and restart the worker.
+  version_->SendMessage(TestMsg_Message(),
+                       CreateReceiverOnCurrentThread(&msg_status));
+  base::RunLoop().RunUntilIdle();
+  EXPECT_EQ(SERVICE_WORKER_OK, msg_status);
+  EXPECT_EQ(ServiceWorkerVersion::RUNNING, version_->status());
+}
+
+TEST_F(ServiceWorkerVersionTest, SendMessageAndRegisterCallback) {
+  // Send multiple messages and verify responses.
+  ServiceWorkerStatusCode status1 = SERVICE_WORKER_ERROR_FAILED;
+  ServiceWorkerStatusCode status2 = SERVICE_WORKER_ERROR_FAILED;
+  int value1 = -1, value2 = -1;
+
+  version_->SendMessageAndRegisterCallback(
+      TestMsg_Request(111),
+      base::Bind(&ReceiveResponse, &status1, &value1));
+  version_->SendMessageAndRegisterCallback(
+      TestMsg_Request(333),
+      base::Bind(&ReceiveResponse, &status2, &value2));
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_EQ(SERVICE_WORKER_OK, status1);
+  EXPECT_EQ(SERVICE_WORKER_OK, status2);
+  EXPECT_EQ(111 * 2, value1);
+  EXPECT_EQ(333 * 2, value2);
 }
 
 }  // namespace content