Factor out the shared parts of IPC channel reading.

ipc/ipc_channel_win.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_win.h"
 
 #include <windows.h>
 
 #include "base/auto_reset.h"
 #include "base/bind.h"
@@ skipping 12 matching lines @@
   context.handler = channel;
 }
 
 Channel::ChannelImpl::State::~State() {
   COMPILE_ASSERT(!offsetof(Channel::ChannelImpl::State, context),
                  starts_with_io_context);
 }
 
 Channel::ChannelImpl::ChannelImpl(const IPC::ChannelHandle &channel_handle,
                                   Mode mode, Listener* listener)
-    : ALLOW_THIS_IN_INITIALIZER_LIST(input_state_(this)),
+    : ChannelReader(listener),
+      ALLOW_THIS_IN_INITIALIZER_LIST(input_state_(this)),
       ALLOW_THIS_IN_INITIALIZER_LIST(output_state_(this)),
       pipe_(INVALID_HANDLE_VALUE),
-      listener_(listener),
       waiting_connect_(mode & MODE_SERVER_FLAG),
       processing_incoming_(false),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {
   CreatePipe(channel_handle, mode);
 }
 
 Channel::ChannelImpl::~ChannelImpl() {
   Close();
 }
 
@@ skipping 59 matching lines @@
 
 
 Channel::ChannelImpl::ReadState Channel::ChannelImpl::ReadData(
     char* buffer,
     int buffer_len,
     int* /* bytes_read */) {
   if (INVALID_HANDLE_VALUE == pipe_)
     return READ_FAILED;
 
   DWORD bytes_read = 0;
-  BOOL ok = ReadFile(pipe_, input_buf_, Channel::kReadBufferSize,
+  BOOL ok = ReadFile(pipe_, buffer, buffer_len,
                      &bytes_read, &input_state_.context.overlapped);
   if (!ok) {
     DWORD err = GetLastError();
     if (err == ERROR_IO_PENDING) {
       input_state_.is_pending = true;
       return READ_PENDING;
     }
     LOG(ERROR) << "pipe error: " << err;
     return READ_FAILED;
   }
 
   // We could return READ_SUCCEEDED here. But the way that this code is
   // structured we instead go back to the message loop. Our completion port
   // will be signalled even in the "synchronously completed" state.
   //
   // This allows us to potentially process some outgoing messages and
   // interleave other work on this thread when we're getting hammered with
   // input messages. Potentially, this could be tuned to be more efficient
   // with some testing.
   input_state_.is_pending = true;
   return READ_PENDING;
 }
 
 bool Channel::ChannelImpl::WillDispatchInputMessage(Message* msg) {
+  // We don't need to do anything here.
   return true;
 }
 
 void Channel::ChannelImpl::HandleHelloMessage(const Message& msg) {
   // The hello message contains one parameter containing the PID.
-  listener_->OnChannelConnected(MessageIterator(msg).NextInt());
+  listener()->OnChannelConnected(MessageIterator(msg).NextInt());
 }
 
 bool Channel::ChannelImpl::DidEmptyInputBuffers() {
+  // We don't need to do anything here.
   return true;
 }
 
-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::IsHelloMessage(const Message& m) const {
-  return m.routing_id() == MSG_ROUTING_NONE && m.type() == HELLO_MESSAGE_TYPE;
-}
-
-bool Channel::ChannelImpl::AsyncReadComplete(int bytes_read) {
-  return DispatchInputData(input_buf_, bytes_read);
-}
-
 // static
 const std::wstring Channel::ChannelImpl::PipeName(
     const std::string& channel_id) {
   std::string name("\\\\.\\pipe\\chrome.");
   return ASCIIToWide(name.append(channel_id));
 }
 
 bool Channel::ChannelImpl::CreatePipe(const IPC::ChannelHandle &channel_handle,
                                       Mode mode) {
   DCHECK_EQ(INVALID_HANDLE_VALUE, pipe_);
@@ skipping 133 matching lines @@
     // The pipe is being closed.
     return false;
   default:
     NOTREACHED();
     return false;
   }
 
   return true;
 }
 
-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::ProcessOutgoingMessages(
     MessageLoopForIO::IOContext* context,
     DWORD bytes_written) {
   DCHECK(!waiting_connect_);  // Why are we trying to send messages if there's
                               // no connection?
   DCHECK(thread_check_->CalledOnValidThread());
 
   if (output_state_.is_pending) {
     DCHECK(context);
     output_state_.is_pending = false;
@@ skipping 80 matching lines @@
     // Request more data.
     if (ok)
       ok = ProcessIncomingMessages();
   } else {
     DCHECK(context == &output_state_.context);
     ok = ProcessOutgoingMessages(context, bytes_transfered);
   }
   if (!ok && INVALID_HANDLE_VALUE != pipe_) {
     // We don't want to re-enter Close().
     Close();
-    listener_->OnChannelError();
+    listener()->OnChannelError();
   }
 }
 
 //------------------------------------------------------------------------------
 // Channel's methods simply call through to ChannelImpl.
 Channel::Channel(const IPC::ChannelHandle &channel_handle, Mode mode,
                  Listener* listener)
     : channel_impl_(new ChannelImpl(channel_handle, mode, listener)) {
 }
 
@@ skipping 16 matching lines @@
 bool Channel::Send(Message* message) {
   return channel_impl_->Send(message);
 }
 
 // static
 bool Channel::IsNamedServerInitialized(const std::string& channel_id) {
   return ChannelImpl::IsNamedServerInitialized(channel_id);
 }
 
 }  // namespace IPC