[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

mojo/edk/system/data_pipe_consumer_dispatcher.cc

Index: mojo/edk/system/data_pipe_consumer_dispatcher.cc
diff --git a/mojo/edk/system/data_pipe_consumer_dispatcher.cc b/mojo/edk/system/data_pipe_consumer_dispatcher.cc
index 1f6f07947736428784dfba2cba1cc6ea58eed517..2f1d80ae03c0bdba57077cdeac582a4b54b0afa3 100644
--- a/mojo/edk/system/data_pipe_consumer_dispatcher.cc
+++ b/mojo/edk/system/data_pipe_consumer_dispatcher.cc
@@ -86,6 +86,29 @@ MojoResult DataPipeConsumerDispatcher::Close() {
   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) {