Mojo C++ bindings: error notification behavior related to sync calls.

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 75 matching lines @@
 Router::Router(ScopedMessagePipeHandle message_pipe,
                FilterChain filters,
                bool expects_sync_requests)
     : thunk_(this),
       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_.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 87 matching lines @@
     if (!weak_self)
       return;
 
     if (!result && !testing_mode_) {
       connector_.RaiseError();
       break;
     }
   }
 
   pending_task_for_messages_ = false;
+
+  // We may have already seen a connection error from the connector, but
+  // haven't notified the user because we want to process all the queued
+  // messages first. We should do it now.
+  if (connector_.encountered_error() && !encountered_error_)
+    OnConnectionError();
 }
 
 bool Router::HandleMessageInternal(Message* message) {
   if (message->has_flag(kMessageExpectsResponse)) {
     if (!incoming_receiver_)
       return false;
 
     MessageReceiverWithStatus* responder =
         new ResponderThunk(weak_factory_.GetWeakPtr());
     bool ok = incoming_receiver_->AcceptWithResponder(message, responder);
@@ skipping 25 matching lines @@
     async_responders_.erase(it);
     return responder->Accept(message);
   } else {
     if (!incoming_receiver_)
       return false;
 
     return incoming_receiver_->Accept(message);
   }
 }
 
+void Router::OnConnectionError() {
+  DCHECK(!encountered_error_);
+
+  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;
+  }
+
+  encountered_error_ = true;
+  error_handler_.Run();
+}
+
 // ----------------------------------------------------------------------------
 
 }  // namespace internal
 }  // namespace mojo