[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

mojo/edk/system/message_pipe_dispatcher.h

Index: mojo/edk/system/message_pipe_dispatcher.h
diff --git a/mojo/edk/system/message_pipe_dispatcher.h b/mojo/edk/system/message_pipe_dispatcher.h
index f898ea8a46fad05cd12b4981beac21cda110f9e1..fd7c5cd73647febdad6edb2a312e34d4e15d509e 100644
--- a/mojo/edk/system/message_pipe_dispatcher.h
+++ b/mojo/edk/system/message_pipe_dispatcher.h
@@ -14,6 +14,7 @@
 #include "mojo/edk/system/awakable_list.h"
 #include "mojo/edk/system/dispatcher.h"
 #include "mojo/edk/system/ports/port_ref.h"
+#include "mojo/edk/system/watcher_set.h"
 
 namespace mojo {
 namespace edk {
@@ -41,17 +42,24 @@ class MessagePipeDispatcher : public Dispatcher {
 
   // Dispatcher:
   Type GetType() const override;
-  MojoResult Close() override;
+  MojoResult Close(RequestContext* request_context) override;
+  MojoResult Watch(MojoHandleSignals signals,
+                   const WatchCallback& callback,
+                   uintptr_t context,
+                   RequestContext* request_context) override;
+  MojoResult CancelWatch(uintptr_t context) override;
   MojoResult WriteMessage(const void* bytes,
                           uint32_t num_bytes,
                           const DispatcherInTransit* dispatchers,
                           uint32_t num_dispatchers,
-                          MojoWriteMessageFlags flags) override;
+                          MojoWriteMessageFlags flags,
+                          RequestContext* request_context) override;
   MojoResult ReadMessage(void* bytes,
                          uint32_t* num_bytes,
                          MojoHandle* handles,
                          uint32_t* num_handles,
-                         MojoReadMessageFlags flags) override;
+                         MojoReadMessageFlags flags,
+                         RequestContext* request_context) override;
   HandleSignalsState GetHandleSignalsState() const override;
   MojoResult AddAwakable(Awakable* awakable,
                          MojoHandleSignals signals,
@@ -65,9 +73,9 @@ class MessagePipeDispatcher : public Dispatcher {
   bool EndSerialize(void* destination,
                     ports::PortName* ports,
                     PlatformHandle* handles) override;
-  bool BeginTransit() override;
-  void CompleteTransitAndClose() override;
-  void CancelTransit() override;
+  bool BeginTransit(RequestContext* request_context) override;
+  void CompleteTransitAndClose(RequestContext* request_context) override;
+  void CancelTransit(RequestContext* request_context) override;
 
   static scoped_refptr<Dispatcher> Deserialize(
       const void* data,
@@ -83,9 +91,10 @@ class MessagePipeDispatcher : public Dispatcher {
 
   ~MessagePipeDispatcher() override;
 
-  MojoResult CloseNoLock();
+  MojoResult CloseNoLock(RequestContext* request_context);
   HandleSignalsState GetHandleSignalsStateNoLock() const;
   void OnPortStatusChanged();
+  void NotifyObserversForStateChangeNoLock(RequestContext* request_context);
 
   // These are safe to access from any thread without locking.
   NodeController* const node_controller_;
@@ -103,6 +112,7 @@ class MessagePipeDispatcher : public Dispatcher {
   bool port_transferred_ = false;
   bool port_closed_ = false;
   AwakableList awakables_;
+  WatcherSet watchers_;
 
   DISALLOW_COPY_AND_ASSIGN(MessagePipeDispatcher);
 };