Mojo Channel: Fix deferred proxy dispatch; support paused channels

ipc/ipc_channel_proxy.cc

 // Copyright (c) 2012 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_proxy.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <utility>
@@ skipping 97 matching lines @@
 #ifdef IPC_MESSAGE_LOG_ENABLED
     if (logger->Enabled())
       logger->OnPostDispatchMessage(message, channel_id_);
 #endif
     return true;
   }
   return false;
 }
 
 // Called on the IPC::Channel thread
+void ChannelProxy::Context::UnpauseChannel(bool flush) {
+  DCHECK(channel_);
+  channel_->Unpause(flush);
+}
+
+// Called on the IPC::Channel thread
+void ChannelProxy::Context::FlushChannel() {
+  DCHECK(channel_);
+  channel_->Flush();
+}
+
+// Called on the IPC::Channel thread
 bool ChannelProxy::Context::OnMessageReceived(const Message& message) {
   // First give a chance to the filters to process this message.
   if (!TryFilters(message))
     OnMessageReceivedNoFilter(message);
   return true;
 }
 
 // Called on the IPC::Channel thread
 bool ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) {
   listener_task_runner_->PostTask(
@@ skipping 24 matching lines @@
 void ChannelProxy::Context::OnChannelError() {
   for (size_t i = 0; i < filters_.size(); ++i)
     filters_[i]->OnChannelError();
 
   // See above comment about using listener_task_runner_ here.
   listener_task_runner_->PostTask(
       FROM_HERE, base::Bind(&Context::OnDispatchError, this));
 }
 
 // Called on the IPC::Channel thread
-void ChannelProxy::Context::OnChannelOpened() {
+void ChannelProxy::Context::OnChannelOpened(bool pause) {
   DCHECK(channel_ != NULL);
 
   // Assume a reference to ourselves on behalf of this thread.  This reference
   // will be released when we are closed.
   AddRef();
 
-  if (!channel_->Connect()) {
+  bool success = pause ? channel_->ConnectPaused() : channel_->Connect();
+  if (!success) {
     OnChannelError();
     return;
   }
 
   for (size_t i = 0; i < filters_.size(); ++i)
     filters_[i]->OnFilterAdded(channel_.get());
 }
 
 // Called on the IPC::Channel thread
 void ChannelProxy::Context::OnChannelClosed() {
@@ skipping 275 matching lines @@
 }
 
 ChannelProxy::~ChannelProxy() {
   DCHECK(CalledOnValidThread());
 
   Close();
 }
 
 void ChannelProxy::Init(const IPC::ChannelHandle& channel_handle,
                         Channel::Mode mode,
-                        bool create_pipe_now) {
+                        bool create_pipe_now,
+                        bool create_paused) {
 #if defined(OS_POSIX)
   // When we are creating a server on POSIX, we need its file descriptor
   // to be created immediately so that it can be accessed and passed
   // to other processes. Forcing it to be created immediately avoids
   // race conditions that may otherwise arise.
   if (mode & Channel::MODE_SERVER_FLAG) {
     create_pipe_now = true;
   }
 #endif  // defined(OS_POSIX)
   Init(
       ChannelFactory::Create(channel_handle, mode, context_->ipc_task_runner()),
-      create_pipe_now);
+      create_pipe_now, create_paused);
 }
 
 void ChannelProxy::Init(std::unique_ptr<ChannelFactory> factory,
-                        bool create_pipe_now) {
+                        bool create_pipe_now,
+                        bool create_paused) {
   DCHECK(CalledOnValidThread());
   DCHECK(!did_init_);
 
   if (create_pipe_now) {
     // Create the channel immediately.  This effectively sets up the
     // low-level pipe so that the client can connect.  Without creating
     // the pipe immediately, it is possible for a listener to attempt
     // to connect and get an error since the pipe doesn't exist yet.
     context_->CreateChannel(std::move(factory));
   } else {
     context_->ipc_task_runner()->PostTask(
         FROM_HERE, base::Bind(&Context::CreateChannel, context_,
                               base::Passed(&factory)));
   }
 
   // complete initialization on the background thread
   context_->ipc_task_runner()->PostTask(
-      FROM_HERE, base::Bind(&Context::OnChannelOpened, context_));
+      FROM_HERE,
+      base::Bind(&Context::OnChannelOpened, context_, create_paused));
 
   did_init_ = true;
   OnChannelInit();
 }
 
+void ChannelProxy::Unpause(bool flush) {
+  context_->ipc_task_runner()->PostTask(
+      FROM_HERE, base::Bind(&Context::UnpauseChannel, context_, flush));
+}
+
+void ChannelProxy::Flush() {
+  context_->ipc_task_runner()->PostTask(
+      FROM_HERE, base::Bind(&Context::FlushChannel, context_));
+}
+
 void ChannelProxy::Close() {
   DCHECK(CalledOnValidThread());
 
   // Clear the backpointer to the listener so that any pending calls to
   // Context::OnDispatchMessage or OnDispatchError will be ignored.  It is
   // possible that the channel could be closed while it is receiving messages!
   context_->Clear();
 
   if (context_->ipc_task_runner()) {
     context_->ipc_task_runner()->PostTask(
@@ skipping 99 matching lines @@
   return channel->TakeClientFileDescriptor();
 }
 #endif
 
 void ChannelProxy::OnChannelInit() {
 }
 
 //-----------------------------------------------------------------------------
 
 }  // namespace IPC