Mojo: introducing a thread safe interface pointer.

mojo/public/cpp/bindings/thread_safe_interface_ptr.h

+// 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.
+
+#ifndef MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_
+#define MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_
+
+namespace mojo {
+
+#include <memory>
+
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/task_runner.h"
+#include "base/threading/thread_task_runner_handle.h"
+#include "mojo/public/cpp/bindings/interface_ptr.h"
+#include "mojo/public/cpp/bindings/lib/interface_ptr_state.h"
+#include "mojo/public/cpp/bindings/message.h"
+
+struct ThreadSafeInterfacePtrDeleter;
+
+// ThreadSafeInterfacePtr is a version of InterfacePtr that lets caller invoke
+// interface methods from any threads. Callbacks are called on the thread that
+// performed the interface call.
+// To create a ThreadSafeInterfacePtr, create first a regular InterfacePtr that
+// you then provide to ThreadSafeInterfacePtr::Create.
+// You can then call methods on the ThreadSafeInterfacePtr from any thread.
+//
+// Ex:
+// frob::FrobinatorPtr frobinator;
+// frob::FrobinatorImpl impl(GetProxy(&frobinator));
+// scoped_refptr<frob::ThreadSafeFrobinatorPtr> thread_safe_frobinator =
+//     frob::ThreadSafeFrobinatorPtr::Create(std::move(frobinator));
+// (*thread_safe_frobinator)->FrobinateToTheMax();
+
+template <typename Interface>
+class ThreadSafeInterfacePtr : public MessageReceiverWithResponder,
+    public base::RefCountedThreadSafe<ThreadSafeInterfacePtr<Interface>,
+                                      ThreadSafeInterfacePtrDeleter> {
+ public:
+  using ProxyType = typename Interface::Proxy_;
+
+  using AcceptCallback = base::Callback<void(Message)>;
+  using AcceptWithResponderCallback =
+      base::Callback<void(Message, std::unique_ptr<MessageReceiver>)>;
+
+  Interface* get() { return &proxy_; }
+  Interface* operator->() { return get(); }
+  Interface& operator*() { return *get(); }
+
+  static ThreadSafeInterfacePtr<Interface>* Create(

[yzshen1, 2016/11/15 18:53:46]

It seems better to return a scoped_refptr.

[Jay Civelli, 2016/11/15 19:36:38]

Good idea, done.

+      InterfacePtr<Interface> interface_ptr) {
+    return new ThreadSafeInterfacePtr(std::move(interface_ptr),

[yzshen1, 2016/11/15 18:53:46]

Please also deal with the case where interface_ptr is not bound. Maybe we return
null in that case?

[Jay Civelli, 2016/11/15 19:36:38]

Done.

+                                      base::ThreadTaskRunnerHandle::Get());
+  }
+

[yzshen1, 2016/11/15 18:53:46]

Do we need to also expose connection error handler?

[Jay Civelli, 2016/11/15 19:36:38]

That's a good question.
Happy to add them to ThreadSafeInterfacePointer if you believe it makes sense.

From my point of view, I wonder if the connection error handlers should be set
on the original InterfacePtr before it gets passed to the
ThreadSafeInterfacePointer.
My thought is just for clarity reasons: with the InterfacePtr it's clear
everything happens on the same thread.
With ThreadSafeInterfacePointer, the handlers would still be invoked on the
InterfacePtr's thread, but it might not be immediately clear to consumers.

[yzshen1, 2016/11/15 20:26:15]

Yeah. I think that makes sense.

+ private:
+  friend class base::RefCountedThreadSafe<ThreadSafeInterfacePtr<Interface>>;
+  friend struct ThreadSafeInterfacePtrDeleter;
+
+  ThreadSafeInterfacePtr(
+      InterfacePtr<Interface> interface_ptr,
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner)
+      : interface_ptr_task_runner_(task_runner),
+        proxy_(this),
+        interface_ptr_(std::move(interface_ptr)) {
+    // Note that it's important we do get the callback after interface_ptr_ has
+    // been set, as they would become invalid if interface_ptr_ is copied.
+    accept_callback_ = interface_ptr_.internal_state()->
+        GetThreadSafePtrAcceptCallback();
+    accept_with_responder_callback_ = interface_ptr_.internal_state()->
+        GetThreadSafePtrAcceptWithResponderCallback();
+  }
+
+  void DeleteOnCorrectThread() const {
+    if (!interface_ptr_task_runner_->BelongsToCurrentThread() &&
+        interface_ptr_task_runner_->DeleteSoon(FROM_HERE, this)) {
+      return;
+    }
+    delete this;
+  }
+
+  // MessageReceiverWithResponder implementation:
+  bool Accept(Message* message) override {
+    interface_ptr_task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(accept_callback_, base::Passed(std::move(*message))));
+    return true;
+  }
+
+  bool AcceptWithResponder(Message* message,
+                           MessageReceiver* responder) override {
+    auto forward_responder = base::MakeUnique<ForwardToCallingThread>(
+        base::WrapUnique(responder));
+    interface_ptr_task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(accept_with_responder_callback_,
+                   base::Passed(std::move(*message)),
+                   base::Passed(std::move(forward_responder))));
+    return true;
+  }
+
+  class ForwardToCallingThread : public MessageReceiver {
+   public:
+    explicit ForwardToCallingThread(std::unique_ptr<MessageReceiver> responder)
+        : responder_(std::move(responder)),
+          caller_task_runner_(base::ThreadTaskRunnerHandle::Get()) {
+    }
+
+   private:
+    bool Accept(Message* message) {
+      // The current instance will be deleted when this method returns, so we
+      // have to relinquish the responder's ownership so it does not get
+      // deleted.
+      caller_task_runner_->PostTask(FROM_HERE,
+          base::Bind(&ForwardToCallingThread::CallAcceptAndDeleteResponder,
+                     base::Passed(std::move(responder_)),
+                     base::Passed(std::move(*message))));
+      return true;
+    }
+
+    static void CallAcceptAndDeleteResponder(
+        std::unique_ptr<MessageReceiver> responder,
+        Message message) {
+      ignore_result(responder->Accept(&message));
+    }
+
+    std::unique_ptr<MessageReceiver> responder_;
+    scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner_;
+  };
+
+  scoped_refptr<base::SingleThreadTaskRunner> interface_ptr_task_runner_;
+  ProxyType proxy_;
+  AcceptCallback accept_callback_;
+  AcceptWithResponderCallback accept_with_responder_callback_;
+  InterfacePtr<Interface> interface_ptr_;
+};
+
+struct ThreadSafeInterfacePtrDeleter {
+  template <typename Interface>
+  static void Destruct(const ThreadSafeInterfacePtr<Interface>* interface_ptr) {
+    interface_ptr->DeleteOnCorrectThread();
+  }
+};
+
+}  // namespace mojo
+
+#endif  // MOJO_PUBLIC_CPP_BINDINGS_THREAD_SAFE_INTERFACE_PTR_H_