// Copyright (c) 2012 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_reader.h" #include #include "ipc/ipc_listener.h" #include "ipc/ipc_logging.h" #include "ipc/ipc_message.h" #include "ipc/ipc_message_attachment_set.h" #include "ipc/ipc_message_macros.h" namespace IPC { namespace internal { ChannelReader::ChannelReader(Listener* listener) : listener_(listener) { memset(input_buf_, 0, sizeof(input_buf_)); } ChannelReader::~ChannelReader() { if (!blocked_ids_.empty()) StopObservingAttachmentBroker(); } ChannelReader::DispatchState ChannelReader::ProcessIncomingMessages() { while (true) { int bytes_read = 0; ReadState read_state = ReadData(input_buf_, Channel::kReadBufferSize, &bytes_read); if (read_state == READ_FAILED) return DISPATCH_ERROR; if (read_state == READ_PENDING) return DISPATCH_FINISHED; DCHECK(bytes_read > 0); if (!TranslateInputData(input_buf_, bytes_read)) return DISPATCH_ERROR; DispatchState state = DispatchMessages(); if (state != DISPATCH_FINISHED) return state; } } ChannelReader::DispatchState ChannelReader::AsyncReadComplete(int bytes_read) { if (!TranslateInputData(input_buf_, bytes_read)) return DISPATCH_ERROR; return DispatchMessages(); } bool ChannelReader::IsInternalMessage(const Message& m) { return m.routing_id() == MSG_ROUTING_NONE && m.type() >= Channel::CLOSE_FD_MESSAGE_TYPE && m.type() <= Channel::HELLO_MESSAGE_TYPE; } bool ChannelReader::IsHelloMessage(const Message& m) { return m.routing_id() == MSG_ROUTING_NONE && m.type() == Channel::HELLO_MESSAGE_TYPE; } bool ChannelReader::TranslateInputData(const char* input_data, int input_data_len) { const char* p; const char* end; // Possibly combine with the overflow buffer to make a larger buffer. if (input_overflow_buf_.empty()) { p = input_data; end = input_data + input_data_len; } else { if (input_overflow_buf_.size() + input_data_len > Channel::kMaximumMessageSize) { input_overflow_buf_.clear(); LOG(ERROR) << "IPC message is too big"; return false; } input_overflow_buf_.append(input_data, input_data_len); p = input_overflow_buf_.data(); end = p + input_overflow_buf_.size(); } // Dispatch all complete messages in the data buffer. while (p < end) { const char* message_tail = Message::FindNext(p, end); if (message_tail) { int len = static_cast(message_tail - p); Message translated_message(p, len); if (!GetNonBrokeredAttachments(&translated_message)) return false; // If there are no queued messages, attempt to immediately dispatch the // newly translated message. if (queued_messages_.empty()) { DCHECK(blocked_ids_.empty()); AttachmentIdSet blocked_ids = GetBrokeredAttachments(&translated_message); if (blocked_ids.empty()) { // Dispatch the message and continue the loop. DispatchMessage(&translated_message); p = message_tail; continue; } blocked_ids_.swap(blocked_ids); StartObservingAttachmentBroker(); } // Make a deep copy of |translated_message| to add to the queue. scoped_ptr m(new Message(translated_message)); queued_messages_.push_back(m.release()); p = message_tail; } else { // Last message is partial. break; } } // Save any partial data in the overflow buffer. input_overflow_buf_.assign(p, end - p); if (input_overflow_buf_.empty() && !DidEmptyInputBuffers()) return false; return true; } ChannelReader::DispatchState ChannelReader::DispatchMessages() { while (!queued_messages_.empty()) { if (!blocked_ids_.empty()) return DISPATCH_WAITING_ON_BROKER; Message* m = queued_messages_.front(); AttachmentIdSet blocked_ids = GetBrokeredAttachments(m); if (!blocked_ids.empty()) { blocked_ids_.swap(blocked_ids); StartObservingAttachmentBroker(); return DISPATCH_WAITING_ON_BROKER; } DispatchMessage(m); queued_messages_.erase(queued_messages_.begin()); } return DISPATCH_FINISHED; } void ChannelReader::DispatchMessage(Message* m) { m->set_sender_pid(GetSenderPID()); #ifdef IPC_MESSAGE_LOG_ENABLED std::string name; Logging::GetInstance()->GetMessageText(m->type(), &name, m, NULL); TRACE_EVENT1("ipc,toplevel", "ChannelReader::DispatchInputData", "name", name); #else TRACE_EVENT2("ipc,toplevel", "ChannelReader::DispatchInputData", "class", IPC_MESSAGE_ID_CLASS(m->type()), "line", IPC_MESSAGE_ID_LINE(m->type())); #endif m->TraceMessageEnd(); bool handled = false; if (IsInternalMessage(*m)) { HandleInternalMessage(*m); handled = true; } #if USE_ATTACHMENT_BROKER if (!handled && IsAttachmentBrokerEndpoint() && GetAttachmentBroker()) { handled = GetAttachmentBroker()->OnMessageReceived(*m); } #endif // USE_ATTACHMENT_BROKER if (!handled) listener_->OnMessageReceived(*m); if (m->dispatch_error()) listener_->OnBadMessageReceived(*m); } ChannelReader::AttachmentIdSet ChannelReader::GetBrokeredAttachments( Message* msg) { std::set blocked_ids; #if USE_ATTACHMENT_BROKER MessageAttachmentSet* set = msg->attachment_set(); for (const scoped_refptr& attachment : set->GetBrokerableAttachmentsForUpdating()) { if (attachment->NeedsBrokering()) { AttachmentBroker* broker = GetAttachmentBroker(); scoped_refptr brokered_attachment; bool result = broker->GetAttachmentWithId(attachment->GetIdentifier(), &brokered_attachment); if (!result) { blocked_ids.insert(attachment->GetIdentifier()); continue; } attachment->PopulateWithAttachment(brokered_attachment.get()); } } #endif // USE_ATTACHMENT_BROKER return blocked_ids; } void ChannelReader::ReceivedBrokerableAttachmentWithId( const BrokerableAttachment::AttachmentId& id) { if (blocked_ids_.empty()) return; auto it = find(blocked_ids_.begin(), blocked_ids_.end(), id); if (it != blocked_ids_.end()) blocked_ids_.erase(it); if (blocked_ids_.empty()) { StopObservingAttachmentBroker(); DispatchMessages(); } } void ChannelReader::StartObservingAttachmentBroker() { #if USE_ATTACHMENT_BROKER GetAttachmentBroker()->AddObserver(this); #endif // USE_ATTACHMENT_BROKER } void ChannelReader::StopObservingAttachmentBroker() { #if USE_ATTACHMENT_BROKER GetAttachmentBroker()->RemoveObserver(this); #endif // USE_ATTACHMENT_BROKER } } // namespace internal } // namespace IPC