Make mojo::Callback safe to copy across threads.

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

 // Copyright 2015 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/interface_endpoint_client.h"
 
 #include <stdint.h>
 #include <utility>
 
 #include "base/bind.h"
+#include "base/location.h"
 #include "base/macros.h"
 #include "base/message_loop/message_loop.h"
+#include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
+#include "base/thread_task_runner_handle.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/lib/multiplex_router.h"
 
 namespace mojo {
 namespace internal {
 
 // ----------------------------------------------------------------------------
 
 namespace {
 
+void DCheckIfValid(const base::WeakPtr<InterfaceEndpointClient>& client,
+                   const std::string& message) {
+  bool is_valid = client && !client->encountered_error();
+  DCHECK(!is_valid) << message;

[yzshen1, 2016/03/19 01:29:54]

So this function and ResponderThunk::DCheckValid should probably be named
DCheck*Invalid*, right?

[Anand Mistry (off Chromium), 2016/03/19 03:56:53]

Done.

+}
+
 // When receiving an incoming message which expects a repsonse,
 // InterfaceEndpointClient creates a ResponderThunk object and passes it to the
 // incoming message receiver. When the receiver finishes processing the message,
 // it can provide a response using this object.
 class ResponderThunk : public MessageReceiverWithStatus {
  public:
   explicit ResponderThunk(
       const base::WeakPtr<InterfaceEndpointClient>& endpoint_client)
-      : endpoint_client_(endpoint_client), accept_was_invoked_(false) {}
+      : endpoint_client_(endpoint_client), accept_was_invoked_(false),
+        task_runner_(base::ThreadTaskRunnerHandle::Get()) {}
   ~ResponderThunk() override {
     if (!accept_was_invoked_) {
       // The Mojo application handled a message that was expecting a response
       // but did not send a response.
-      if (endpoint_client_) {
-        // We raise an error to signal the calling application that an error
-        // condition occurred. Without this the calling application would have
-        // no way of knowing it should stop waiting for a response.
-        //
-        // We raise the error asynchronously and only if |endpoint_client_| is
-        // still alive when the task is run. That way it won't break the case
-        // where the user abandons the interface endpoint client soon after
-        // he/she abandons the callback.
-        base::MessageLoop::current()->PostTask(
+      if (task_runner_->RunsTasksOnCurrentThread()) {
+        if (endpoint_client_) {
+          // We raise an error to signal the calling application that an error
+          // condition occurred. Without this the calling application would have
+          // no way of knowing it should stop waiting for a response.
+          //
+          // We raise the error asynchronously and only if |endpoint_client_| is
+          // still alive when the task is run. That way it won't break the case
+          // where the user abandons the interface endpoint client soon after
+          // he/she abandons the callback.
+          task_runner_->PostTask(
+              FROM_HERE,
+              base::Bind(&InterfaceEndpointClient::RaiseError,
+                         endpoint_client_));
+        }
+      } else {
+        task_runner_->PostTask(
             FROM_HERE,
             base::Bind(&InterfaceEndpointClient::RaiseError, endpoint_client_));
       }
     }
   }
 
   // MessageReceiver implementation:
   bool Accept(Message* message) override {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
     accept_was_invoked_ = true;
     DCHECK(message->has_flag(kMessageIsResponse));
 
     bool result = false;
 
     if (endpoint_client_)
       result = endpoint_client_->Accept(message);
 
     return result;
   }
 
   // MessageReceiverWithStatus implementation:
   bool IsValid() override {
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
     return endpoint_client_ && !endpoint_client_->encountered_error();
   }
 
+  void DCheckValid(const std::string& message) override {
+    if (task_runner_->RunsTasksOnCurrentThread()) {
+      DCheckIfValid(endpoint_client_, message);
+    } else {
+      task_runner_->PostTask(
+          FROM_HERE, base::Bind(&DCheckIfValid, endpoint_client_, message));
+    }
+ }
+
  private:
   base::WeakPtr<InterfaceEndpointClient> endpoint_client_;
   bool accept_was_invoked_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
   DISALLOW_COPY_AND_ASSIGN(ResponderThunk);
 };
 
 }  // namespace
 
 // ----------------------------------------------------------------------------
 
 InterfaceEndpointClient::HandleIncomingMessageThunk::HandleIncomingMessageThunk(
     InterfaceEndpointClient* owner)
@@ skipping 143 matching lines @@
   } else {
     if (!incoming_receiver_)
       return false;
 
     return incoming_receiver_->Accept(message);
   }
 }
 
 }  // namespace internal
 }  // namespace mojo