Support mojo connections between unrelated peer processes.

mojo/edk/system/node_controller.cc

 // Copyright 2016 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 "mojo/edk/system/node_controller.h"
 
 #include <algorithm>
 #include <limits>
 
 #include "base/bind.h"
@@ skipping 236 matching lines @@
   }
 #endif
 
   io_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&NodeController::ConnectToParentOnIOThread,
                  base::Unretained(this),
                  base::Passed(&platform_handle)));
 }
 
+void NodeController::ConnectToPeer(ScopedPlatformHandle handle,
+                                   const ports::PortRef& port) {
+  ports::NodeName node_name;
+  GenerateRandomName(&node_name);
+  io_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&NodeController::ConnectToPeerOnIOThread,
+                            base::Unretained(this), base::Passed(&handle),
+                            node_name, port));
+}
+
 void NodeController::SetPortObserver(
     const ports::PortRef& port,
     const scoped_refptr<PortObserver>& observer) {
   node_->SetUserData(port, observer);
 }
 
 void NodeController::ClosePort(const ports::PortRef& port) {
   SetPortObserver(port, nullptr);
   int rv = node_->ClosePort(port);
   DCHECK_EQ(rv, ports::OK) << " Failed to close port: " << port.name();
@@ skipping 160 matching lines @@
     // Prevent the parent pipe handle from being closed on shutdown. Pipe
     // closure is used by the parent to detect the child process has exited.
     // Relying on message pipes to be closed is not enough because the parent
     // may see the message pipe closure before the child is dead, causing the
     // child process to be unexpectedly SIGKILL'd.
     bootstrap_parent_channel_->LeakHandleOnShutdown();
   }
   bootstrap_parent_channel_->Start();
 }
 
+void NodeController::ConnectToPeerOnIOThread(ScopedPlatformHandle handle,
+                                             ports::NodeName token,
+                                             ports::PortRef port) {
+  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+
+  scoped_refptr<NodeChannel> channel =
+      NodeChannel::Create(this, std::move(handle), io_task_runner_, {});
+  pending_peers_.insert({token, {channel, port}});
+
+  channel->SetRemoteNodeName(token);
+  channel->Start();
+
+  channel->AcceptPeer(name_, token, port.name());
+}
+
 scoped_refptr<NodeChannel> NodeController::GetPeerChannel(
     const ports::NodeName& name) {
   base::AutoLock lock(peers_lock_);
   auto it = peers_.find(name);
   if (it == peers_.end())
     return nullptr;
   return it->second;
 }
 
 scoped_refptr<NodeChannel> NodeController::GetParentChannel() {
@@ skipping 267 matching lines @@
 
   {
     base::AutoLock lock(peers_lock_);
     for (const auto& peer : peers_)
       all_peers.push_back(peer.second);
     for (const auto& peer : pending_children_)
       all_peers.push_back(peer.second);
     peers_.clear();
     pending_children_.clear();
     pending_peer_messages_.clear();
+    pending_peers_.clear();
   }
 
   for (const auto& peer : all_peers)
     peer->ShutDown();
 
   if (destroy_on_io_thread_shutdown_)
     delete this;
 }
 
 void NodeController::GenerateRandomPortName(ports::PortName* port_name) {
@@ skipping 488 matching lines @@
   if (GetPeerChannel(from_node) != GetBrokerChannel()) {
     LOG(ERROR) << "Refusing relayed message from non-broker node.";
     DropPeer(from_node, nullptr);
     return;
   }
 
   OnPortsMessage(source_node, std::move(message));
 }
 #endif
 
+void NodeController::OnAcceptPeer(const ports::NodeName& from_node,
+                                  const ports::NodeName& token,
+                                  const ports::NodeName& peer_name,
+                                  const ports::PortName& port_name) {
+  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+
+  auto it = pending_peers_.find(from_node);
+  if (it == pending_peers_.end()) {
+    DLOG(ERROR) << "Received unexpected AcceptPeer message from " << from_node;
+    DropPeer(from_node, nullptr);
+    return;
+  }
+
+  scoped_refptr<NodeChannel> channel = it->second.first;
+  ports::PortRef local_port = it->second.second;
+  pending_peers_.erase(it);
+  DCHECK(channel);
+
+  DVLOG(1) << "Node " << name_ << " accepted peer " << peer_name;
+
+  AddPeer(peer_name, channel, false /* start_channel */);
+
+  // We need to choose one side to initiate the port merge. It doesn't matter
+  // who does it as long as they don't both try. Simple solution: pick the one
+  // with the "smaller" port name.
+  if (local_port.name() < port_name) {
+    node()->MergePorts(local_port, peer_name, port_name);
+  }
+}
+
 void NodeController::OnChannelError(const ports::NodeName& from_node,
                                     NodeChannel* channel) {
   if (io_task_runner_->RunsTasksOnCurrentThread()) {
     DropPeer(from_node, channel);
     // DropPeer may have caused local port closures, so be sure to process any
     // pending local messages.
     AcceptIncomingMessages();
   } else {
     io_task_runner_->PostTask(
         FROM_HERE,
@@ skipping 40 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo