mojo/public/cpp/system/watcher.cc

// 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 "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()) {
      // TODO(yzshen): Remove this notification. crbug.com/604762
      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(scoped_refptr<base::SingleThreadTaskRunner> runner)
    : task_runner_(std::move(runner)),
      is_default_task_runner_(task_runner_ ==
                              base::ThreadTaskRunnerHandle::Get()),
      weak_factory_(this) {
  DCHECK(task_runner_->BelongsToCurrentThread());
  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();
  // |result| may be MOJO_RESULT_INVALID_ARGUMENT if |handle_| has closed, but
  // OnHandleReady has not yet been called.
  DCHECK(result == MOJO_RESULT_INVALID_ARGUMENT || 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,
                                MojoWatchNotificationFlags flags) {
  // 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| through the Watcher API.
  // Current HandleWatcher users have no need for it, so it's omitted here.
  Watcher* watcher = reinterpret_cast<Watcher*>(context);
  if ((flags & MOJO_WATCH_NOTIFICATION_FLAG_FROM_SYSTEM) &&
      watcher->task_runner_->RunsTasksOnCurrentThread() &&
      watcher->is_default_task_runner_) {
    // System notifications will trigger from the task runner passed to
    // mojo::edk::InitIPCSupport(). In Chrome this happens to always be the
    // default task runner for the IO thread.
    watcher->OnHandleReady(result);
  } else {
    watcher->task_runner_->PostTask(
        FROM_HERE,
        base::Bind(&Watcher::OnHandleReady, watcher->weak_self_, result));
  }
}

}  // namespace mojo