Factor out the shared parts of IPC channel reading.

ipc/ipc_channel_posix.cc

 // 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_posix.h"
 
 #include <errno.h>
 #include <fcntl.h>
 #include <stddef.h>
 #include <sys/types.h>
@@ skipping 282 matching lines @@
 
 }  // namespace
 //------------------------------------------------------------------------------
 
 #if defined(OS_LINUX)
 int Channel::ChannelImpl::global_pid_ = 0;
 #endif  // OS_LINUX
 
 Channel::ChannelImpl::ChannelImpl(const IPC::ChannelHandle& channel_handle,
                                   Mode mode, Listener* listener)
-    : mode_(mode),
+    : ChannelReader(listener),
+      mode_(mode),
       is_blocked_on_write_(false),
       waiting_connect_(true),
       message_send_bytes_written_(0),
       server_listen_pipe_(-1),
       pipe_(-1),
       client_pipe_(-1),
 #if defined(IPC_USES_READWRITE)
       fd_pipe_(-1),
       remote_fd_pipe_(-1),
 #endif  // IPC_USES_READWRITE
       pipe_name_(channel_handle.name),
-      listener_(listener),
       must_unlink_(false) {
-  memset(input_buf_, 0, sizeof(input_buf_));
   memset(input_cmsg_buf_, 0, sizeof(input_cmsg_buf_));
   if (!CreatePipe(channel_handle)) {
     // The pipe may have been closed already.
     const char *modestr = (mode_ & MODE_SERVER_FLAG) ? "server" : "client";
     LOG(WARNING) << "Unable to create pipe named \"" << channel_handle.name
                  << "\" in " << modestr << " mode";
   }
 }
 
 Channel::ChannelImpl::~ChannelImpl() {
@@ skipping 143 matching lines @@
         true,
         MessageLoopForIO::WATCH_READ,
         &server_listen_connection_watcher_,
         this);
   } else {
     did_connect = AcceptConnection();
   }
   return did_connect;
 }
 
