Implement Bus and ObjectProxy classes for our D-Bus library.

dbus/bus.cc

+// Copyright (c) 2011 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.
+//
+// TODO(satorux):
+// - Handle "disconnected" signal.
+// - Add support for signal sending
+// - Add support for signal monitoring
+// - Collect metrics (ex. # of method calls, method call time, etc.)
+
+#include "dbus/bus.h"
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/message_loop.h"
+#include "base/stl_util.h"
+#include "base/threading/thread.h"
+#include "base/threading/thread_restrictions.h"
+#include "dbus/error.h"
+#include "dbus/exported_object.h"
+#include "dbus/object_proxy.h"
+
+namespace dbus {
+
+namespace {
+
+// The class is used for watching the file descriptor used for D-Bus
+// communication.
+class Watch : public base::MessagePumpLibevent::Watcher {
+ public:
+  Watch(DBusWatch* watch)
+      : raw_watch_(watch) {
+    dbus_watch_set_data(raw_watch_, this, NULL);
+  }
+
+  ~Watch() {
+    dbus_watch_set_data(raw_watch_, NULL, NULL);
+  }
+
+  // Returns true if the underlying file descriptor is ready to be watched.
+  bool IsReadyToBeWatched() {
+    return dbus_watch_get_enabled(raw_watch_);
+  }
+
+  // Starts watching the underlying file descriptor.
+  void StartWatching() {
+    const int file_descriptor = dbus_watch_get_unix_fd(raw_watch_);
+    const int flags = dbus_watch_get_flags(raw_watch_);
+
+    MessageLoopForIO::Mode mode;
+    if ((flags & DBUS_WATCH_READABLE) && (flags & DBUS_WATCH_WRITABLE))
+      mode = MessageLoopForIO::WATCH_READ_WRITE;
+    else if (flags & DBUS_WATCH_READABLE)
+      mode = MessageLoopForIO::WATCH_READ;
+    else if (flags & DBUS_WATCH_WRITABLE)
+      mode = MessageLoopForIO::WATCH_WRITE;
+    else
+      NOTREACHED();
+
+    const bool persistent = true;  // Watch persistently.
+    const bool success = MessageLoopForIO::current()->WatchFileDescriptor(
+        file_descriptor,
+        persistent,
+        mode,
+        &file_descriptor_watcher_,
+        this);
+    CHECK(success) << "Unable to allocate memory";
+  }
+
+  // Stops watching the underlying file descriptor.
+  void StopWatching() {
+    file_descriptor_watcher_.StopWatchingFileDescriptor();
+  }
+
+ private:
+  // Implement MessagePumpLibevent::Watcher.
+  virtual void OnFileCanReadWithoutBlocking(int file_descriptor) {
+    const bool success = dbus_watch_handle(raw_watch_, DBUS_WATCH_READABLE);
+    CHECK(success) << "Unable to allocate memory";
+  }
+
+  // Implement MessagePumpLibevent::Watcher.
+  virtual void OnFileCanWriteWithoutBlocking(int file_descriptor) {
+    const bool success = dbus_watch_handle(raw_watch_, DBUS_WATCH_WRITABLE);
+    CHECK(success) << "Unable to allocate memory";
+  }
+
+  DBusWatch* raw_watch_;
+  base::MessagePumpLibevent::FileDescriptorWatcher file_descriptor_watcher_;
+};
+
+// The class is used for monitoring the timeout used for D-Bus method
+// calls.
+//
+// Unlike Watch, Timeout is a ref counted object, to ensure that |this| of
+// the object is is alive when HandleTimeout() is called. It's unlikely
+// but it may be possible that HandleTimeout() is called after
+// Bus::OnRemoveTimeout(). Hence why we don't simply delete the object in
+// Bus::OnRemoveTimeout().
+class Timeout : public base::RefCountedThreadSafe<Timeout> {
+ public:
+  Timeout(DBusTimeout* timeout)
+      : raw_timeout_(timeout),
+        monitoring_is_active_(false),
+        is_destroyed_(false) {
+    dbus_timeout_set_data(raw_timeout_, this, NULL);
+  }
+
+  // Returns true if the timeout is ready to be monitored.
+  bool IsReadyToBeMonitored() {
+    return dbus_timeout_get_enabled(raw_timeout_);
+  }
+
+  // Starts monitoring the timeout.
+  void StartMonitoring(dbus::Bus* bus) {
+    bus->PostDelayedTaskToDBusThread(FROM_HERE,
+                                     base::Bind(&Timeout::HandleTimeout,
+                                                this),
+                                     GetIntervalInMs());
+    monitoring_is_active_ = true;
+  }
+
+  // Stops monitoring the timeout.
+  void StopMonitoring() {
+    // We cannot take back the delayed task we posted in
+    // StartMonitoring(), so we just mark the monitoring is inactive now.
+    monitoring_is_active_ = false;
+  }
+
+  // Returns the interval in milliseconds.
+  int GetIntervalInMs() {
+    return dbus_timeout_get_interval(raw_timeout_);
+  }
+
+  // Cleans up the raw_timeout and marks that this object is destroyed.
+  // See the class comment above for why we are doing this.
+  void Destroy() {

[stevenjb, 2011/08/10 19:40:07]

"Destroy" and "is_destroyed_" is a bit confusing, since we're not actually
delsting/destroying the object.  "Complete" and "is_completed_" might be better.

[satorux1, 2011/08/10 21:14:27]

You are right. Changed accordingly.

+    dbus_timeout_set_data(raw_timeout_, NULL, NULL);
+    is_destroyed_ = true;
+  }
+
+ private:
+  friend class base::RefCountedThreadSafe<Timeout>;
+  ~Timeout() {
+  }
+
+  // Handles the timeout.
+  void HandleTimeout() {
+    // If the object is marked destroyed, we should do nothing. This can
+    // occur if this function is called after Bus::OnRemoveTimeout().
+    if (is_destroyed_)
+      return;
+    // Skip if monitoring is cancled.
+    if (!monitoring_is_active_)
+      return;
+
+    const bool success = dbus_timeout_handle(raw_timeout_);
+    CHECK(success) << "Unable to allocate memory";
+  }
+
+  DBusTimeout* raw_timeout_;
+  bool monitoring_is_active_;
+  bool is_destroyed_;
+};
+
+}  // namespace
+
+Bus::Options::Options()
+  : bus_type(SESSION),
+    connection_type(PRIVATE),
+    dbus_thread(NULL) {
+}
+
+Bus::Options::~Options() {
+}
+
+Bus::Bus(const Options& options)
+    : bus_type_(options.bus_type),
+      connection_type_(options.connection_type),
+      dbus_thread_(options.dbus_thread),
+      connection_(NULL),
+      origin_loop_(MessageLoop::current()),
+      origin_thread_id_(base::PlatformThread::CurrentId()),
+      dbus_thread_id_(base::kInvalidThreadId),
+      async_operations_are_set_up_(false) {
+  if (dbus_thread_) {
+    dbus_thread_id_ = dbus_thread_->thread_id();
+    DCHECK(dbus_thread_->IsRunning())
+        << "The D-Bus thread should be running";
+    DCHECK_EQ(MessageLoop::TYPE_IO,
+              dbus_thread_->message_loop()->type())
+        << "The D-Bus thread should have an MessageLoopForIO attached";
+  }
+
+  // This is safe to call multiple times.
+  dbus_threads_init_default();
+}
+
+Bus::~Bus() {
+  DCHECK(!connection_);
+  DCHECK(owned_service_names_.empty());
+  for (size_t i = 0; i < exported_objects_.size(); ++i)
+    exported_objects_[i]->Release();
+  exported_objects_.clear();
+
+  for (size_t i = 0; i < object_proxies_.size(); ++i)
+    object_proxies_[i]->Release();
+  object_proxies_.clear();
+}
+
+ObjectProxy* Bus::GetObjectProxy(const std::string& service_name,
+                                 const std::string& object_path) {
+  AssertOnOriginThread();
+
+  ObjectProxy* object_proxy =
+      new ObjectProxy(this, service_name, object_path);
+  object_proxy->AddRef();
+  object_proxies_.push_back(object_proxy);

[stevenjb, 2011/08/10 19:40:07]

Rather than calling AddRef and Release manually, shouldn't object_proxies_ just
be a list of scoped_refptr<ObjectProxy>?

[satorux1, 2011/08/10 21:14:27]

Good point. I thought vector<scoped_refptr> wouldn't compile just like
vector<scoped_ptr> does not. However, it turned out that vector<scoped_refptr>
was widely used in Chrome. Changed accordingly.

+
+  return object_proxy;
+}
+
+ExportedObject* Bus::GetExportedObject(const std::string& service_name,
+                                       const std::string& object_path) {
+  AssertOnOriginThread();
+
+  ExportedObject* exported_object =
+      new ExportedObject(this, service_name, object_path);
+  exported_object->AddRef();
+  exported_objects_.push_back(exported_object);

[stevenjb, 2011/08/10 19:40:07]

scoped_refptr<> here also?

[satorux1, 2011/08/10 21:14:27]

Done.

+
+  return exported_object;
+}
+
+bool Bus::Connect() {
+  // dbus_bus_get_private() and dbus_bus_get() are blocking calls.
+  AssertOnDBusThread();
+
+  // Check if it's already initialized.
+  if (connection_)
+    return true;
+
+  ScopedDBusError error;
+  const DBusBusType dbus_bus_type = static_cast<DBusBusType>(bus_type_);
+  if (connection_type_ == PRIVATE) {
+    connection_ = dbus_bus_get_private(dbus_bus_type, error.get());
+  } else {
+    connection_ = dbus_bus_get(dbus_bus_type, error.get());
+  }
+  if (!connection_) {
+    LOG(ERROR) << "Failed to connect to the bus: "
+               << (dbus_error_is_set(error.get()) ? error.message() : "");
+    return false;
+  }
+  // We shouldn't exit on the disconnected signal.
+  dbus_connection_set_exit_on_disconnect(connection_, false);
+
+  return true;
+}
+
+void Bus::ShutdownAndBlock() {
+  AssertOnDBusThread();
+
+  // Delete exported objects. Need to unregister them beforehand.
+  for (size_t i = 0; i < exported_objects_.size(); ++i) {
+    exported_objects_[i]->Unregister();
+  }
+
+  // Release all service names. We should not use an iterator here, as
+  // we'll modify the set in ReleaseOwnership().
+  while (!owned_service_names_.empty()) {
+    ReleaseOwnership(*owned_service_names_.begin());
+  }

[stevenjb, 2011/08/10 19:40:07]

ReleaseOwnership may fail to release a service, and will log an error but not
crash, causing this to spin infinitely.
A better way to do this would be:

for (std::set<std::string>::iterator iter = owner_service_names_.begin();
    iter != owner_service_names_.end(); ) {
  const std::string& service_name = *iter++;  // safely increments iter
  ReleaseOwnership(service_name);
}

if (!owner_service_names_.empty())
  // log an error and/or force release

[satorux1, 2011/08/10 21:14:27]

Very good catch! Changed accordingly.

+
+  // Private connection should be closed.
+  if (connection_ && connection_type_ == PRIVATE) {
+    dbus_connection_close(connection_);
+  }
+  // dbus_connection_close() won't unref.
+  dbus_connection_unref(connection_);
+
+  connection_ = NULL;
+}
+
+void Bus::Shutdown(OnShutdownCallback callback) {
+  AssertOnOriginThread();
+
+  PostTaskToDBusThread(FROM_HERE, base::Bind(&Bus::ShutdownInternal,
+                                             this,
+                                             callback));
+}
+
+bool Bus::RequestOwnership(const std::string& service_name) {
+  DCHECK(connection_);
+  // dbus_bus_request_name() is a blocking call.
+  AssertOnDBusThread();
+
+  // Check if we already own the service name.
+  if (owned_service_names_.find(service_name) != owned_service_names_.end()) {
+    return true;
+  }
+
+  ScopedDBusError error;
+  const int result = dbus_bus_request_name(connection_,
+                                           service_name.c_str(),
+                                           DBUS_NAME_FLAG_DO_NOT_QUEUE,
+                                           error.get());
+  if (result != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER) {
+    LOG(ERROR) << "failed to get the onwership of " << service_name << ": "
+               << (dbus_error_is_set(error.get()) ? error.message() : "");
+    return false;
+  }
+  owned_service_names_.insert(service_name);
+  return true;
+}
+
+bool Bus::ReleaseOwnership(const std::string& service_name) {
+  DCHECK(connection_);
+  // dbus_bus_request_name() is a blocking call.
+  AssertOnDBusThread();
+
+  // Check if we already own the service name.
+  if (owned_service_names_.find(service_name) == owned_service_names_.end()) {
+    LOG(ERROR) << service_name << "is not owned by the bus";
+    return false;
+  }
+
+  ScopedDBusError error;
+  const int result = dbus_bus_release_name(connection_, service_name.c_str(),
+                                           error.get());
+  if (result == DBUS_RELEASE_NAME_REPLY_RELEASED) {
+    owned_service_names_.erase(service_name);
+    return true;
+  } else {
+    LOG(ERROR) << "failed to release the onwership of " << service_name << ": "
+               << (error.is_set() ? error.message() : "");
+    return false;
+  }
+}
+
+bool Bus::SetUpAsyncOperations() {
+  DCHECK(connection_);
+  AssertOnDBusThread();
+
+  if (async_operations_are_set_up_)
+    return true;
+
+  // Process all the incoming data if any, so that OnDispatchStatus() will
+  // be called when the incoming data is ready.
+  ProcessAllIncomingDataIfAny();
+
+  bool success = dbus_connection_set_watch_functions(connection_,
+                                                     &Bus::OnAddWatchThunk,
+                                                     &Bus::OnRemoveWatchThunk,
+                                                     &Bus::OnToggleWatchThunk,
+                                                     this,
+                                                     NULL);
+  CHECK(success) << "Unable to allocate memory";
+
+  // TODO(satorux): Timeout is not yet implemented.
+  success = dbus_connection_set_timeout_functions(connection_,
+                                                  &Bus::OnAddTimeoutThunk,
+                                                  &Bus::OnRemoveTimeoutThunk,
+                                                  &Bus::OnToggleTimeoutThunk,
+                                                  this,
+                                                  NULL);
+  CHECK(success) << "Unable to allocate memory";
+
+  dbus_connection_set_dispatch_status_function(
+      connection_,
+      &Bus::OnDispatchStatusChangedThunk,
+      this,
+      NULL);
+
+  async_operations_are_set_up_ = true;
+
+  return true;
+}
+
+DBusMessage* Bus::SendWithReplyAndBlock(DBusMessage* request,
+                                        int timeout_ms,
+                                        DBusError* error) {
+  DCHECK(connection_);
+  AssertOnDBusThread();
+
+  return dbus_connection_send_with_reply_and_block(
+      connection_, request, timeout_ms, error);
+}
+
+void Bus::SendWithReply(DBusMessage* request,
+                        DBusPendingCall** pending_call,
+                        int timeout_ms) {
+  DCHECK(connection_);
+  AssertOnDBusThread();
+
+  const bool success = dbus_connection_send_with_reply(
+      connection_, request, pending_call, timeout_ms);
+  CHECK(success) << "Unable to allocate memory";
+}
+
+bool Bus::TryRegisterObjectPath(const std::string& object_path,
+                                const DBusObjectPathVTable* vtable,
+                                void* user_data,
+                                DBusError* error) {
+  DCHECK(connection_);
+  AssertOnDBusThread();
+
+  return dbus_connection_try_register_object_path(
+      connection_,
+      object_path.c_str(),
+      vtable,
+      user_data,
+      error);
+}
+
+bool Bus::UnregisterObjectPath(const std::string& object_path) {
+  DCHECK(connection_);
+  AssertOnDBusThread();
+
+  return dbus_connection_unregister_object_path(
+      connection_,
+      object_path.c_str());
+}
+
+void Bus::ShutdownInternal(OnShutdownCallback callback) {
+  AssertOnDBusThread();
+
+  ShutdownAndBlock();
+  PostTaskToOriginThread(FROM_HERE, callback);
+}
+
+void Bus::ProcessAllIncomingDataIfAny() {
+  AssertOnDBusThread();
+
+  // As mentioned at the class comment in .h file, connection_ can be NULL.
+  if (!connection_ || !dbus_connection_get_is_connected(connection_))
+    return;
+
+  if (dbus_connection_get_dispatch_status(connection_) ==
+      DBUS_DISPATCH_DATA_REMAINS) {
+    while (dbus_connection_dispatch(connection_) ==
+           DBUS_DISPATCH_DATA_REMAINS);
+  }
+}
+
+void Bus::PostTaskToOriginThread(const tracked_objects::Location& from_here,
+                                 const base::Closure& task) {
+  origin_loop_->PostTask(from_here, task);
+}
+
+void Bus::PostTaskToDBusThread(const tracked_objects::Location& from_here,
+                               const base::Closure& task) {
+  if (dbus_thread_)
+    dbus_thread_->message_loop()->PostTask(from_here, task);
+  else
+    origin_loop_->PostTask(from_here, task);
+}
+
+void Bus::PostDelayedTaskToDBusThread(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    int delay_ms) {
+  if (dbus_thread_)
+    dbus_thread_->message_loop()->PostDelayedTask(from_here, task, delay_ms);
+  else
+    origin_loop_->PostDelayedTask(from_here, task, delay_ms);
+}
+
+void Bus::AssertOnOriginThread() {
+  DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
+}
+
+void Bus::AssertOnDBusThread() {
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  if (dbus_thread_) {
+    DCHECK_EQ(dbus_thread_id_, base::PlatformThread::CurrentId());
+  } else {
+    AssertOnOriginThread();
+  }
+}
+
+dbus_bool_t Bus::OnAddWatch(DBusWatch* raw_watch) {
+  AssertOnDBusThread();
+
+  // watch will be deleted when raw_watch is removed in OnRemoveWatch().

[stevenjb, 2011/08/10 19:40:07]

Is OnRemoveWatch() guaranteed to be called, even if StartWatching() never is?

[satorux1, 2011/08/10 21:14:27]

I think it's guaranteed, but I wasn't able to find the documentation saying it
so. So far, valgrind hasn't detected any leaks from the Watch object.

Anyway, I think your point is valid. I've added a counter to make sure that # of
AddWatch and # of RemoveWatch are balanced. I also did the same for Timeout.

+  Watch* watch = new Watch(raw_watch);
+  if (watch->IsReadyToBeWatched()) {
+    watch->StartWatching();
+  }
+  return true;
+}
+
+void Bus::OnRemoveWatch(DBusWatch* raw_watch) {
+  AssertOnDBusThread();
+
+  Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
+  delete watch;
+}
+
+void Bus::OnToggleWatch(DBusWatch* raw_watch) {
+  AssertOnDBusThread();
+
+  Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
+  if (watch->IsReadyToBeWatched()) {
+    watch->StartWatching();
+  } else {
+    // It's safe to call this if StartWatching() wasn't called, per
+    // message_pump_libevent.h.
+    watch->StopWatching();
+  }
+}
+
+dbus_bool_t Bus::OnAddTimeout(DBusTimeout* raw_timeout) {
+  AssertOnDBusThread();
+
+  // timeout will be deleted when raw_timeout is removed in
+  // OnRemoveTimeoutThunk().
+  Timeout* timeout = new Timeout(raw_timeout);
+  timeout->AddRef();  // Balanced on OnRemoveTimeout().

[stevenjb, 2011/08/10 19:40:07]

It might be more clear to put AddRef() in Timeout() and Release() in
Destroy/Complete()

[satorux1, 2011/08/10 21:14:27]

Agreed. Done.

+  if (timeout->IsReadyToBeMonitored()) {
+    timeout->StartMonitoring(this);
+  }
+  return true;
+}
+
+void Bus::OnRemoveTimeout(DBusTimeout* raw_timeout) {
+  AssertOnDBusThread();
+
+  Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
+  timeout->Destroy();
+  timeout->Release();
+}
+
+void Bus::OnToggleTimeout(DBusTimeout* raw_timeout) {
+  AssertOnDBusThread();
+
+  Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
+  if (timeout->IsReadyToBeMonitored()) {
+    timeout->StartMonitoring(this);
+  } else {
+    timeout->StopMonitoring();
+  }
+}
+
+void Bus::OnDispatchStatusChanged(DBusConnection* connection,
+                                  DBusDispatchStatus status) {
+  DCHECK_EQ(connection, connection_);
+  AssertOnDBusThread();
+
+  if (!dbus_connection_get_is_connected(connection))
+    return;
+
+  // We cannot call ProcessAllIncomingDataIfAny() here, as calling
+  // dbus_connection_dispatch() inside DBusDispatchStatusFunction is
+  // prohibited by the D-Bus library. Hence, we post a task here instead.
+  // See comments for dbus_connection_set_dispatch_status_function().
+  PostTaskToDBusThread(FROM_HERE,
+                       base::Bind(&Bus::ProcessAllIncomingDataIfAny,
+                                  this));
+}
+
+dbus_bool_t Bus::OnAddWatchThunk(DBusWatch* raw_watch, void* data) {
+  Bus* self = static_cast<Bus*>(data);
+  return self->OnAddWatch(raw_watch);
+}
+
+void Bus::OnRemoveWatchThunk(DBusWatch* raw_watch, void* data) {
+  Bus* self = static_cast<Bus*>(data);
+  return self->OnRemoveWatch(raw_watch);
+}
+
+void Bus::OnToggleWatchThunk(DBusWatch* raw_watch, void* data) {
+  Bus* self = static_cast<Bus*>(data);
+  return self->OnToggleWatch(raw_watch);
+}
+
+dbus_bool_t Bus::OnAddTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
+  Bus* self = static_cast<Bus*>(data);
+  return self->OnAddTimeout(raw_timeout);
+}
+
+void Bus::OnRemoveTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
+  Bus* self = static_cast<Bus*>(data);
+  return self->OnRemoveTimeout(raw_timeout);
+}
+
+void Bus::OnToggleTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
+  Bus* self = static_cast<Bus*>(data);
+  return self->OnToggleTimeout(raw_timeout);
+}
+
+void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
+                                       DBusDispatchStatus status,
+                                       void* data) {
+  Bus* self = static_cast<Bus*>(data);
+  return self->OnDispatchStatusChanged(connection, status);
+}
+
+}  // namespace dbus