Mojo Channel: Fix deferred proxy dispatch; support paused channels

ipc/ipc_channel_mojo.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_channel_mojo.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <memory>
@@ skipping 264 matching lines @@
   // ChannelMojo from a different thread.
   bootstrap_ =
       MojoBootstrap::Create(std::move(handle), mode, this, ipc_task_runner);
 }
 
 ChannelMojo::~ChannelMojo() {
   Close();
 }
 
 bool ChannelMojo::Connect() {
+  if (!ConnectPaused())
+    return false;
+
+  Unpause(true);
+  return true;
+}
+
+bool ChannelMojo::ConnectPaused() {
   WillConnect();
 
   DCHECK(!task_runner_);
   task_runner_ = base::ThreadTaskRunnerHandle::Get();
   DCHECK(!message_reader_);
 
   bootstrap_->Connect();
   return true;
 }
 
+void ChannelMojo::Unpause(bool flush) {
+  bootstrap_->Start();
+
+  // Ensure that no matter what messages have been queued so far, the first
+  // message we send is always the peer PID.
+  DCHECK(message_reader_);
+  message_reader_->sender()->SetPeerPid(GetSelfPID());
+
+  if (flush)
+    Flush();
+}
+
+void ChannelMojo::Flush() {
+  bootstrap_->Flush();
+}
+
 void ChannelMojo::Close() {
   // NOTE: The MessagePipeReader's destructor may re-enter this function. Use
   // caution when changing this method.
   std::unique_ptr<internal::MessagePipeReader> reader =
       std::move(message_reader_);
   reader.reset();
 
   base::AutoLock lock(associated_interface_lock_);
   associated_interfaces_.clear();
 }
 
 // MojoBootstrap::Delegate implementation
 void ChannelMojo::OnPipesAvailable(mojom::ChannelAssociatedPtr sender,
                                    mojom::ChannelAssociatedRequest receiver) {
-  sender->SetPeerPid(GetSelfPID());
   message_reader_.reset(new internal::MessagePipeReader(
       pipe_, std::move(sender), std::move(receiver), this));
 }
 
 void ChannelMojo::OnPipeError() {
   DCHECK(task_runner_);
   if (task_runner_->RunsTasksOnCurrentThread()) {
     listener_->OnChannelError();
   } else {
     task_runner_->PostTask(
@@ skipping 161 matching lines @@
 }
 
 void ChannelMojo::GetGenericRemoteAssociatedInterface(
     const std::string& name,
     mojo::ScopedInterfaceEndpointHandle handle) {
   if (message_reader_)
     message_reader_->GetRemoteInterface(name, std::move(handle));
 }
 
 }  // namespace IPC