[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

mojo/edk/system/message_pipe_dispatcher.cc

Index: mojo/edk/system/message_pipe_dispatcher.cc
diff --git a/mojo/edk/system/message_pipe_dispatcher.cc b/mojo/edk/system/message_pipe_dispatcher.cc
index c7b4a32da6bad0e0cbea48ddc8b3e60683a01966..ce35f646d8d81e5dfbb84955c01259ba4486615f 100644
--- a/mojo/edk/system/message_pipe_dispatcher.cc
+++ b/mojo/edk/system/message_pipe_dispatcher.cc
@@ -13,6 +13,7 @@
 #include "mojo/edk/system/core.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/macros.h"
 
 namespace mojo {
@@ -97,6 +98,27 @@ MojoResult MessagePipeDispatcher::Close() {
   return CloseNoLock();
 }
 
+MojoResult MessagePipeDispatcher::Watch(MojoHandleSignals signals,
+                                        const Watcher::WatchCallback& callback,
+                                        uintptr_t context) {
+  base::AutoLock lock(signal_lock_);
+
+  if (port_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return awakables_.AddWatcher(
+      signals, callback, context, GetHandleSignalsStateNoLock());
+}
+
+MojoResult MessagePipeDispatcher::CancelWatch(uintptr_t context) {
+  base::AutoLock lock(signal_lock_);
+
+  if (port_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return awakables_.RemoveWatcher(context);
+}
+
 MojoResult MessagePipeDispatcher::WriteMessage(
     const void* bytes,
     uint32_t num_bytes,
@@ -579,6 +601,8 @@ HandleSignalsState MessagePipeDispatcher::GetHandleSignalsStateNoLock() const {
 }
 
 void MessagePipeDispatcher::OnPortStatusChanged() {
+  RequestContext request_context;
+
   base::AutoLock lock(signal_lock_);
 
   // We stop observing our port as soon as it's transferred, but this can race