[mojo-edk] Broadcast surprise port disruptions

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 544 matching lines @@
     // (synchronously) in response to Node's ClosePort, SendMessage, or
     // AcceptMessage, we flush the queue after calling any of those methods.
     base::AutoLock lock(messages_lock_);
     incoming_messages_.emplace(std::move(message));
     incoming_messages_flag_.Set(true);
   } else {
     SendPeerMessage(node, std::move(message));
   }
 }
 
+void NodeController::BroadcastMessage(ports::ScopedMessage message) {
+  Channel::MessagePtr channel_message =
+      static_cast<PortsMessage*>(message.get())->TakeChannelMessage();
+  scoped_refptr<NodeChannel> broker = GetBrokerChannel();
+  if (broker)
+    broker->Broadcast(std::move(channel_message));
+  else
+    OnBroadcast(std::move(channel_message));
+}
+
 void NodeController::PortStatusChanged(const ports::PortRef& port) {
   scoped_refptr<ports::UserData> user_data;
   node_->GetUserData(port, &user_data);
 
   PortObserver* observer = static_cast<PortObserver*>(user_data.get());
   if (observer) {
     observer->OnPortStatusChanged();
   } else {
     DVLOG(2) << "Ignoring status change for " << port.name() << " because it "
              << "doesn't have an observer.";
@@ skipping 302 matching lines @@
     return;
   }
 
   scoped_refptr<NodeChannel> channel =
       NodeChannel::Create(this, std::move(channel_handle), io_task_runner_);
 
   DVLOG(1) << "Adding new peer " << name << " via parent introduction.";
   AddPeer(name, channel, true /* start_channel */);
 }
 
+void NodeController::OnBroadcast(Channel::MessagePtr message) {
+  {
+    DCHECK(!message->has_handles());
+    base::AutoLock lock(peers_lock_);
+    for (auto iter : peers_) {

[Anand Mistry (off Chromium), 2016/05/16 04:03:57]

auto&

[Ken Rockot(use gerrit already), 2016/05/16 04:42:36]

Done

+      Channel::MessagePtr peer_message(
+          new Channel::Message(message->payload_size(), 0));
+      memcpy(peer_message->mutable_payload(), message->payload(),
+             message->payload_size());
+      iter.second->PortsMessage(std::move(peer_message));
+    }
+  }
+}
+
 #if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
 void NodeController::OnRelayPortsMessage(const ports::NodeName& from_node,
                                          base::ProcessHandle from_process,
                                          const ports::NodeName& destination,
                                          Channel::MessagePtr message) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   if (GetBrokerChannel()) {
     // Only the broker should be asked to relay a message.
     LOG(ERROR) << "Non-broker refusing to relay message.";
@@ skipping 101 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo