EDK: Add TaskRunner and PlatformHandleWatcher to RawChannel.

mojo/edk/system/raw_channel.cc

 // 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 "mojo/edk/system/raw_channel.h"
 
 #include <string.h>
 
 #include <algorithm>
 #include <utility>
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop.h"
 #include "mojo/edk/system/message_in_transit.h"
 #include "mojo/edk/system/transport_data.h"
 
 using mojo::platform::PlatformHandle;
+using mojo::platform::PlatformHandleWatcher;
 using mojo::platform::ScopedPlatformHandle;
+using mojo::platform::TaskRunner;
 using mojo::util::MutexLocker;
+using mojo::util::RefPtr;
 
 namespace mojo {
 namespace system {
 
 const size_t kReadSize = 4096;
 
 // RawChannel::ReadBuffer ------------------------------------------------------
 
 RawChannel::ReadBuffer::ReadBuffer() : buffer_(kReadSize), num_valid_bytes_(0) {
 }
@@ skipping 120 matching lines @@
   Buffer buffer2 = {
       static_cast<const char*>(message->transport_data()->buffer()),
       transport_data_buffer_size};
   buffers->push_back(buffer2);
 }
 
 // RawChannel ------------------------------------------------------------------
 
 RawChannel::RawChannel()
     : message_loop_for_io_(nullptr),
+      io_watcher_(nullptr),
       delegate_(nullptr),
       set_on_shutdown_(nullptr),
       write_stopped_(false),
-      weak_ptr_factory_(this) {
-}
+      weak_ptr_factory_(this) {}
 
 RawChannel::~RawChannel() {
   DCHECK(!read_buffer_);
   DCHECK(!write_buffer_);
 
   // No need to take |write_mutex_| here -- if there are still weak pointers
   // outstanding, then we're hosed anyway (since we wouldn't be able to
   // invalidate them cleanly, since we might not be on the I/O thread).
   DCHECK(!weak_ptr_factory_.HasWeakPtrs());
 }
 
-void RawChannel::Init(Delegate* delegate) {
+void RawChannel::Init(RefPtr<TaskRunner>&& io_task_runner,
+                      PlatformHandleWatcher* io_watcher,
+                      Delegate* delegate) {
+  DCHECK(io_task_runner);
+  DCHECK(io_watcher);
   DCHECK(delegate);
 
   DCHECK(!delegate_);
   delegate_ = delegate;
+  DCHECK(!io_task_runner_);
+  io_task_runner_ = io_task_runner;
+  DCHECK(!io_watcher_);
+  io_watcher_ = io_watcher;
 
   CHECK_EQ(base::MessageLoop::current()->type(), base::MessageLoop::TYPE_IO);
   DCHECK(!message_loop_for_io_);
   message_loop_for_io_ =
       static_cast<base::MessageLoopForIO*>(base::MessageLoop::current());
 
   // No need to take the lock. No one should be using us yet.
   DCHECK(!read_buffer_);
   read_buffer_.reset(new ReadBuffer);
   DCHECK(!write_buffer_);
@@ skipping 316 matching lines @@
 
   write_stopped_ = true;
   write_buffer_->message_queue_.Clear();
   write_buffer_->platform_handles_offset_ = 0;
   write_buffer_->data_offset_ = 0;
   return false;
 }
 
 }  // namespace system
 }  // namespace mojo