[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

mojo/edk/system/data_pipe_producer_dispatcher.cc

Index: mojo/edk/system/data_pipe_producer_dispatcher.cc
diff --git a/mojo/edk/system/data_pipe_producer_dispatcher.cc b/mojo/edk/system/data_pipe_producer_dispatcher.cc
index cbbade8b0249dfc177cf6ccb83eb584791421d6b..634fb6814ae99f0b9fd84a60f60613d6b3e8ee38 100644
--- a/mojo/edk/system/data_pipe_producer_dispatcher.cc
+++ b/mojo/edk/system/data_pipe_producer_dispatcher.cc
@@ -20,6 +20,8 @@
 #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 {
@@ -85,6 +87,28 @@ MojoResult DataPipeProducerDispatcher::Close() {
   return CloseNoLock();
 }
 
+MojoResult DataPipeProducerDispatcher::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 DataPipeProducerDispatcher::CancelWatch(uintptr_t context) {
+  base::AutoLock lock(lock_);
+
+  if (is_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return awakable_list_.RemoveWatcher(context);
+}
+
 MojoResult DataPipeProducerDispatcher::WriteData(const void* elements,
                                                  uint32_t* num_bytes,
                                                  MojoWriteDataFlags flags) {
@@ -452,6 +476,8 @@ void DataPipeProducerDispatcher::OnPortStatusChanged() {
 }
 
 void DataPipeProducerDispatcher::UpdateSignalsStateNoLock() {
+  RequestContext request_context;
+
   lock_.AssertAcquired();
 
   bool was_peer_closed = peer_closed_;