[mojo-edk] Revert ObserveProxy retransmission behavior

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 634 matching lines @@
     scoped_refptr<NodeChannel> broker = GetBrokerChannel();
     if (!broker) {
       DVLOG(1) << "Dropping message for unknown peer: " << name;
       return;
     }
     broker->RequestIntroduction(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 (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.
 
-      while (!messages.empty()) {
-        node_->AcceptMessage(std::move(messages.front()));
-        messages.pop();
-      }
+    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();
     }
   }
 
   AttemptShutdownIfRequested();
 }
 
 void NodeController::ProcessIncomingMessages() {
   RequestContext request_context(RequestContext::Source::SYSTEM);
 
   {
@@ skipping 61 matching lines @@
     // ForwardMessage. Because ForwardMessage is only ever called
     // (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_);
     // |io_task_runner_| may be null in tests or processes that don't require
     // multi-process Mojo.
     schedule_pump_task = incoming_messages_.empty() && io_task_runner_ &&
         !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));
   }
 
   if (schedule_pump_task) {
     // Normally, the queue is processed after the action that added the local
     // message is done (i.e. SendMessage, ClosePort, etc). However, it's also
     // possible for a local message to be added as a result of a remote message,
     // and OnChannelMessage() doesn't process this queue (although
     // OnPortsMessage() does). There may also be other code paths, now or added
@@ skipping 495 matching lines @@
 
 void NodeController::AttemptShutdownIfRequested() {
   if (!shutdown_callback_flag_)
     return;
 
   base::Closure callback;
   {
     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;
     }
 
     callback = shutdown_callback_;
     shutdown_callback_.Reset();
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo