[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

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>
 #include <limits>
 #include <utility>
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop/message_loop.h"
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_shared_buffer.h"
 #include "mojo/edk/system/core.h"
 #include "mojo/edk/system/data_pipe_control_message.h"
 #include "mojo/edk/system/node_controller.h"
 #include "mojo/edk/system/ports_message.h"
+#include "mojo/edk/system/request_context.h"
 #include "mojo/public/c/system/data_pipe.h"
 
 namespace mojo {
 namespace edk {
 
 namespace {
 
 struct MOJO_ALIGNAS(8) SerializedState {
   MojoCreateDataPipeOptions options;
   uint64_t pipe_id;
@@ skipping 45 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(
+    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());
+}
+
+MojoResult DataPipeConsumerDispatcher::CancelWatch(uintptr_t context) {
+  base::AutoLock lock(lock_);
+
+  if (is_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return awakable_list_.RemoveWatcher(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_)
     return MOJO_RESULT_BUSY;
 
@@ skipping 368 matching lines @@
   // to ignore.
   if (transferred_)
     return;
 
   DVLOG(1) << "Control port status changed for data pipe producer " << pipe_id_;
 
   UpdateSignalsStateNoLock();
 }
 
 void DataPipeConsumerDispatcher::UpdateSignalsStateNoLock() {
+  RequestContext request_context;
+
   lock_.AssertAcquired();
 
   bool was_peer_closed = peer_closed_;
   size_t previous_bytes_available = bytes_available_;
 
   ports::PortStatus port_status;
   if (node_controller_->node()->GetStatus(control_port_, &port_status) !=
           ports::OK ||
       !port_status.receiving_messages) {
     DVLOG(1) << "Data pipe consumer " << pipe_id_ << " is aware of peer closure"
@@ skipping 37 matching lines @@
   }
 
   if (peer_closed_ != was_peer_closed ||
       bytes_available_ != previous_bytes_available) {
     awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
   }
 }
 
 }  // namespace edk
 }  // namespace mojo