Mojo C++ bindings: refactor SyncHandleWatcher.

mojo/public/cpp/bindings/lib/router.cc

 // Copyright 2014 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/public/cpp/bindings/lib/router.h"
 
 #include <stdint.h>
 #include <utility>
 
 #include "base/bind.h"
@@ skipping 106 matching lines @@
       filters_(std::move(filters)),
       connector_(std::move(message_pipe), Connector::SINGLE_THREADED_SEND),
       incoming_receiver_(nullptr),
       next_request_id_(0),
       testing_mode_(false),
       pending_task_for_messages_(false),
       encountered_error_(false),
       weak_factory_(this) {
   filters_.SetSink(&thunk_);
   if (expects_sync_requests)
-    connector_.RegisterSyncHandleWatch();
+    connector_.AllowWokenUpBySyncWatchOnSameThread();
   connector_.set_incoming_receiver(filters_.GetHead());
   connector_.set_connection_error_handler([this]() { OnConnectionError(); });
 }
 
 Router::~Router() {}
 
 bool Router::Accept(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(!message->has_flag(kMessageExpectsResponse));
   return connector_.Accept(message);
@@ skipping 11 matching lines @@
   message->set_request_id(request_id);
   if (!connector_.Accept(message))
     return false;
 
   if (!message->has_flag(kMessageIsSync)) {
     // We assume ownership of |responder|.
     async_responders_[request_id] = make_scoped_ptr(responder);
     return true;
   }
 
-  if (!connector_.RegisterSyncHandleWatch())
-    return false;
-
   bool response_received = false;
   scoped_ptr<MessageReceiver> sync_responder(responder);
   sync_responses_.insert(std::make_pair(
       request_id, make_scoped_ptr(new SyncResponseInfo(&response_received))));
 
   base::WeakPtr<Router> weak_self = weak_factory_.GetWeakPtr();
-  bool result = connector_.RunSyncHandleWatch(&response_received);
+  bool result = connector_.SyncWatch(&response_received);
   // Make sure that this instance hasn't been destroyed.
   if (weak_self) {
     DCHECK(ContainsKey(sync_responses_, request_id));
     auto iter = sync_responses_.find(request_id);
     DCHECK_EQ(&response_received, iter->second->response_received);
     if (result && response_received) {
       scoped_ptr<Message> response = std::move(iter->second->response);
       ignore_result(sync_responder->Accept(response.get()));
     }
     sync_responses_.erase(iter);
-
-    connector_.UnregisterSyncHandleWatch();
   }
 
   // Return true means that we take ownership of |responder|.
   return true;
 }
 
 void Router::EnableTestingMode() {
   DCHECK(thread_checker_.CalledOnValidThread());
   testing_mode_ = true;
   connector_.set_enforce_errors_from_incoming_receiver(false);
@@ skipping 88 matching lines @@
     return responder->Accept(message);
   } else {
     if (!incoming_receiver_)
       return false;
 
     return incoming_receiver_->Accept(message);
   }
 }
 
 void Router::OnConnectionError() {
-  DCHECK(!encountered_error_);
+  if (encountered_error_)
+    return;
 
   if (!pending_messages_.empty()) {
     // After all the pending messages are processed, we will check whether an
     // error has been encountered and run the user's connection error handler
     // if necessary.
     DCHECK(pending_task_for_messages_);
     return;
   }
 
+  if (connector_.during_sync_handle_watcher_callback()) {
+    // We don't want the error handler to reenter an ongoing sync call.
+    base::MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&Router::OnConnectionError, weak_factory_.GetWeakPtr()));
+    return;
+  }
+
   encountered_error_ = true;
   error_handler_.Run();
 }
 
 // ----------------------------------------------------------------------------
 
 }  // namespace internal
 }  // namespace mojo