-bool Channel::ChannelImpl::ProcessIncomingMessages() {
-  while (true) {
-    int bytes_read = 0;
-    ReadState read_state = ReadData(input_buf_, Channel::kReadBufferSize,
-                                    &bytes_read);
-    if (read_state == READ_FAILED)
-      return false;
-    if (read_state == READ_PENDING)
-      return true;
-
-    DCHECK(bytes_read > 0);
-    if (!DispatchInputData(input_buf_, bytes_read))
-      return false;
-  }
-}
-
-bool Channel::ChannelImpl::DispatchInputData(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() >
-        kMaximumMessageSize - input_data_len) {
-      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<int>(message_tail - p);
-      Message m(p, len);
-      if (!WillDispatchInputMessage(&m))
-        return false;
-
-      if (IsHelloMessage(&m))
-        HandleHelloMessage(m);
-      else
-        listener_->OnMessageReceived(m);
-      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;
-}
-
 bool Channel::ChannelImpl::ProcessOutgoingMessages() {
   DCHECK(!waiting_connect_);  // Why are we trying to send messages if there's
                               // no connection?
   if (output_queue_.empty())
     return true;
 
   if (pipe_ == -1)
     return false;
 
   // Write out all the messages we can till the write blocks or there are no
@@ skipping 41 matching lines @@
       cmsg->cmsg_len = CMSG_LEN(sizeof(int) * num_fds);
       msg->file_descriptor_set()->GetDescriptors(
           reinterpret_cast<int*>(CMSG_DATA(cmsg)));
       msgh.msg_controllen = cmsg->cmsg_len;
 
       // DCHECK_LE above already checks that
       // num_fds < kMaxDescriptorsPerMessage so no danger of overflow.
       msg->header()->num_fds = static_cast<uint16>(num_fds);
 
 #if defined(IPC_USES_READWRITE)
-      if (!IsHelloMessage(msg)) {
+      if (!IsHelloMessage(*msg)) {
         // Only the Hello message sends the file descriptor with the message.
         // Subsequently, we can send file descriptors on the dedicated
         // fd_pipe_ which makes Seccomp sandbox operation more efficient.
         struct iovec fd_pipe_iov = { const_cast<char *>(""), 1 };
         msgh.msg_iov = &fd_pipe_iov;
         fd_written = fd_pipe_;
         bytes_written = HANDLE_EINTR(sendmsg(fd_pipe_, &msgh, MSG_DONTWAIT));
         msgh.msg_iov = &iov;
         msgh.msg_controllen = 0;
         if (bytes_written > 0) {
           msg->file_descriptor_set()->CommitAll();
         }
       }
 #endif  // IPC_USES_READWRITE
     }
 
     if (bytes_written == 1) {
       fd_written = pipe_;
 #if defined(IPC_USES_READWRITE)
-      if ((mode_ & MODE_CLIENT_FLAG) && IsHelloMessage(msg)) {
+      if ((mode_ & MODE_CLIENT_FLAG) && IsHelloMessage(*msg)) {
         DCHECK_EQ(msg->file_descriptor_set()->size(), 1U);
       }
       if (!msgh.msg_controllen) {
         bytes_written = HANDLE_EINTR(write(pipe_, out_bytes, amt_to_write));
       } else
 #endif  // IPC_USES_READWRITE
       {
         bytes_written = HANDLE_EINTR(sendmsg(pipe_, &msgh, MSG_DONTWAIT));
       }
     }
@@ skipping 171 matching lines @@
 }
 #endif  // OS_LINUX
 
 // Called by libevent when we can read from the pipe without blocking.
 void Channel::ChannelImpl::OnFileCanReadWithoutBlocking(int fd) {
   bool send_server_hello_msg = false;
   if (fd == server_listen_pipe_) {
     int new_pipe = 0;
     if (!ServerAcceptConnection(server_listen_pipe_, &new_pipe)) {
       Close();
-      listener_->OnChannelListenError();
+      listener()->OnChannelListenError();
     }
 
     if (pipe_ != -1) {
       // We already have a connection. We only handle one at a time.
       // close our new descriptor.
       if (HANDLE_EINTR(shutdown(new_pipe, SHUT_RDWR)) < 0)
         DPLOG(ERROR) << "shutdown " << pipe_name_;
       if (HANDLE_EINTR(close(new_pipe)) < 0)
         DPLOG(ERROR) << "close " << pipe_name_;
-      listener_->OnChannelDenied();
+      listener()->OnChannelDenied();
       return;
     }
     pipe_ = new_pipe;
 
     if ((mode_ & MODE_OPEN_ACCESS_FLAG) == 0) {
       // Verify that the IPC channel peer is running as the same user.
       uid_t client_euid;
       if (!GetClientEuid(&client_euid)) {
         DLOG(ERROR) << "Unable to query client euid";
         ResetToAcceptingConnectionState();
@@ skipping 63 matching lines @@
     return true;
   } else {
     NOTREACHED();
     return false;
   }
 }
 
 void Channel::ChannelImpl::ClosePipeOnError() {
   if (HasAcceptedConnection()) {
     ResetToAcceptingConnectionState();
-    listener_->OnChannelError();
+    listener()->OnChannelError();
   } else {
     Close();
     if (AcceptsConnections()) {
-      listener_->OnChannelListenError();
+      listener()->OnChannelListenError();
     } else {
-      listener_->OnChannelError();
+      listener()->OnChannelError();
     }
   }
 }
 
 int Channel::ChannelImpl::GetHelloMessageProcId() {
   int pid = base::GetCurrentProcId();
 #if defined(OS_LINUX)
   // Our process may be in a sandbox with a separate PID namespace.
   if (global_pid_) {
     pid = global_pid_;
@@ skipping 16 matching lines @@
     if (!msg->WriteFileDescriptor(base::FileDescriptor(remote_fd_pipe_,
                                                        false))) {
       NOTREACHED() << "Unable to pickle hello message file descriptors";
     }
     DCHECK_EQ(msg->file_descriptor_set()->size(), 1U);
   }
 #endif  // IPC_USES_READWRITE
   output_queue_.push(msg.release());
 }
 
-bool Channel::ChannelImpl::IsHelloMessage(const Message* m) const {
-  return m->routing_id() == MSG_ROUTING_NONE && m->type() == HELLO_MESSAGE_TYPE;
-}
-
 Channel::ChannelImpl::ReadState Channel::ChannelImpl::ReadData(
     char* buffer,
     int buffer_len,
     int* bytes_read) {
   if (pipe_ == -1)
     return READ_FAILED;
 
   struct msghdr msg = {0};
 
   struct iovec iov = {buffer, buffer_len};
@@ skipping 58 matching lines @@
 
   if (bytes_received != 1)
     return true;  // No message waiting.
 
   if (!ExtractFileDescriptorsFromMsghdr(&msg))
     return false;
   return true;
 }
 #endif
 
+// On Posix, we need to fix up the file descriptors before the input message
+// is dispatched.
+//
+// This will read from the input_fds_ (READWRITE mode only) and read more
+// handles from the FD pipe if necessary.
 bool Channel::ChannelImpl::WillDispatchInputMessage(Message* msg) {
   uint16 header_fds = msg->header()->num_fds;
   if (!header_fds)
     return true;  // Nothing to do.
 
   // The message has file descriptors.
   const char* error = NULL;
   if (header_fds > input_fds_.size()) {
     // The message has been completely received, but we didn't get
     // enough file descriptors.
@@ skipping 90 matching lines @@
     // subsequent file descriptor passing.
     DCHECK_EQ(msg.file_descriptor_set()->size(), 1U);
     base::FileDescriptor descriptor;
     if (!msg.ReadFileDescriptor(&iter, &descriptor)) {
       NOTREACHED();
     }
     fd_pipe_ = descriptor.fd;
     CHECK(descriptor.auto_close);
   }
 #endif  // IPC_USES_READWRITE
-  listener_->OnChannelConnected(pid);
+  listener()->OnChannelConnected(pid);
 }
 
 void Channel::ChannelImpl::Close() {
   // Close can be called multiple time, so we need to make sure we're
   // idempotent.
 
   ResetToAcceptingConnectionState();
 
   if (must_unlink_) {
     unlink(pipe_name_.c_str());
@@ skipping 67 matching lines @@
 }
 
 #if defined(OS_LINUX)
 // static
 void Channel::SetGlobalPid(int pid) {
   ChannelImpl::SetGlobalPid(pid);
 }
 #endif  // OS_LINUX
 
 }  // namespace IPC