Mojo Channel: Fix deferred proxy dispatch; support paused channels

ipc/ipc_mojo_bootstrap.cc

Index: ipc/ipc_mojo_bootstrap.cc
diff --git a/ipc/ipc_mojo_bootstrap.cc b/ipc/ipc_mojo_bootstrap.cc
index 24c14644a4bd138ed04ace3f5940d95596367219..affcf18b9df2d6cb8bbf2a5b1d8f5b5059127d7f 100644
--- a/ipc/ipc_mojo_bootstrap.cc
+++ b/ipc/ipc_mojo_bootstrap.cc
@@ -70,7 +70,14 @@ class ChannelAssociatedGroupController
     connector_->set_connection_error_handler(
         base::Bind(&ChannelAssociatedGroupController::OnPipeError,
                    base::Unretained(this)));
+  }
+
+  void Start() {
+    DCHECK(!started_);
+    started_ = true;
+  }
 
+  void FlushOutgoingMessages() {
     std::vector<mojo::Message> outgoing_messages;
     std::swap(outgoing_messages, outgoing_messages_);
     for (auto& message : outgoing_messages)
@@ -466,8 +473,9 @@ class ChannelAssociatedGroupController
   bool SendMessage(mojo::Message* message) {
     if (task_runner_->BelongsToCurrentThread()) {
       DCHECK(thread_checker_.CalledOnValidThread());
-      if (!connector_) {
-        // Pipe may not be bound yet, so we queue the message.
+      if (!connector_ || !started_) {
+        // Pipe may not be bound yet or the channel may still be paused, so we
+        // queue the message.
         outgoing_messages_.emplace_back(std::move(*message));
         return true;
       }
@@ -591,11 +599,10 @@ class ChannelAssociatedGroupController
     DCHECK(mojo::IsValidInterfaceId(id));
 
     base::AutoLock locker(lock_);
-    Endpoint* endpoint = GetEndpointForDispatch(id);
-    if (!endpoint)
-      return true;
-
-    mojo::InterfaceEndpointClient* client = endpoint->client();
+    Endpoint* endpoint =
+        GetEndpointForDispatch(id, false /* close_on_insert */);
+    mojo::InterfaceEndpointClient* client =
+        endpoint ? endpoint->client() : nullptr;
     if (!client || !endpoint->task_runner()->BelongsToCurrentThread()) {
       // No client has been bound yet or the client runs tasks on another
       // thread. We assume the other thread must always be the one on which
@@ -642,7 +649,7 @@ class ChannelAssociatedGroupController
     DCHECK(mojo::IsValidInterfaceId(id) && !mojo::IsMasterInterfaceId(id));
 
     base::AutoLock locker(lock_);
-    Endpoint* endpoint = GetEndpointForDispatch(id);
+    Endpoint* endpoint = GetEndpointForDispatch(id, true /* close_on_insert */);
     if (!endpoint)
       return;
 
@@ -669,7 +676,8 @@ class ChannelAssociatedGroupController
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
     base::AutoLock locker(lock_);
-    Endpoint* endpoint = GetEndpointForDispatch(interface_id);
+    Endpoint* endpoint =
+        GetEndpointForDispatch(interface_id, true /* close_on_insert */);
     if (!endpoint)
       return;
 
@@ -695,11 +703,11 @@ class ChannelAssociatedGroupController
       RaiseError();
   }
 
-  Endpoint* GetEndpointForDispatch(mojo::InterfaceId id) {
+  Endpoint* GetEndpointForDispatch(mojo::InterfaceId id, bool close_on_insert) {
     lock_.AssertAcquired();
     bool inserted = false;
     Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
-    if (inserted) {
+    if (inserted && close_on_insert) {
       MarkClosedAndMaybeRemove(endpoint);
       if (!mojo::IsMasterInterfaceId(id))
         control_message_proxy_.NotifyPeerEndpointClosed(id);
@@ -752,6 +760,7 @@ class ChannelAssociatedGroupController
 
   scoped_refptr<base::SingleThreadTaskRunner> proxy_task_runner_;
   const bool set_interface_id_namespace_bit_;
+  bool started_ = false;
   std::unique_ptr<mojo::Connector> connector_;
   mojo::FilterChain filters_;
   mojo::PipeControlMessageHandler control_message_handler_;
@@ -803,6 +812,14 @@ class MojoBootstrapImpl : public MojoBootstrap {
     delegate_->OnPipesAvailable(std::move(sender), std::move(receiver));
   }
 
+  void Start() override {
+    controller_->Start();
+  }
+
+  void Flush() override {
+    controller_->FlushOutgoingMessages();
+  }
+
   mojo::AssociatedGroup* GetAssociatedGroup() override {
     return associated_group_.get();
   }