[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

mojo/edk/system/message_pipe_dispatcher.h

 // Copyright 2015 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.
 
 #ifndef MOJO_EDK_SYSTEM_MESSAGE_PIPE_DISPATCHER_H_
 #define MOJO_EDK_SYSTEM_MESSAGE_PIPE_DISPATCHER_H_
 
 #include <stdint.h>
 
 #include <queue>
 
 #include "base/macros.h"
 #include "base/memory/scoped_ptr.h"
 #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 {
 
 class NodeController;
 class PortsMessage;
 
 class MessagePipeDispatcher : public Dispatcher {
  public:
   // Constructs a MessagePipeDispatcher permanently tied to a specific port.
   // |connected| must indicate the state of the port at construction time; if
   // the port is initialized with a peer, |connected| must be true. Otherwise it
   // must be false.
   //
   // A MessagePipeDispatcher may not be transferred while in a disconnected
   // state, and one can never return to a disconnected once connected.
   //
   // |pipe_id| is a unique identifier which can be used to track pipe endpoints
   // as they're passed around. |endpoint| is either 0 or 1 and again is only
   // used for tracking pipes (one side is always 0, the other is always 1.)
   MessagePipeDispatcher(NodeController* node_controller,
                         const ports::PortRef& port,
                         uint64_t pipe_id,
                         int endpoint);
 
   // 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,
                          uintptr_t context,
                          HandleSignalsState* signals_state) override;
   void RemoveAwakable(Awakable* awakable,
                       HandleSignalsState* signals_state) override;
   void StartSerialize(uint32_t* num_bytes,
                       uint32_t* num_ports,
                       uint32_t* num_handles) override;
   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,
       size_t num_bytes,
       const ports::PortName* ports,
       size_t num_ports,
       PlatformHandle* handles,
       size_t num_handles);
 
  private:
   class PortObserverThunk;
   friend class PortObserverThunk;
 
   ~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_;
   const ports::PortRef port_;
   const uint64_t pipe_id_;
   const int endpoint_;
 
   // Guards access to all the fields below.
   mutable base::Lock signal_lock_;
 
   // This is not the same is |port_transferred_|. It's only held true between
   // BeginTransit() and Complete/CancelTransit().
   bool in_transit_ = false;
 
   bool port_transferred_ = false;
   bool port_closed_ = false;
   AwakableList awakables_;
+  WatcherSet watchers_;
 
   DISALLOW_COPY_AND_ASSIGN(MessagePipeDispatcher);
 };
 
 }  // namespace edk
 }  // namespace mojo
 
 #endif  // MOJO_EDK_SYSTEM_MESSAGE_PIPE_DISPATCHER_H_