Mojo: Armed Watchers

mojo/edk/system/core.cc

 // Copyright 2013 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.
 
 #include "mojo/edk/system/core.h"
 
 #include <string.h>
 
 #include <utility>
 
@@ skipping 17 matching lines @@
 #include "mojo/edk/system/handle_signals_state.h"
 #include "mojo/edk/system/message_for_transit.h"
 #include "mojo/edk/system/message_pipe_dispatcher.h"
 #include "mojo/edk/system/platform_handle_dispatcher.h"
 #include "mojo/edk/system/ports/name.h"
 #include "mojo/edk/system/ports/node.h"
 #include "mojo/edk/system/request_context.h"
 #include "mojo/edk/system/shared_buffer_dispatcher.h"
 #include "mojo/edk/system/wait_set_dispatcher.h"
 #include "mojo/edk/system/waiter.h"
+#include "mojo/edk/system/watcher_dispatcher.h"
 
 namespace mojo {
 namespace edk {
 
 namespace {
 
 // This is an unnecessarily large limit that is relatively easy to enforce.
 const uint32_t kMaxHandlesPerMessage = 1024 * 1024;
 
 // TODO(rockot): Maybe we could negotiate a debugging pipe ID for cross-process
 // pipes too; for now we just use a constant. This only affects bootstrap pipes.
 const uint64_t kUnknownPipeIdForDebug = 0x7f7f7f7f7f7f7f7fUL;
 
-void CallWatchCallback(MojoWatchCallback callback,
-                       uintptr_t context,
-                       MojoResult result,
-                       const HandleSignalsState& signals_state,
-                       MojoWatchNotificationFlags flags) {
-  callback(context, result, static_cast<MojoHandleSignalsState>(signals_state),
-      flags);
-}
-
 MojoResult MojoPlatformHandleToScopedPlatformHandle(
     const MojoPlatformHandle* platform_handle,
     ScopedPlatformHandle* out_handle) {
   if (platform_handle->struct_size != sizeof(MojoPlatformHandle))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (platform_handle->type == MOJO_PLATFORM_HANDLE_TYPE_INVALID) {
     out_handle->reset();
     return MOJO_RESULT_OK;
   }
@@ skipping 350 matching lines @@
     // Note: The |reinterpret_cast| is safe, since |HandleSignalsState| is a
     // subclass of |MojoHandleSignalsState| that doesn't add any data members.
     rv = WaitManyInternal(handles, signals, num_handles, deadline, &index,
                           reinterpret_cast<HandleSignalsState*>(signals_state));
   }
   if (index != static_cast<uint32_t>(-1) && result_index)
     *result_index = index;
   return rv;
 }
 
-MojoResult Core::Watch(MojoHandle handle,
+MojoResult Core::CreateWatcher(MojoWatcherCallback callback,
+                               MojoHandle* watcher_handle) {
+  RequestContext request_context;
+  if (!watcher_handle)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  *watcher_handle = AddDispatcher(new WatcherDispatcher(callback));
+  if (*watcher_handle == MOJO_HANDLE_INVALID)
+    return MOJO_RESULT_RESOURCE_EXHAUSTED;
+  return MOJO_RESULT_OK;
+}
+
+MojoResult Core::Watch(MojoHandle watcher_handle,
+                       MojoHandle handle,
                        MojoHandleSignals signals,
-                       MojoWatchCallback callback,
                        uintptr_t context) {
   RequestContext request_context;
+  scoped_refptr<Dispatcher> watcher = GetDispatcher(watcher_handle);
+  if (!watcher || watcher->GetType() != Dispatcher::Type::WATCHER)
+    return MOJO_RESULT_INVALID_ARGUMENT;
   scoped_refptr<Dispatcher> dispatcher = GetDispatcher(handle);
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
-  return dispatcher->Watch(
-      signals, base::Bind(&CallWatchCallback, callback, context), context);
+  return watcher->WatchDispatcher(dispatcher, signals, context);
 }
 
-MojoResult Core::CancelWatch(MojoHandle handle, uintptr_t context) {
+MojoResult Core::CancelWatch(MojoHandle watcher_handle, uintptr_t context) {
   RequestContext request_context;
-  scoped_refptr<Dispatcher> dispatcher = GetDispatcher(handle);
-  if (!dispatcher)
+  scoped_refptr<Dispatcher> watcher = GetDispatcher(watcher_handle);
+  if (!watcher || watcher->GetType() != Dispatcher::Type::WATCHER)
     return MOJO_RESULT_INVALID_ARGUMENT;
-  return dispatcher->CancelWatch(context);
+  return watcher->CancelWatch(context);
+}
+
+MojoResult Core::ArmWatcher(MojoHandle watcher_handle,
+                            uint32_t* num_ready_contexts,
+                            uintptr_t* ready_contexts,
+                            MojoResult* ready_results,
+                            MojoHandleSignalsState* ready_signals_states) {
+  RequestContext request_context;
+  scoped_refptr<Dispatcher> watcher = GetDispatcher(watcher_handle);
+  if (!watcher || watcher->GetType() != Dispatcher::Type::WATCHER)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  return watcher->Arm(num_ready_contexts, ready_contexts, ready_results,
+                      ready_signals_states);
 }
 
 MojoResult Core::AllocMessage(uint32_t num_bytes,
                               const MojoHandle* handles,
                               uint32_t num_handles,
                               MojoAllocMessageFlags flags,
                               MojoMessageHandle* message) {
   if (!message)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
@@ skipping 711 matching lines @@
     std::unique_ptr<NodeController> node_controller) {
   // It's OK to leak this reference. At this point we know the IO loop is still
   // running, and we know the NodeController will observe its eventual
   // destruction. This tells the NodeController to delete itself when that
   // happens.
   node_controller.release()->DestroyOnIOThreadShutdown();
 }
 
 }  // namespace edk
 }  // namespace mojo