EDK: Add TaskRunner and PlatformHandleWatcher to RawChannel.

mojo/edk/system/master_connection_manager.cc

 // Copyright 2015 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/master_connection_manager.h"
 
 #include <memory>
 #include <unordered_map>
 #include <utility>
 #include <vector>
@@ skipping 64 matching lines @@
 // |MasterConnectionManager::Helper| is not thread-safe, and must only be used
 // on its |owner_|'s private thread.
 class MasterConnectionManager::Helper final : public RawChannel::Delegate {
  public:
   Helper(MasterConnectionManager* owner,
          ProcessIdentifier process_identifier,
          embedder::SlaveInfo slave_info,
          ScopedPlatformHandle platform_handle);
   ~Helper() override;
 
-  void Init();
+  void Init(RefPtr<TaskRunner>&& task_runner,
+            PlatformHandleWatcher* platform_handle_watcher);
   embedder::SlaveInfo Shutdown();
 
  private:
   // |RawChannel::Delegate| methods:
   void OnReadMessage(const MessageInTransit::View& message_view,
                      std::unique_ptr<std::vector<ScopedPlatformHandle>>
                          platform_handles) override;
   void OnError(Error error) override;
 
   // Handles an error that's fatal to this object. Note that this probably
@@ skipping 15 matching lines @@
                                         ScopedPlatformHandle platform_handle)
     : owner_(owner),
       process_identifier_(process_identifier),
       slave_info_(slave_info),
       raw_channel_(RawChannel::Create(platform_handle.Pass())) {}
 
 MasterConnectionManager::Helper::~Helper() {
   DCHECK(!raw_channel_);
 }
 
-void MasterConnectionManager::Helper::Init() {
-  raw_channel_->Init(this);
+void MasterConnectionManager::Helper::Init(
+    RefPtr<TaskRunner>&& task_runner,
+    PlatformHandleWatcher* platform_handle_watcher) {
+  raw_channel_->Init(std::move(task_runner), platform_handle_watcher, this);
 }
 
 embedder::SlaveInfo MasterConnectionManager::Helper::Shutdown() {
   raw_channel_->Shutdown();
   raw_channel_.reset();
   return slave_info_;
 }
 
 void MasterConnectionManager::Helper::OnReadMessage(
     const MessageInTransit::View& message_view,
@@ skipping 189 matching lines @@
       process_connections_;  // "Owns" any valid platform handles.
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ProcessConnections);
 };
 
 // MasterConnectionManager -----------------------------------------------------
 
 MasterConnectionManager::MasterConnectionManager(
     embedder::PlatformSupport* platform_support)
     : ConnectionManager(platform_support),
-      master_process_delegate_(),
+      master_process_delegate_(nullptr),
+      private_thread_platform_handle_watcher_(nullptr),
       next_process_identifier_(kFirstSlaveProcessIdentifier) {
   connections_[kMasterProcessIdentifier] = new ProcessConnections();
 }
 
 MasterConnectionManager::~MasterConnectionManager() {
   DCHECK(!delegate_thread_task_runner_);
   DCHECK(!master_process_delegate_);
   DCHECK(!private_thread_);
   DCHECK(helpers_.empty());
   DCHECK(pending_connects_.empty());
 }
 
 void MasterConnectionManager::Init(
     RefPtr<TaskRunner>&& delegate_thread_task_runner,
     embedder::MasterProcessDelegate* master_process_delegate) {
   DCHECK(delegate_thread_task_runner);
   DCHECK(master_process_delegate);
   DCHECK(!delegate_thread_task_runner_);
   DCHECK(!master_process_delegate_);
   DCHECK(!private_thread_);
 
   delegate_thread_task_runner_ = std::move(delegate_thread_task_runner);
   master_process_delegate_ = master_process_delegate;
-  // TODO(vtl): We'll need to plumb this in further.
-  PlatformHandleWatcher* platform_handle_watcher = nullptr;
   private_thread_ = platform::CreateAndStartIOThread(
-      &private_thread_task_runner_, &platform_handle_watcher);
+      &private_thread_task_runner_, &private_thread_platform_handle_watcher_);
 }
 
 ProcessIdentifier MasterConnectionManager::AddSlave(
     embedder::SlaveInfo slave_info,
     ScopedPlatformHandle platform_handle) {
   // We don't really care if |slave_info| is non-null or not.
   DCHECK(platform_handle.is_valid());
   AssertNotOnPrivateThread();
 
   ProcessIdentifier slave_process_identifier;
@@ skipping 44 matching lines @@
   AssertNotOnPrivateThread();
   DCHECK(master_process_delegate_);
   DCHECK(private_thread_);
 
   // The |Stop()| will actually finish all posted tasks.
   private_thread_task_runner_->PostTask(
       [this]() { ShutdownOnPrivateThread(); });
   private_thread_->Stop();
   private_thread_.reset();
   private_thread_task_runner_ = nullptr;
+  private_thread_platform_handle_watcher_ = nullptr;
   DCHECK(helpers_.empty());
   DCHECK(pending_connects_.empty());
   master_process_delegate_ = nullptr;
   delegate_thread_task_runner_ = nullptr;
 }
 
 bool MasterConnectionManager::AllowConnect(
     const ConnectionIdentifier& connection_id) {
   AssertNotOnPrivateThread();
   return AllowConnectImpl(kMasterProcessIdentifier, connection_id);
@@ skipping 248 matching lines @@
     embedder::SlaveInfo slave_info,
     ScopedPlatformHandle platform_handle,
     ProcessIdentifier slave_process_identifier,
     AutoResetWaitableEvent* event) {
   DCHECK(platform_handle.is_valid());
   DCHECK(event);
   AssertOnPrivateThread();
 
   std::unique_ptr<Helper> helper(new Helper(
       this, slave_process_identifier, slave_info, platform_handle.Pass()));
-  helper->Init();
+  helper->Init(private_thread_task_runner_.Clone(),
+               private_thread_platform_handle_watcher_);
 
   DCHECK(helpers_.find(slave_process_identifier) == helpers_.end());
   helpers_[slave_process_identifier] = helper.release();
 
   DVLOG(1) << "Added slave process identifier " << slave_process_identifier;
   event->Signal();
 }
 
 void MasterConnectionManager::OnError(ProcessIdentifier process_identifier) {
   DCHECK_NE(process_identifier, kInvalidProcessIdentifier);
@@ skipping 43 matching lines @@
   DCHECK(!private_thread_task_runner_->RunsTasksOnCurrentThread());
 }
 
 void MasterConnectionManager::AssertOnPrivateThread() const {
   // This should only be called after |Init()| and before |Shutdown()|.
   DCHECK(private_thread_task_runner_->RunsTasksOnCurrentThread());
 }
 
 }  // namespace system
 }  // namespace mojo