Mojo: Armed Watchers

mojo/edk/system/data_pipe_consumer_dispatcher.cc

 // Copyright 2013 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/data_pipe_consumer_dispatcher.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <algorithm>
@@ skipping 79 matching lines @@
 Dispatcher::Type DataPipeConsumerDispatcher::GetType() const {
   return Type::DATA_PIPE_CONSUMER;
 }
 
 MojoResult DataPipeConsumerDispatcher::Close() {
   base::AutoLock lock(lock_);
   DVLOG(1) << "Closing data pipe consumer " << pipe_id_;
   return CloseNoLock();
 }
 
-
-MojoResult DataPipeConsumerDispatcher::Watch(
+MojoResult DataPipeConsumerDispatcher::RegisterWatcher(
     MojoHandleSignals signals,
     const Watcher::WatchCallback& callback,
     uintptr_t context) {
   base::AutoLock lock(lock_);
 
   if (is_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return awakable_list_.AddWatcher(
-      signals, callback, context, GetHandleSignalsStateNoLock());
+  return watchers_.Add(signals, callback, context,
+                       GetHandleSignalsStateNoLock());
 }
 
-MojoResult DataPipeConsumerDispatcher::CancelWatch(uintptr_t context) {
+MojoResult DataPipeConsumerDispatcher::ArmWatcher(uintptr_t context) {
   base::AutoLock lock(lock_);
 
   if (is_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return awakable_list_.RemoveWatcher(context);
+  return watchers_.Arm(context, GetHandleSignalsStateNoLock());
+}
+
+MojoResult DataPipeConsumerDispatcher::UnregisterWatcher(uintptr_t context) {
+  base::AutoLock lock(lock_);
+
+  if (is_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return watchers_.Remove(context);
 }
 
 MojoResult DataPipeConsumerDispatcher::ReadData(void* elements,
                                                 uint32_t* num_bytes,
                                                 MojoReadDataFlags flags) {
   base::AutoLock lock(lock_);
   if (!shared_ring_buffer_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (in_two_phase_read_)
@@ skipping 130 matching lines @@
     bytes_available_ -= num_bytes_read;
 
     base::AutoUnlock unlock(lock_);
     NotifyRead(num_bytes_read);
   }
 
   in_two_phase_read_ = false;
   two_phase_max_bytes_read_ = 0;
 
   HandleSignalsState new_state = GetHandleSignalsStateNoLock();
-  if (!new_state.equals(old_state))
+  if (!new_state.equals(old_state)) {
     awakable_list_.AwakeForStateChange(new_state);
+    watchers_.NotifyState(new_state);
+  }
 
   return rv;
 }
 
 HandleSignalsState DataPipeConsumerDispatcher::GetHandleSignalsState() const {
   base::AutoLock lock(lock_);
   return GetHandleSignalsStateNoLock();
 }
 
 MojoResult DataPipeConsumerDispatcher::AddAwakable(
@@ skipping 170 matching lines @@
 
 MojoResult DataPipeConsumerDispatcher::CloseNoLock() {
   lock_.AssertAcquired();
   if (is_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
   is_closed_ = true;
   ring_buffer_mapping_.reset();
   shared_ring_buffer_ = nullptr;
 
   awakable_list_.CancelAll();
+  watchers_.NotifyClosed();
   if (!transferred_) {
     base::AutoUnlock unlock(lock_);
     node_controller_->ClosePort(control_port_);
   }
 
   return MOJO_RESULT_OK;
 }
 
 HandleSignalsState
 DataPipeConsumerDispatcher::GetHandleSignalsStateNoLock() const {
@@ skipping 85 matching lines @@
                  << m->num_bytes << " bytes were written. [control_port="
                  << control_port_.name() << "]";
 
         bytes_available_ += m->num_bytes;
       }
     } while (message);
   }
 
   if (peer_closed_ != was_peer_closed ||
       bytes_available_ != previous_bytes_available) {
-    awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+    HandleSignalsState state = GetHandleSignalsStateNoLock();
+    awakable_list_.AwakeForStateChange(state);
+    watchers_.NotifyState(state);
   }
 }
 
 }  // namespace edk
 }  // namespace mojo