123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185 |
- // 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_message_pipe_reader.h"
- #include <stdint.h>
- #include <utility>
- #include "base/bind.h"
- #include "base/callback_helpers.h"
- #include "base/containers/span.h"
- #include "base/location.h"
- #include "base/logging.h"
- #include "base/task/single_thread_task_runner.h"
- #include "base/threading/thread_task_runner_handle.h"
- #include "base/trace_event/trace_event.h"
- #include "ipc/ipc_channel_mojo.h"
- #include "mojo/public/cpp/bindings/message.h"
- #include "mojo/public/cpp/bindings/thread_safe_proxy.h"
- namespace IPC {
- namespace internal {
- namespace {
- class ThreadSafeProxy : public mojo::ThreadSafeProxy {
- public:
- using Forwarder = base::RepeatingCallback<void(mojo::Message)>;
- ThreadSafeProxy(scoped_refptr<base::SequencedTaskRunner> task_runner,
- Forwarder forwarder,
- mojo::AssociatedGroupController& group_controller)
- : task_runner_(std::move(task_runner)),
- forwarder_(std::move(forwarder)),
- group_controller_(group_controller) {}
- // mojo::ThreadSafeProxy:
- void SendMessage(mojo::Message& message) override {
- message.SerializeHandles(&group_controller_);
- task_runner_->PostTask(FROM_HERE,
- base::BindOnce(forwarder_, std::move(message)));
- }
- void SendMessageWithResponder(
- mojo::Message& message,
- std::unique_ptr<mojo::MessageReceiver> responder) override {
- // We don't bother supporting this because it's not used in practice.
- NOTREACHED();
- }
- private:
- ~ThreadSafeProxy() override = default;
- const scoped_refptr<base::SequencedTaskRunner> task_runner_;
- const Forwarder forwarder_;
- mojo::AssociatedGroupController& group_controller_;
- };
- } // namespace
- MessagePipeReader::MessagePipeReader(
- mojo::MessagePipeHandle pipe,
- mojo::PendingAssociatedRemote<mojom::Channel> sender,
- mojo::PendingAssociatedReceiver<mojom::Channel> receiver,
- scoped_refptr<base::SequencedTaskRunner> task_runner,
- MessagePipeReader::Delegate* delegate)
- : delegate_(delegate),
- sender_(std::move(sender), task_runner),
- receiver_(this, std::move(receiver), task_runner) {
- thread_safe_sender_ =
- std::make_unique<mojo::ThreadSafeForwarder<mojom::Channel>>(
- base::MakeRefCounted<ThreadSafeProxy>(
- task_runner,
- base::BindRepeating(&MessagePipeReader::ForwardMessage,
- weak_ptr_factory_.GetWeakPtr()),
- *sender_.internal_state()->associated_group()->GetController()));
- thread_checker_.DetachFromThread();
- }
- MessagePipeReader::~MessagePipeReader() {
- DCHECK(thread_checker_.CalledOnValidThread());
- // The pipe should be closed before deletion.
- }
- void MessagePipeReader::FinishInitializationOnIOThread(
- base::ProcessId self_pid) {
- sender_.set_disconnect_handler(
- base::BindOnce(&MessagePipeReader::OnPipeError, base::Unretained(this),
- MOJO_RESULT_FAILED_PRECONDITION));
- receiver_.set_disconnect_handler(
- base::BindOnce(&MessagePipeReader::OnPipeError, base::Unretained(this),
- MOJO_RESULT_FAILED_PRECONDITION));
- sender_->SetPeerPid(self_pid);
- }
- void MessagePipeReader::Close() {
- DCHECK(thread_checker_.CalledOnValidThread());
- sender_.reset();
- if (receiver_.is_bound())
- receiver_.reset();
- }
- bool MessagePipeReader::Send(std::unique_ptr<Message> message) {
- CHECK(message->IsValid());
- TRACE_EVENT_WITH_FLOW0("toplevel.flow", "MessagePipeReader::Send",
- message->flags(), TRACE_EVENT_FLAG_FLOW_OUT);
- absl::optional<std::vector<mojo::native::SerializedHandlePtr>> handles;
- MojoResult result = MOJO_RESULT_OK;
- result = ChannelMojo::ReadFromMessageAttachmentSet(message.get(), &handles);
- if (result != MOJO_RESULT_OK)
- return false;
- if (!sender_)
- return false;
- base::span<const uint8_t> bytes(static_cast<const uint8_t*>(message->data()),
- message->size());
- sender_->Receive(MessageView(bytes, std::move(handles)));
- DVLOG(4) << "Send " << message->type() << ": " << message->size();
- return true;
- }
- void MessagePipeReader::GetRemoteInterface(
- mojo::GenericPendingAssociatedReceiver receiver) {
- if (!sender_.is_bound())
- return;
- sender_->GetAssociatedInterface(std::move(receiver));
- }
- void MessagePipeReader::SetPeerPid(int32_t peer_pid) {
- delegate_->OnPeerPidReceived(peer_pid);
- }
- void MessagePipeReader::Receive(MessageView message_view) {
- if (message_view.bytes().empty()) {
- delegate_->OnBrokenDataReceived();
- return;
- }
- Message message(reinterpret_cast<const char*>(message_view.bytes().data()),
- message_view.bytes().size());
- if (!message.IsValid()) {
- delegate_->OnBrokenDataReceived();
- return;
- }
- DVLOG(4) << "Receive " << message.type() << ": " << message.size();
- MojoResult write_result = ChannelMojo::WriteToMessageAttachmentSet(
- message_view.TakeHandles(), &message);
- if (write_result != MOJO_RESULT_OK) {
- OnPipeError(write_result);
- return;
- }
- TRACE_EVENT_WITH_FLOW0("toplevel.flow", "MessagePipeReader::Receive",
- message.flags(), TRACE_EVENT_FLAG_FLOW_IN);
- delegate_->OnMessageReceived(message);
- }
- void MessagePipeReader::GetAssociatedInterface(
- mojo::GenericPendingAssociatedReceiver receiver) {
- DCHECK(thread_checker_.CalledOnValidThread());
- if (delegate_)
- delegate_->OnAssociatedInterfaceRequest(std::move(receiver));
- }
- void MessagePipeReader::OnPipeError(MojoResult error) {
- DCHECK(thread_checker_.CalledOnValidThread());
- Close();
- // NOTE: The delegate call below may delete |this|.
- if (delegate_)
- delegate_->OnPipeError();
- }
- void MessagePipeReader::ForwardMessage(mojo::Message message) {
- sender_.internal_state()->ForwardMessage(std::move(message));
- }
- } // namespace internal
- } // namespace IPC
|