Mojo C++ Bindings: Eliminate unbound ThreadSafeInterfacePtr

ipc/ipc_channel_mojo.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 "ipc/ipc_channel_mojo.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <memory>
@@ skipping 256 matching lines @@
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
   return base::MakeUnique<MojoChannelFactory>(
       std::move(handle), Channel::MODE_CLIENT, ipc_task_runner);
 }
 
 ChannelMojo::ChannelMojo(
     mojo::ScopedMessagePipeHandle handle,
     Mode mode,
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
-    : pipe_(handle.get()), listener_(listener), weak_factory_(this) {
-  // Create MojoBootstrap after all members are set as it touches
-  // ChannelMojo from a different thread.
-  bootstrap_ =
-      MojoBootstrap::Create(std::move(handle), mode, this, ipc_task_runner);
+    : task_runner_(ipc_task_runner),
+      pipe_(handle.get()),
+      listener_(listener),
+      weak_factory_(this) {
+  bootstrap_ = MojoBootstrap::Create(std::move(handle), mode, ipc_task_runner);
+}
+
+void ChannelMojo::ForwardMessageFromThreadSafePtr(mojo::Message message) {
+  DCHECK(task_runner_->RunsTasksOnCurrentThread());
+  if (!message_reader_)
+    return;
+  message_reader_->sender().internal_state()->ForwardMessage(
+      std::move(message));
+}
+
+void ChannelMojo::ForwardMessageWithResponderFromThreadSafePtr(
+    mojo::Message message,
+    std::unique_ptr<mojo::MessageReceiver> responder) {
+  DCHECK(task_runner_->RunsTasksOnCurrentThread());
+  if (!message_reader_)
+    return;
+  message_reader_->sender().internal_state()->ForwardMessageWithResponder(
+      std::move(message), std::move(responder));
 }
 
 ChannelMojo::~ChannelMojo() {
+  DCHECK(task_runner_->RunsTasksOnCurrentThread());
   Close();
 }
 
 bool ChannelMojo::Connect() {
+  DCHECK(task_runner_->RunsTasksOnCurrentThread());
+
   WillConnect();
 
-  DCHECK(!task_runner_);
-  task_runner_ = base::ThreadTaskRunnerHandle::Get();
+  mojom::ChannelAssociatedPtr sender;
+  mojom::ChannelAssociatedRequest receiver;
+  bootstrap_->Connect(&sender, &receiver);
+
   DCHECK(!message_reader_);
-
-  bootstrap_->Connect();
+  sender->SetPeerPid(GetSelfPID());
+  message_reader_.reset(new internal::MessagePipeReader(
+      pipe_, std::move(sender), std::move(receiver), this));
   return true;
 }
 
 void ChannelMojo::Pause() {
   bootstrap_->Pause();
 }
 
 void ChannelMojo::Unpause(bool flush) {
   bootstrap_->Unpause();
   if (flush)
     Flush();
 }
 
 void ChannelMojo::Flush() {
   bootstrap_->Flush();
 }
 
 void ChannelMojo::Close() {
   // NOTE: The MessagePipeReader's destructor may re-enter this function. Use
   // caution when changing this method.
   std::unique_ptr<internal::MessagePipeReader> reader =
       std::move(message_reader_);
   reader.reset();
 
   base::AutoLock lock(associated_interface_lock_);
   associated_interfaces_.clear();
 }
 
-// MojoBootstrap::Delegate implementation
-void ChannelMojo::OnPipesAvailable(mojom::ChannelAssociatedPtr sender,
-                                   mojom::ChannelAssociatedRequest receiver) {
-  sender->SetPeerPid(GetSelfPID());
-  message_reader_.reset(new internal::MessagePipeReader(
-      pipe_, std::move(sender), std::move(receiver), this));
-}
-
 void ChannelMojo::OnPipeError() {
   DCHECK(task_runner_);
   if (task_runner_->RunsTasksOnCurrentThread()) {
     listener_->OnChannelError();
   } else {
     task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&ChannelMojo::OnPipeError, weak_factory_.GetWeakPtr()));
   }
 }
@@ skipping 28 matching lines @@
   // called.
   //
   // With Mojo, there's no OnFileCanReadWithoutBlocking, but we expect the
   // pipe's connection error handler will be invoked in its place.
   return message_reader_->Send(std::move(scoped_message));
 }
 
 Channel::AssociatedInterfaceSupport*
 ChannelMojo::GetAssociatedInterfaceSupport() { return this; }
 
+std::unique_ptr<mojo::ThreadSafeForwarder<mojom::Channel>>
+ChannelMojo::CreateThreadSafeChannel() {
+  return base::MakeUnique<mojo::ThreadSafeForwarder<mojom::Channel>>(
+      task_runner_, base::Bind(&ChannelMojo::ForwardMessageFromThreadSafePtr,
+                               weak_factory_.GetWeakPtr()),
+      base::Bind(&ChannelMojo::ForwardMessageWithResponderFromThreadSafePtr,
+                 weak_factory_.GetWeakPtr()));
+}
+
 void ChannelMojo::OnPeerPidReceived(int32_t peer_pid) {
   listener_->OnChannelConnected(peer_pid);
 }
 
 void ChannelMojo::OnMessageReceived(const Message& message) {
   TRACE_EVENT2("ipc,toplevel", "ChannelMojo::OnMessageReceived",
                "class", IPC_MESSAGE_ID_CLASS(message.type()),
                "line", IPC_MESSAGE_ID_LINE(message.type()));
   listener_->OnMessageReceived(message);
   if (message.dispatch_error())
@@ skipping 66 matching lines @@
 }
 
 void ChannelMojo::GetGenericRemoteAssociatedInterface(
     const std::string& name,
     mojo::ScopedInterfaceEndpointHandle handle) {
   if (message_reader_)
     message_reader_->GetRemoteInterface(name, std::move(handle));
 }
 
 }  // namespace IPC