ipc_message_pipe_reader.cc 5.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185
  1. // Copyright 2014 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "ipc/ipc_message_pipe_reader.h"
  5. #include <stdint.h>
  6. #include <utility>
  7. #include "base/bind.h"
  8. #include "base/callback_helpers.h"
  9. #include "base/containers/span.h"
  10. #include "base/location.h"
  11. #include "base/logging.h"
  12. #include "base/task/single_thread_task_runner.h"
  13. #include "base/threading/thread_task_runner_handle.h"
  14. #include "base/trace_event/trace_event.h"
  15. #include "ipc/ipc_channel_mojo.h"
  16. #include "mojo/public/cpp/bindings/message.h"
  17. #include "mojo/public/cpp/bindings/thread_safe_proxy.h"
  18. namespace IPC {
  19. namespace internal {
  20. namespace {
  21. class ThreadSafeProxy : public mojo::ThreadSafeProxy {
  22. public:
  23. using Forwarder = base::RepeatingCallback<void(mojo::Message)>;
  24. ThreadSafeProxy(scoped_refptr<base::SequencedTaskRunner> task_runner,
  25. Forwarder forwarder,
  26. mojo::AssociatedGroupController& group_controller)
  27. : task_runner_(std::move(task_runner)),
  28. forwarder_(std::move(forwarder)),
  29. group_controller_(group_controller) {}
  30. // mojo::ThreadSafeProxy:
  31. void SendMessage(mojo::Message& message) override {
  32. message.SerializeHandles(&group_controller_);
  33. task_runner_->PostTask(FROM_HERE,
  34. base::BindOnce(forwarder_, std::move(message)));
  35. }
  36. void SendMessageWithResponder(
  37. mojo::Message& message,
  38. std::unique_ptr<mojo::MessageReceiver> responder) override {
  39. // We don't bother supporting this because it's not used in practice.
  40. NOTREACHED();
  41. }
  42. private:
  43. ~ThreadSafeProxy() override = default;
  44. const scoped_refptr<base::SequencedTaskRunner> task_runner_;
  45. const Forwarder forwarder_;
  46. mojo::AssociatedGroupController& group_controller_;
  47. };
  48. } // namespace
  49. MessagePipeReader::MessagePipeReader(
  50. mojo::MessagePipeHandle pipe,
  51. mojo::PendingAssociatedRemote<mojom::Channel> sender,
  52. mojo::PendingAssociatedReceiver<mojom::Channel> receiver,
  53. scoped_refptr<base::SequencedTaskRunner> task_runner,
  54. MessagePipeReader::Delegate* delegate)
  55. : delegate_(delegate),
  56. sender_(std::move(sender), task_runner),
  57. receiver_(this, std::move(receiver), task_runner) {
  58. thread_safe_sender_ =
  59. std::make_unique<mojo::ThreadSafeForwarder<mojom::Channel>>(
  60. base::MakeRefCounted<ThreadSafeProxy>(
  61. task_runner,
  62. base::BindRepeating(&MessagePipeReader::ForwardMessage,
  63. weak_ptr_factory_.GetWeakPtr()),
  64. *sender_.internal_state()->associated_group()->GetController()));
  65. thread_checker_.DetachFromThread();
  66. }
  67. MessagePipeReader::~MessagePipeReader() {
  68. DCHECK(thread_checker_.CalledOnValidThread());
  69. // The pipe should be closed before deletion.
  70. }
  71. void MessagePipeReader::FinishInitializationOnIOThread(
  72. base::ProcessId self_pid) {
  73. sender_.set_disconnect_handler(
  74. base::BindOnce(&MessagePipeReader::OnPipeError, base::Unretained(this),
  75. MOJO_RESULT_FAILED_PRECONDITION));
  76. receiver_.set_disconnect_handler(
  77. base::BindOnce(&MessagePipeReader::OnPipeError, base::Unretained(this),
  78. MOJO_RESULT_FAILED_PRECONDITION));
  79. sender_->SetPeerPid(self_pid);
  80. }
  81. void MessagePipeReader::Close() {
  82. DCHECK(thread_checker_.CalledOnValidThread());
  83. sender_.reset();
  84. if (receiver_.is_bound())
  85. receiver_.reset();
  86. }
  87. bool MessagePipeReader::Send(std::unique_ptr<Message> message) {
  88. CHECK(message->IsValid());
  89. TRACE_EVENT_WITH_FLOW0("toplevel.flow", "MessagePipeReader::Send",
  90. message->flags(), TRACE_EVENT_FLAG_FLOW_OUT);
  91. absl::optional<std::vector<mojo::native::SerializedHandlePtr>> handles;
  92. MojoResult result = MOJO_RESULT_OK;
  93. result = ChannelMojo::ReadFromMessageAttachmentSet(message.get(), &handles);
  94. if (result != MOJO_RESULT_OK)
  95. return false;
  96. if (!sender_)
  97. return false;
  98. base::span<const uint8_t> bytes(static_cast<const uint8_t*>(message->data()),
  99. message->size());
  100. sender_->Receive(MessageView(bytes, std::move(handles)));
  101. DVLOG(4) << "Send " << message->type() << ": " << message->size();
  102. return true;
  103. }
  104. void MessagePipeReader::GetRemoteInterface(
  105. mojo::GenericPendingAssociatedReceiver receiver) {
  106. if (!sender_.is_bound())
  107. return;
  108. sender_->GetAssociatedInterface(std::move(receiver));
  109. }
  110. void MessagePipeReader::SetPeerPid(int32_t peer_pid) {
  111. delegate_->OnPeerPidReceived(peer_pid);
  112. }
  113. void MessagePipeReader::Receive(MessageView message_view) {
  114. if (message_view.bytes().empty()) {
  115. delegate_->OnBrokenDataReceived();
  116. return;
  117. }
  118. Message message(reinterpret_cast<const char*>(message_view.bytes().data()),
  119. message_view.bytes().size());
  120. if (!message.IsValid()) {
  121. delegate_->OnBrokenDataReceived();
  122. return;
  123. }
  124. DVLOG(4) << "Receive " << message.type() << ": " << message.size();
  125. MojoResult write_result = ChannelMojo::WriteToMessageAttachmentSet(
  126. message_view.TakeHandles(), &message);
  127. if (write_result != MOJO_RESULT_OK) {
  128. OnPipeError(write_result);
  129. return;
  130. }
  131. TRACE_EVENT_WITH_FLOW0("toplevel.flow", "MessagePipeReader::Receive",
  132. message.flags(), TRACE_EVENT_FLAG_FLOW_IN);
  133. delegate_->OnMessageReceived(message);
  134. }
  135. void MessagePipeReader::GetAssociatedInterface(
  136. mojo::GenericPendingAssociatedReceiver receiver) {
  137. DCHECK(thread_checker_.CalledOnValidThread());
  138. if (delegate_)
  139. delegate_->OnAssociatedInterfaceRequest(std::move(receiver));
  140. }
  141. void MessagePipeReader::OnPipeError(MojoResult error) {
  142. DCHECK(thread_checker_.CalledOnValidThread());
  143. Close();
  144. // NOTE: The delegate call below may delete |this|.
  145. if (delegate_)
  146. delegate_->OnPipeError();
  147. }
  148. void MessagePipeReader::ForwardMessage(mojo::Message message) {
  149. sender_.internal_state()->ForwardMessage(std::move(message));
  150. }
  151. } // namespace internal
  152. } // namespace IPC