Mojo Channel: Fix deferred proxy dispatch; support paused channels

ipc/ipc_mojo_bootstrap.cc

 // Copyright 2014 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 "ipc/ipc_mojo_bootstrap.h"
 
 #include <stdint.h>
 
 #include <map>
 #include <memory>
@@ skipping 52 matching lines @@
     DCHECK(thread_checker_.CalledOnValidThread());
     DCHECK(task_runner_->BelongsToCurrentThread());
 
     connector_.reset(new mojo::Connector(
         std::move(handle), mojo::Connector::SINGLE_THREADED_SEND,
         task_runner_));
     connector_->set_incoming_receiver(&filters_);
     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)
       SendMessage(&message);
   }
 
   void CreateChannelEndpoints(mojom::ChannelAssociatedPtr* sender,
                               mojom::ChannelAssociatedRequest* receiver) {
     mojo::InterfaceId sender_id, receiver_id;
     if (set_interface_id_namespace_bit_) {
@@ skipping 375 matching lines @@
       DCHECK(endpoint->closed());
       MarkPeerClosedAndMaybeRemove(endpoint);
     }
 
     DCHECK(endpoints_.empty());
   }
 
   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;
       }
       return connector_->Accept(message);
     } else {
       // We always post tasks to the master endpoint thread when called from the
       // proxy thread in order to simulate IPC::ChannelProxy::Send behavior.
       DCHECK(proxy_task_runner_->BelongsToCurrentThread());
       task_runner_->PostTask(
           FROM_HERE,
@@ skipping 103 matching lines @@
   bool Accept(mojo::Message* message) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
     if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message))
       return control_message_handler_.Accept(message);
 
     mojo::InterfaceId id = message->interface_id();
     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
       // |proxy_task_runner_| runs tasks, since that's the only valid scenario.
       //
       // If the client is not yet bound, it must be bound by the time this task
       // runs or else it's programmer error.
       DCHECK(proxy_task_runner_);
 
       if (message->has_flag(mojo::Message::kFlagIsSync)) {
@@ skipping 26 matching lines @@
     return client->HandleIncomingMessage(message);
   }
 
   void AcceptOnProxyThread(mojo::Message message) {
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
     mojo::InterfaceId id = message.interface_id();
     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;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
 
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
 
     // Sync messages should never make their way to this method.
     DCHECK(!message.has_flag(mojo::Message::kFlagIsSync));
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
       result = client->HandleIncomingMessage(&message);
     }
 
     if (!result)
       RaiseError();
   }
 
   void AcceptSyncMessage(mojo::InterfaceId interface_id, uint32_t message_id) {
     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;
 
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
     mojo::Message message = endpoint->PopSyncMessage(message_id);
 
     // The message must have already been dequeued by the endpoint waking up
     // from a sync wait. Nothing to do.
     if (message.IsNull())
       return;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
       result = client->HandleIncomingMessage(&message);
     }
 
     if (!result)
       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);
       return nullptr;
     }
 
     if (endpoint->closed())
       return nullptr;
 
     return endpoint;
@@ skipping 32 matching lines @@
     return true;
   }
 
   // Checked in places which must be run on the master endpoint's thread.
   base::ThreadChecker thread_checker_;
 
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
   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_;
   ControlMessageProxyThunk control_message_proxy_thunk_;
   mojo::PipeControlMessageProxy control_message_proxy_;
 
   // Outgoing messages that were sent before this controller was bound to a
   // real message pipe.
   std::vector<mojo::Message> outgoing_messages_;
 
@@ skipping 31 matching lines @@
   void Connect() override {
     controller_->Bind(std::move(handle_));
 
     IPC::mojom::ChannelAssociatedPtr sender;
     IPC::mojom::ChannelAssociatedRequest receiver;
     controller_->CreateChannelEndpoints(&sender, &receiver);
 
     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();
   }
 
   scoped_refptr<ChannelAssociatedGroupController> controller_;
 
   mojo::ScopedMessagePipeHandle handle_;
   Delegate* delegate_;
   std::unique_ptr<mojo::AssociatedGroup> associated_group_;
 
   DISALLOW_COPY_AND_ASSIGN(MojoBootstrapImpl);
 };
 
 }  // namespace
 
 // static
 std::unique_ptr<MojoBootstrap> MojoBootstrap::Create(
     mojo::ScopedMessagePipeHandle handle,
     Channel::Mode mode,
     Delegate* delegate,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
   return base::MakeUnique<MojoBootstrapImpl>(
       std::move(handle), delegate,
       new ChannelAssociatedGroupController(mode == Channel::MODE_SERVER,
                                            ipc_task_runner));
 }
 
 }  // namespace IPC