[mojo-bindings] Use Watch API instead of MessagePumpMojo

mojo/public/cpp/system/watcher.cc

Index: mojo/public/cpp/system/watcher.cc
diff --git a/mojo/public/cpp/system/watcher.cc b/mojo/public/cpp/system/watcher.cc
new file mode 100644
index 0000000000000000000000000000000000000000..ad965ff04364b2280f1c367b8126de47d58d6cb7
--- /dev/null
+++ b/mojo/public/cpp/system/watcher.cc
@@ -0,0 +1,134 @@
+// Copyright 2016 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/public/cpp/system/watcher.h"
+
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/macros.h"
+#include "base/message_loop/message_loop.h"
+#include "base/thread_task_runner_handle.h"
+#include "mojo/public/c/system/functions.h"
+
+namespace mojo {
+
+class Watcher::MessageLoopObserver
+    : public base::MessageLoop::DestructionObserver {
+ public:
+  explicit MessageLoopObserver(Watcher* watcher) : watcher_(watcher) {
+    base::MessageLoop::current()->AddDestructionObserver(this);
+  }
+
+  ~MessageLoopObserver() override {
+    StopObservingIfNecessary();
+  }
+
+ private:
+  // base::MessageLoop::DestructionObserver:
+  void WillDestroyCurrentMessageLoop() override {
+    StopObservingIfNecessary();
+    if (watcher_->IsWatching())
+      watcher_->OnHandleReady(MOJO_RESULT_ABORTED);
+  }
+
+  void StopObservingIfNecessary() {
+    if (is_observing_) {
+      is_observing_ = false;
+      base::MessageLoop::current()->RemoveDestructionObserver(this);
+    }
+  }
+
+  bool is_observing_ = true;
+  Watcher* watcher_;
+
+  DISALLOW_COPY_AND_ASSIGN(MessageLoopObserver);
+};
+
+Watcher::Watcher()
+    : task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      weak_factory_(this) {
+  weak_self_ = weak_factory_.GetWeakPtr();
+}
+
+Watcher::~Watcher() {
+  if(IsWatching())
+    Cancel();
+}
+
+bool Watcher::IsWatching() const {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  return handle_.is_valid();
+}
+
+MojoResult Watcher::Start(Handle handle,
+                          MojoHandleSignals signals,
+                          const ReadyCallback& callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!IsWatching());
+  DCHECK(!callback.is_null());
+
+  message_loop_observer_.reset(new MessageLoopObserver(this));
+  callback_ = callback;
+  handle_ = handle;
+  MojoResult result = MojoWatch(handle_.value(), signals,
+                                &Watcher::CallOnHandleReady,
+                                reinterpret_cast<uintptr_t>(this));
+  if (result != MOJO_RESULT_OK) {
+    handle_.set_value(kInvalidHandleValue);
+    callback_.Reset();
+    message_loop_observer_.reset();
+    DCHECK(result == MOJO_RESULT_FAILED_PRECONDITION ||
+           result == MOJO_RESULT_INVALID_ARGUMENT);
+    return result;
+  }
+
+  return MOJO_RESULT_OK;
+}
+
+void Watcher::Cancel() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // The watch may have already been cancelled if the handle was closed.
+  if (!handle_.is_valid())
+    return;
+
+  MojoResult result =
+      MojoCancelWatch(handle_.value(), reinterpret_cast<uintptr_t>(this));
+  message_loop_observer_.reset();
+  DCHECK_EQ(result, MOJO_RESULT_OK);
+  handle_.set_value(kInvalidHandleValue);
+  callback_.Reset();
+}
+
+void Watcher::OnHandleReady(MojoResult result) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  ReadyCallback callback = callback_;
+  if (result == MOJO_RESULT_CANCELLED) {
+    message_loop_observer_.reset();
+    handle_.set_value(kInvalidHandleValue);
+    callback_.Reset();
+  }
+
+  // NOTE: It's legal for |callback| to delete |this|.
+  if (!callback.is_null())
+    callback.Run(result);
+}
+
+// static
+void Watcher::CallOnHandleReady(uintptr_t context,
+                                MojoResult result,
+                                MojoHandleSignalsState signals_state) {
+  // NOTE: It is safe to assume the Watcher still exists because this callback
+  // will never be run after the Watcher's destructor.
+  //
+  // TODO: Maybe we should also expose |signals_state| throught he Watcher API.
+  // Current HandleWatcher users have no need for it, so it's omitted here.
+  Watcher* watcher = reinterpret_cast<Watcher*>(context);
+  watcher->task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&Watcher::OnHandleReady, watcher->weak_self_, result));
+}
+
+}  // namespace mojo