[mojo-edk] Revert ObserveProxy retransmission behavior

mojo/edk/system/node_controller.cc

Index: mojo/edk/system/node_controller.cc
diff --git a/mojo/edk/system/node_controller.cc b/mojo/edk/system/node_controller.cc
index a3ef2cda8197911c530f48f82b3dd614bb6726a4..912ecee77468d2fa0bfb4c7d3526b96ee7b4e3fa 100644
--- a/mojo/edk/system/node_controller.cc
+++ b/mojo/edk/system/node_controller.cc
@@ -652,19 +652,28 @@ void NodeController::SendPeerMessage(const ports::NodeName& name,
 }
 
 void NodeController::AcceptIncomingMessages() {
-  {
-    base::AutoLock lock(messages_lock_);
-    if (!incoming_messages_.empty()) {
-      // libstdc++'s deque creates an internal buffer on construction, even when
-      // the size is 0. So avoid creating it until it is necessary.
-      std::queue<ports::ScopedMessage> messages;
-      std::swap(messages, incoming_messages_);
-      base::AutoUnlock unlock(messages_lock_);
-
-      while (!messages.empty()) {
-        node_->AcceptMessage(std::move(messages.front()));
-        messages.pop();
-      }
+  while (incoming_messages_flag_) {
+    // TODO: We may need to be more careful to avoid starving the rest of the
+    // thread here. Revisit this if it turns out to be a problem. One
+    // alternative would be to schedule a task to continue pumping messages
+    // after flushing once.
+
+    messages_lock_.Acquire();
+    if (incoming_messages_.empty()) {
+      messages_lock_.Release();
+      break;
+    }
+
+    // libstdc++'s deque creates an internal buffer on construction, even when
+    // the size is 0. So avoid creating it until it is necessary.
+    std::queue<ports::ScopedMessage> messages;
+    std::swap(messages, incoming_messages_);
+    incoming_messages_flag_.Set(false);
+    messages_lock_.Release();
+
+    while (!messages.empty()) {
+      node_->AcceptMessage(std::move(messages.front()));
+      messages.pop();
     }
   }
 
@@ -746,6 +755,7 @@ void NodeController::ForwardMessage(const ports::NodeName& node,
         !incoming_messages_task_posted_;
     incoming_messages_task_posted_ |= schedule_pump_task;
     incoming_messages_.emplace(std::move(message));
+    incoming_messages_flag_.Set(true);
   } else {
     SendPeerMessage(node, std::move(message));
   }
@@ -1261,7 +1271,8 @@ void NodeController::AttemptShutdownIfRequested() {
     base::AutoLock lock(shutdown_lock_);
     if (shutdown_callback_.is_null())
       return;
-    if (!node_->CanShutdownCleanly(true /* allow_local_ports */)) {
+    if (!node_->CanShutdownCleanly(
+          ports::Node::ShutdownPolicy::ALLOW_LOCAL_PORTS)) {
       DVLOG(2) << "Unable to cleanly shut down node " << name_;
       return;
     }