Adds Channel-associated interface support on ChannelProxy's thread

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>
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/ref_counted.h"
 #include "base/profiler/scoped_tracker.h"
 #include "base/single_thread_task_runner.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "build/build_config.h"
 #include "ipc/ipc_channel_factory.h"
 #include "ipc/ipc_listener.h"
 #include "ipc/ipc_logging.h"
 #include "ipc/ipc_message_macros.h"
 #include "ipc/message_filter.h"
 #include "ipc/message_filter_router.h"
 
 namespace IPC {
 
+namespace {
+
+void BindAssociatedInterfaceOnTaskRunner(
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    const ChannelProxy::GenericAssociatedInterfaceFactory& factory,
+    mojo::ScopedInterfaceEndpointHandle handle) {
+  task_runner->PostTask(FROM_HERE, base::Bind(factory, base::Passed(&handle)));
+}
+
+}  // namespace
+
 //------------------------------------------------------------------------------
 
 ChannelProxy::Context::Context(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
     : listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
       listener_(listener),
       ipc_task_runner_(ipc_task_runner),
       channel_connected_called_(false),
       channel_send_thread_safe_(false),
@@ skipping 100 matching lines @@
   // will be released when we are closed.
   AddRef();
 
   if (!channel_->Connect()) {
     OnChannelError();
     return;
   }
 
   for (size_t i = 0; i < filters_.size(); ++i)
     filters_[i]->OnFilterAdded(channel_.get());
+
+  Channel::AssociatedInterfaceSupport* support =
+      channel_->GetAssociatedInterfaceSupport();
+  if (support) {
+    support->SetProxyTaskRunner(listener_task_runner_);
+    for (auto& entry : io_thread_interfaces_)
+      support->AddGenericAssociatedInterface(entry.first, entry.second);
+    for (auto& entry : proxy_thread_interfaces_) {
+      support->AddGenericAssociatedInterface(
+          entry.first, base::Bind(&BindAssociatedInterfaceOnTaskRunner,
+                                  listener_task_runner_, entry.second));
+    }
+  } else {
+    // Sanity check to ensure nobody's expecting to use associated interfaces on
+    // a Channel that doesn't support them.
+    DCHECK(io_thread_interfaces_.empty() && proxy_thread_interfaces_.empty());
+  }
 }
 
 // Called on the IPC::Channel thread
 void ChannelProxy::Context::OnChannelClosed() {
   // TODO(pkasting): Remove ScopedTracker below once crbug.com/477117 is fixed.
   tracked_objects::ScopedTracker tracking_profile(
       FROM_HERE_WITH_EXPLICIT_FUNCTION(
           "477117 ChannelProxy::Context::OnChannelClosed"));
   // It's okay for IPC::ChannelProxy::Close to be called more than once, which
   // would result in this branch being taken.
@@ skipping 129 matching lines @@
   if (logger->Enabled())
     logger->OnPostDispatchMessage(message, channel_id_);
 #endif
 }
 
 // Called on the listener's thread
 void ChannelProxy::Context::OnDispatchConnected() {
   if (channel_connected_called_)
     return;
 
+  Channel::AssociatedInterfaceSupport* associated_interface_support =
+      channel_->GetAssociatedInterfaceSupport();
+  if (associated_interface_support) {
+    channel_associated_group_.reset(new mojo::AssociatedGroup(
+        *associated_interface_support->GetAssociatedGroup()));
+  }
+
   channel_connected_called_ = true;
   if (listener_)
     listener_->OnChannelConnected(peer_pid_);
 }
 
 // Called on the listener's thread
 void ChannelProxy::Context::OnDispatchError() {
   if (listener_)
     listener_->OnChannelError();
 }
@@ skipping 25 matching lines @@
 
   ipc_task_runner()->PostTask(
       FROM_HERE, base::Bind(&ChannelProxy::Context::OnSendMessage, this,
                             base::Passed(base::WrapUnique(message))));
 }
 
 bool ChannelProxy::Context::IsChannelSendThreadSafe() const {
   return channel_send_thread_safe_;
 }
 
+// Called on the IPC::Channel thread
+void ChannelProxy::Context::GetRemoteAssociatedInterface(
+    const std::string& name,
+    mojo::ScopedInterfaceEndpointHandle handle) {
+  if (!channel_)
+    return;
+  Channel::AssociatedInterfaceSupport* associated_interface_support =
+      channel_->GetAssociatedInterfaceSupport();
+  DCHECK(associated_interface_support);
+  associated_interface_support->GetGenericRemoteAssociatedInterface(
+      name, std::move(handle));
+}
+
 //-----------------------------------------------------------------------------
 
 // static
 std::unique_ptr<ChannelProxy> ChannelProxy::Create(
     const IPC::ChannelHandle& channel_handle,
     Channel::Mode mode,
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
   std::unique_ptr<ChannelProxy> channel(
       new ChannelProxy(listener, ipc_task_runner));
@@ skipping 118 matching lines @@
 }
 
 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
   DCHECK(CalledOnValidThread());
 
   context_->ipc_task_runner()->PostTask(
       FROM_HERE, base::Bind(&Context::OnRemoveFilter, context_.get(),
                             base::RetainedRef(filter)));
 }
 
+void ChannelProxy::AddGenericAssociatedInterfaceForIOThread(
+    const std::string& name,
+    const GenericAssociatedInterfaceFactory& factory) {
+  DCHECK(CalledOnValidThread());
+  DCHECK(!did_init_);
+  context_->io_thread_interfaces_.insert({ name, factory });
+}
+
+void ChannelProxy::AddGenericAssociatedInterface(
+    const std::string& name,
+    const GenericAssociatedInterfaceFactory& factory) {
+  DCHECK(CalledOnValidThread());
+  DCHECK(!did_init_);
+  context_->proxy_thread_interfaces_.insert({ name, factory });
+}
+
+mojo::AssociatedGroup* ChannelProxy::GetAssociatedGroup() {
+  return context_->channel_associated_group_.get();
+}
+
+void ChannelProxy::GetGenericRemoteAssociatedInterface(
+    const std::string& name,
+    mojo::ScopedInterfaceEndpointHandle handle) {
+  context_->ipc_task_runner()->PostTask(

[yzshen1, 2016/07/13 23:25:09]

Posting a task means the associated interface ptr cannot be used immediately
because we cannot start sending messages until |handle| gets into the pipe,
right?

[Ken Rockot(use gerrit already), 2016/07/14 00:39:09]

Discussed offline, the endpoint's SendMessage ultimately posts to the same task
runner, so this is a non issue - sent messages will only be written to the pipe
after this task runs.

+      FROM_HERE, base::Bind(&Context::GetRemoteAssociatedInterface,
+                            context_.get(), name, base::Passed(&handle)));
+}
+
 void ChannelProxy::ClearIPCTaskRunner() {
   DCHECK(CalledOnValidThread());
 
   context()->ClearIPCTaskRunner();
 }
 
 base::ProcessId ChannelProxy::GetPeerPID() const {
   return context_->peer_pid_;
 }
 
@@ skipping 23 matching lines @@
   return channel->TakeClientFileDescriptor();
 }
 #endif
 
 void ChannelProxy::OnChannelInit() {
 }
 
 //-----------------------------------------------------------------------------
 
 }  // namespace IPC