dbus/bus.cc

// Copyright (c) 2012 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 "dbus/bus.h"

#include "base/bind.h"
#include "base/logging.h"
#include "base/message_loop/message_loop.h"
#include "base/message_loop/message_loop_proxy.h"
#include "base/stl_util.h"
#include "base/strings/stringprintf.h"
#include "base/threading/thread.h"
#include "base/threading/thread_restrictions.h"
#include "base/time/time.h"
#include "dbus/exported_object.h"
#include "dbus/message.h"
#include "dbus/object_manager.h"
#include "dbus/object_path.h"
#include "dbus/object_proxy.h"
#include "dbus/scoped_dbus_error.h"

namespace dbus {

namespace {

const char kDisconnectedSignal[] = "Disconnected";
const char kDisconnectedMatchRule[] =
    "type='signal', path='/org/freedesktop/DBus/Local',"
    "interface='org.freedesktop.DBus.Local', member='Disconnected'";

// The NameOwnerChanged member in org.freedesktop.DBus
const char kNameOwnerChangedSignal[] = "NameOwnerChanged";

// The match rule used to filter for changes to a given service name owner.
const char kServiceNameOwnerChangeMatchRule[] =
    "type='signal',interface='org.freedesktop.DBus',"
    "member='NameOwnerChanged',path='/org/freedesktop/DBus',"
    "sender='org.freedesktop.DBus',arg0='%s'";

// The class is used for watching the file descriptor used for D-Bus
// communication.
class Watch : public base::MessagePumpLibevent::Watcher {
 public:
  explicit Watch(DBusWatch* watch)
      : raw_watch_(watch) {
    dbus_watch_set_data(raw_watch_, this, NULL);
  }

  virtual ~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_);

    base::MessageLoopForIO::Mode mode = base::MessageLoopForIO::WATCH_READ;
    if ((flags & DBUS_WATCH_READABLE) && (flags & DBUS_WATCH_WRITABLE))
      mode = base::MessageLoopForIO::WATCH_READ_WRITE;
    else if (flags & DBUS_WATCH_READABLE)
      mode = base::MessageLoopForIO::WATCH_READ;
    else if (flags & DBUS_WATCH_WRITABLE)
      mode = base::MessageLoopForIO::WATCH_WRITE;
    else
      NOTREACHED();

    const bool persistent = true;  // Watch persistently.
    const bool success = base::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) OVERRIDE {
    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) OVERRIDE {
    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(). That's why we don't simply delete the object in
// Bus::OnRemoveTimeout().
class Timeout : public base::RefCountedThreadSafe<Timeout> {
 public:
  explicit Timeout(DBusTimeout* timeout)
      : raw_timeout_(timeout),
        monitoring_is_active_(false),
        is_completed(false) {
    dbus_timeout_set_data(raw_timeout_, this, NULL);
    AddRef();  // Balanced on Complete().
  }

  // 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(Bus* bus) {
    bus->GetDBusTaskRunner()->PostDelayedTask(
        FROM_HERE,
        base::Bind(&Timeout::HandleTimeout, this),
        GetInterval());
    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.
  base::TimeDelta GetInterval() {
    return base::TimeDelta::FromMilliseconds(
        dbus_timeout_get_interval(raw_timeout_));
  }

  // Cleans up the raw_timeout and marks that timeout is completed.
  // See the class comment above for why we are doing this.
  void Complete() {
    dbus_timeout_set_data(raw_timeout_, NULL, NULL);
    is_completed = true;
    Release();
  }

 private:
  friend class base::RefCountedThreadSafe<Timeout>;
  ~Timeout() {
  }

  // Handles the timeout.
  void HandleTimeout() {
    // If the timeout is marked completed, we should do nothing. This can
    // occur if this function is called after Bus::OnRemoveTimeout().
    if (is_completed)
      return;
    // Skip if monitoring is canceled.
    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_completed;
};

}  // namespace

Bus::Options::Options()
  : bus_type(SESSION),
    connection_type(PRIVATE) {
}

Bus::Options::~Options() {
}

Bus::Bus(const Options& options)
    : bus_type_(options.bus_type),
      connection_type_(options.connection_type),
      dbus_task_runner_(options.dbus_task_runner),
      on_shutdown_(false /* manual_reset */, false /* initially_signaled */),
      connection_(NULL),
      origin_thread_id_(base::PlatformThread::CurrentId()),
      async_operations_set_up_(false),
      shutdown_completed_(false),
      num_pending_watches_(0),
      num_pending_timeouts_(0),
      address_(options.address),
      on_disconnected_closure_(options.disconnected_callback) {
  // This is safe to call multiple times.
  dbus_threads_init_default();
  // The origin message loop is unnecessary if the client uses synchronous
  // functions only.
  if (base::MessageLoop::current())
    origin_task_runner_ = base::MessageLoop::current()->message_loop_proxy();
}

Bus::~Bus() {
  DCHECK(!connection_);
  DCHECK(owned_service_names_.empty());
  DCHECK(match_rules_added_.empty());
  DCHECK(filter_functions_added_.empty());
  DCHECK(registered_object_paths_.empty());
  DCHECK_EQ(0, num_pending_watches_);
  // TODO(satorux): This check fails occasionally in browser_tests for tests
  // that run very quickly. Perhaps something does not have time to clean up.
  // Despite the check failing, the tests seem to run fine. crosbug.com/23416
  // DCHECK_EQ(0, num_pending_timeouts_);
}

ObjectProxy* Bus::GetObjectProxy(const std::string& service_name,
                                 const ObjectPath& object_path) {
  return GetObjectProxyWithOptions(service_name, object_path,
                                   ObjectProxy::DEFAULT_OPTIONS);
}

ObjectProxy* Bus::GetObjectProxyWithOptions(const std::string& service_name,
                                            const ObjectPath& object_path,
                                            int options) {
  AssertOnOriginThread();

  // Check if we already have the requested object proxy.
  const ObjectProxyTable::key_type key(service_name + object_path.value(),
                                       options);
  ObjectProxyTable::iterator iter = object_proxy_table_.find(key);
  if (iter != object_proxy_table_.end()) {
    return iter->second.get();
  }

  scoped_refptr<ObjectProxy> object_proxy =
      new ObjectProxy(this, service_name, object_path, options);
  object_proxy_table_[key] = object_proxy;

  return object_proxy.get();
}

bool Bus::RemoveObjectProxy(const std::string& service_name,
                            const ObjectPath& object_path,
                            const base::Closure& callback) {
  return RemoveObjectProxyWithOptions(service_name, object_path,
                                      ObjectProxy::DEFAULT_OPTIONS,
                                      callback);
}

bool Bus::RemoveObjectProxyWithOptions(const std::string& service_name,
                                       const ObjectPath& object_path,
                                       int options,
                                       const base::Closure& callback) {
  AssertOnOriginThread();

  // Check if we have the requested object proxy.
  const ObjectProxyTable::key_type key(service_name + object_path.value(),
                                       options);
  ObjectProxyTable::iterator iter = object_proxy_table_.find(key);
  if (iter != object_proxy_table_.end()) {
    scoped_refptr<ObjectProxy> object_proxy = iter->second;
    object_proxy_table_.erase(iter);
    // Object is present. Remove it now and Detach on the DBus thread.
    GetDBusTaskRunner()->PostTask(
        FROM_HERE,
        base::Bind(&Bus::RemoveObjectProxyInternal,
                   this, object_proxy, callback));
    return true;
  }
  return false;
}

void Bus::RemoveObjectProxyInternal(scoped_refptr<ObjectProxy> object_proxy,
                                    const base::Closure& callback) {
  AssertOnDBusThread();

  object_proxy.get()->Detach();

  GetOriginTaskRunner()->PostTask(FROM_HERE, callback);
}

ExportedObject* Bus::GetExportedObject(const ObjectPath& object_path) {
  AssertOnOriginThread();

  // Check if we already have the requested exported object.
  ExportedObjectTable::iterator iter = exported_object_table_.find(object_path);
  if (iter != exported_object_table_.end()) {
    return iter->second.get();
  }

  scoped_refptr<ExportedObject> exported_object =
      new ExportedObject(this, object_path);
  exported_object_table_[object_path] = exported_object;

  return exported_object.get();
}

void Bus::UnregisterExportedObject(const ObjectPath& object_path) {
  AssertOnOriginThread();

  // Remove the registered object from the table first, to allow a new
  // GetExportedObject() call to return a new object, rather than this one.
  ExportedObjectTable::iterator iter = exported_object_table_.find(object_path);
  if (iter == exported_object_table_.end())
    return;

  scoped_refptr<ExportedObject> exported_object = iter->second;
  exported_object_table_.erase(iter);

  // Post the task to perform the final unregistration to the D-Bus thread.
  // Since the registration also happens on the D-Bus thread in
  // TryRegisterObjectPath(), and the task runner we post to is a
  // SequencedTaskRunner, there is a guarantee that this will happen before any
  // future registration call.
  GetDBusTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::UnregisterExportedObjectInternal,
                 this, exported_object));
}

void Bus::UnregisterExportedObjectInternal(
    scoped_refptr<ExportedObject> exported_object) {
  AssertOnDBusThread();

  exported_object->Unregister();
}

ObjectManager* Bus::GetObjectManager(const std::string& service_name,
                                     const ObjectPath& object_path) {
  AssertOnOriginThread();

  // Check if we already have the requested object manager.
  const ObjectManagerTable::key_type key(service_name + object_path.value());
  ObjectManagerTable::iterator iter = object_manager_table_.find(key);
  if (iter != object_manager_table_.end()) {
    return iter->second.get();
  }

  scoped_refptr<ObjectManager> object_manager =
      new ObjectManager(this, service_name, object_path);
  object_manager_table_[key] = object_manager;

  return object_manager.get();
}

bool Bus::RemoveObjectManager(const std::string& service_name,
                              const ObjectPath& object_path,
                              const base::Closure& callback) {
  AssertOnOriginThread();
  DCHECK(!callback.is_null());

  const ObjectManagerTable::key_type key(service_name + object_path.value());
  ObjectManagerTable::iterator iter = object_manager_table_.find(key);
  if (iter == object_manager_table_.end())
    return false;

  // ObjectManager is present. Remove it now and CleanUp on the DBus thread.
  scoped_refptr<ObjectManager> object_manager = iter->second;
  object_manager_table_.erase(iter);

  GetDBusTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::RemoveObjectManagerInternal,
                 this, object_manager, callback));

  return true;
}

void Bus::RemoveObjectManagerInternal(
      scoped_refptr<dbus::ObjectManager> object_manager,
      const base::Closure& callback) {
  AssertOnDBusThread();
  DCHECK(object_manager.get());

  object_manager->CleanUp();

  // The ObjectManager has to be deleted on the origin thread since it was
  // created there.
  GetOriginTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::RemoveObjectManagerInternalHelper,
                 this, object_manager, callback));
}

void Bus::RemoveObjectManagerInternalHelper(
      scoped_refptr<dbus::ObjectManager> object_manager,
      const base::Closure& callback) {
  AssertOnOriginThread();
  DCHECK(object_manager.get());

  // Release the object manager and run the callback.
  object_manager = NULL;
  callback.Run();
}

void Bus::GetManagedObjects() {
  for (ObjectManagerTable::iterator iter = object_manager_table_.begin();
       iter != object_manager_table_.end(); ++iter) {
    iter->second->GetManagedObjects();
  }
}

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;
  if (bus_type_ == CUSTOM_ADDRESS) {
    if (connection_type_ == PRIVATE) {
      connection_ = dbus_connection_open_private(address_.c_str(), error.get());
    } else {
      connection_ = dbus_connection_open(address_.c_str(), error.get());
    }
  } else {
    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: "
               << (error.is_set() ? error.message() : "");
    return false;
  }

  if (bus_type_ == CUSTOM_ADDRESS) {
    // We should call dbus_bus_register here, otherwise unique name can not be
    // acquired. According to dbus specification, it is responsible to call
    // org.freedesktop.DBus.Hello method at the beging of bus connection to
    // acquire unique name. In the case of dbus_bus_get, dbus_bus_register is
    // called internally.
    if (!dbus_bus_register(connection_, error.get())) {
      LOG(ERROR) << "Failed to register the bus component: "
                 << (error.is_set() ? error.message() : "");
      return false;
    }
  }
  // We shouldn't exit on the disconnected signal.
  dbus_connection_set_exit_on_disconnect(connection_, false);

  // Watch Disconnected signal.
  AddFilterFunction(Bus::OnConnectionDisconnectedFilter, this);
  AddMatch(kDisconnectedMatchRule, error.get());

  return true;
}

void Bus::ClosePrivateConnection() {
  // dbus_connection_close is blocking call.
  AssertOnDBusThread();
  DCHECK_EQ(PRIVATE, connection_type_)
      << "non-private connection should not be closed";
  dbus_connection_close(connection_);
}

void Bus::ShutdownAndBlock() {
  AssertOnDBusThread();

  if (shutdown_completed_)
    return;  // Already shutdowned, just return.

  // Unregister the exported objects.
  for (ExportedObjectTable::iterator iter = exported_object_table_.begin();
       iter != exported_object_table_.end(); ++iter) {
    iter->second->Unregister();
  }

  // Release all service names.
  for (std::set<std::string>::iterator iter = owned_service_names_.begin();
       iter != owned_service_names_.end();) {
    // This is a bit tricky but we should increment the iter here as
    // ReleaseOwnership() may remove |service_name| from the set.
    const std::string& service_name = *iter++;
    ReleaseOwnership(service_name);
  }
  if (!owned_service_names_.empty()) {
    LOG(ERROR) << "Failed to release all service names. # of services left: "
               << owned_service_names_.size();
  }

  // Detach from the remote objects.
  for (ObjectProxyTable::iterator iter = object_proxy_table_.begin();
       iter != object_proxy_table_.end(); ++iter) {
    iter->second->Detach();
  }

  // Clean up the object managers.
  for (ObjectManagerTable::iterator iter = object_manager_table_.begin();
       iter != object_manager_table_.end(); ++iter) {
    iter->second->CleanUp();
  }

  // Release object proxies and exported objects here. We should do this
  // here rather than in the destructor to avoid memory leaks due to
  // cyclic references.
  object_proxy_table_.clear();
  exported_object_table_.clear();

  // Private connection should be closed.
  if (connection_) {
    // Remove Disconnected watcher.
    ScopedDBusError error;
    RemoveFilterFunction(Bus::OnConnectionDisconnectedFilter, this);
    RemoveMatch(kDisconnectedMatchRule, error.get());

    if (connection_type_ == PRIVATE)
      ClosePrivateConnection();
    // dbus_connection_close() won't unref.
    dbus_connection_unref(connection_);
  }

  connection_ = NULL;
  shutdown_completed_ = true;
}

void Bus::ShutdownOnDBusThreadAndBlock() {
  AssertOnOriginThread();
  DCHECK(dbus_task_runner_.get());

  GetDBusTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::ShutdownOnDBusThreadAndBlockInternal, this));

  // http://crbug.com/125222
  base::ThreadRestrictions::ScopedAllowWait allow_wait;

  // Wait until the shutdown is complete on the D-Bus thread.
  // The shutdown should not hang, but set timeout just in case.
  const int kTimeoutSecs = 3;
  const base::TimeDelta timeout(base::TimeDelta::FromSeconds(kTimeoutSecs));
  const bool signaled = on_shutdown_.TimedWait(timeout);
  LOG_IF(ERROR, !signaled) << "Failed to shutdown the bus";
}

void Bus::RequestOwnership(const std::string& service_name,
                           ServiceOwnershipOptions options,
                           OnOwnershipCallback on_ownership_callback) {
  AssertOnOriginThread();

  GetDBusTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::RequestOwnershipInternal,
                 this, service_name, options, on_ownership_callback));
}

void Bus::RequestOwnershipInternal(const std::string& service_name,
                                   ServiceOwnershipOptions options,
                                   OnOwnershipCallback on_ownership_callback) {
  AssertOnDBusThread();

  bool success = Connect();
  if (success)
    success = RequestOwnershipAndBlock(service_name, options);

  GetOriginTaskRunner()->PostTask(FROM_HERE,
                                  base::Bind(on_ownership_callback,
                                             service_name,
                                             success));
}

bool Bus::RequestOwnershipAndBlock(const std::string& service_name,
                                   ServiceOwnershipOptions options) {
  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(),
                                           options,
                                           error.get());
  if (result != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER) {
    LOG(ERROR) << "Failed to get the ownership of " << service_name << ": "
               << (error.is_set() ? 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.
  std::set<std::string>::iterator found =
      owned_service_names_.find(service_name);
  if (found == 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(found);
    return true;
  } else {
    LOG(ERROR) << "Failed to release the ownership of " << service_name << ": "
               << (error.is_set() ? error.message() : "")
               << ", result code: " << result;
    return false;
  }
}

bool Bus::SetUpAsyncOperations() {
  DCHECK(connection_);
  AssertOnDBusThread();

  if (async_operations_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";

  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_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";
}

void Bus::Send(DBusMessage* request, uint32* serial) {
  DCHECK(connection_);
  AssertOnDBusThread();

  const bool success = dbus_connection_send(connection_, request, serial);
  CHECK(success) << "Unable to allocate memory";
}

bool Bus::AddFilterFunction(DBusHandleMessageFunction filter_function,
                            void* user_data) {
  DCHECK(connection_);
  AssertOnDBusThread();

  std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
      std::make_pair(filter_function, user_data);
  if (filter_functions_added_.find(filter_data_pair) !=
      filter_functions_added_.end()) {
    VLOG(1) << "Filter function already exists: " << filter_function
            << " with associated data: " << user_data;
    return false;
  }

  const bool success = dbus_connection_add_filter(
      connection_, filter_function, user_data, NULL);
  CHECK(success) << "Unable to allocate memory";
  filter_functions_added_.insert(filter_data_pair);
  return true;
}

bool Bus::RemoveFilterFunction(DBusHandleMessageFunction filter_function,
                               void* user_data) {
  DCHECK(connection_);
  AssertOnDBusThread();

  std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
      std::make_pair(filter_function, user_data);
  if (filter_functions_added_.find(filter_data_pair) ==
      filter_functions_added_.end()) {
    VLOG(1) << "Requested to remove an unknown filter function: "
            << filter_function
            << " with associated data: " << user_data;
    return false;
  }

  dbus_connection_remove_filter(connection_, filter_function, user_data);
  filter_functions_added_.erase(filter_data_pair);
  return true;
}

void Bus::AddMatch(const std::string& match_rule, DBusError* error) {
  DCHECK(connection_);
  AssertOnDBusThread();

  std::map<std::string, int>::iterator iter =
      match_rules_added_.find(match_rule);
  if (iter != match_rules_added_.end()) {
    // The already existing rule's counter is incremented.
    iter->second++;

    VLOG(1) << "Match rule already exists: " << match_rule;
    return;
  }

  dbus_bus_add_match(connection_, match_rule.c_str(), error);
  match_rules_added_[match_rule] = 1;
}

bool Bus::RemoveMatch(const std::string& match_rule, DBusError* error) {
  DCHECK(connection_);
  AssertOnDBusThread();

  std::map<std::string, int>::iterator iter =
      match_rules_added_.find(match_rule);
  if (iter == match_rules_added_.end()) {
    LOG(ERROR) << "Requested to remove an unknown match rule: " << match_rule;
    return false;
  }

  // The rule's counter is decremented and the rule is deleted when reachs 0.
  iter->second--;
  if (iter->second == 0) {
    dbus_bus_remove_match(connection_, match_rule.c_str(), error);
    match_rules_added_.erase(match_rule);
  }
  return true;
}

bool Bus::TryRegisterObjectPath(const ObjectPath& object_path,
                                const DBusObjectPathVTable* vtable,
                                void* user_data,
                                DBusError* error) {
  DCHECK(connection_);
  AssertOnDBusThread();

  if (registered_object_paths_.find(object_path) !=
      registered_object_paths_.end()) {
    LOG(ERROR) << "Object path already registered: " << object_path.value();
    return false;
  }

  const bool success = dbus_connection_try_register_object_path(
      connection_,
      object_path.value().c_str(),
      vtable,
      user_data,
      error);
  if (success)
    registered_object_paths_.insert(object_path);
  return success;
}

void Bus::UnregisterObjectPath(const ObjectPath& object_path) {
  DCHECK(connection_);
  AssertOnDBusThread();

  if (registered_object_paths_.find(object_path) ==
      registered_object_paths_.end()) {
    LOG(ERROR) << "Requested to unregister an unknown object path: "
               << object_path.value();
    return;
  }

  const bool success = dbus_connection_unregister_object_path(
      connection_,
      object_path.value().c_str());
  CHECK(success) << "Unable to allocate memory";
  registered_object_paths_.erase(object_path);
}

void Bus::ShutdownOnDBusThreadAndBlockInternal() {
  AssertOnDBusThread();

  ShutdownAndBlock();
  on_shutdown_.Signal();
}

void Bus::ProcessAllIncomingDataIfAny() {
  AssertOnDBusThread();

  // As mentioned at the class comment in .h file, connection_ can be NULL.
  if (!connection_)
    return;

  // It is safe and necessary to call dbus_connection_get_dispatch_status even
  // if the connection is lost. Otherwise we will miss "Disconnected" signal.
  // (crbug.com/174431)
  if (dbus_connection_get_dispatch_status(connection_) ==
      DBUS_DISPATCH_DATA_REMAINS) {
    while (dbus_connection_dispatch(connection_) ==
           DBUS_DISPATCH_DATA_REMAINS) {
    }
  }
}

base::TaskRunner* Bus::GetDBusTaskRunner() {
  if (dbus_task_runner_.get())
    return dbus_task_runner_.get();
  else
    return GetOriginTaskRunner();
}

base::TaskRunner* Bus::GetOriginTaskRunner() {
  DCHECK(origin_task_runner_.get());
  return origin_task_runner_.get();
}

bool Bus::HasDBusThread() {
  return dbus_task_runner_.get() != NULL;
}

void Bus::AssertOnOriginThread() {
  DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
}

void Bus::AssertOnDBusThread() {
  base::ThreadRestrictions::AssertIOAllowed();

  if (dbus_task_runner_.get()) {
    DCHECK(dbus_task_runner_->RunsTasksOnCurrentThread());
  } else {
    AssertOnOriginThread();
  }
}

std::string Bus::GetServiceOwnerAndBlock(const std::string& service_name,
                                         GetServiceOwnerOption options) {
  AssertOnDBusThread();

  MethodCall get_name_owner_call("org.freedesktop.DBus", "GetNameOwner");
  MessageWriter writer(&get_name_owner_call);
  writer.AppendString(service_name);
  VLOG(1) << "Method call: " << get_name_owner_call.ToString();

  const ObjectPath obj_path("/org/freedesktop/DBus");
  if (!get_name_owner_call.SetDestination("org.freedesktop.DBus") ||
      !get_name_owner_call.SetPath(obj_path)) {
    if (options == REPORT_ERRORS)
      LOG(ERROR) << "Failed to get name owner.";
    return "";
  }

  ScopedDBusError error;
  DBusMessage* response_message =
      SendWithReplyAndBlock(get_name_owner_call.raw_message(),
                            ObjectProxy::TIMEOUT_USE_DEFAULT,
                            error.get());
  if (!response_message) {
    if (options == REPORT_ERRORS) {
      LOG(ERROR) << "Failed to get name owner. Got " << error.name() << ": "
                 << error.message();
    }
    return "";
  }

  scoped_ptr<Response> response(Response::FromRawMessage(response_message));
  MessageReader reader(response.get());

  std::string service_owner;
  if (!reader.PopString(&service_owner))
    service_owner.clear();
  return service_owner;
}

void Bus::GetServiceOwner(const std::string& service_name,
                          const GetServiceOwnerCallback& callback) {
  AssertOnOriginThread();

  GetDBusTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::GetServiceOwnerInternal, this, service_name, callback));
}

void Bus::GetServiceOwnerInternal(const std::string& service_name,
                                  const GetServiceOwnerCallback& callback) {
  AssertOnDBusThread();

  std::string service_owner;
  if (Connect())
    service_owner = GetServiceOwnerAndBlock(service_name, SUPPRESS_ERRORS);
  GetOriginTaskRunner()->PostTask(FROM_HERE,
                                  base::Bind(callback, service_owner));
}

void Bus::ListenForServiceOwnerChange(
    const std::string& service_name,
    const GetServiceOwnerCallback& callback) {
  AssertOnOriginThread();
  DCHECK(!service_name.empty());
  DCHECK(!callback.is_null());

  GetDBusTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::ListenForServiceOwnerChangeInternal,
                 this, service_name, callback));
}

void Bus::ListenForServiceOwnerChangeInternal(
    const std::string& service_name,
    const GetServiceOwnerCallback& callback) {
  AssertOnDBusThread();
  DCHECK(!service_name.empty());
  DCHECK(!callback.is_null());

  if (!Connect() || !SetUpAsyncOperations())
    return;

  if (service_owner_changed_listener_map_.empty()) {
    bool filter_added =
        AddFilterFunction(Bus::OnServiceOwnerChangedFilter, this);
    DCHECK(filter_added);
  }

  ServiceOwnerChangedListenerMap::iterator it =
      service_owner_changed_listener_map_.find(service_name);
  if (it == service_owner_changed_listener_map_.end()) {
    // Add a match rule for the new service name.
    const std::string name_owner_changed_match_rule =
        base::StringPrintf(kServiceNameOwnerChangeMatchRule,
                           service_name.c_str());
    ScopedDBusError error;
    AddMatch(name_owner_changed_match_rule, error.get());
    if (error.is_set()) {
      LOG(ERROR) << "Failed to add match rule for " << service_name
                 << ". Got " << error.name() << ": " << error.message();
      return;
    }

    service_owner_changed_listener_map_[service_name].push_back(callback);
    return;
  }

  // Check if the callback has already been added.
  std::vector<GetServiceOwnerCallback>& callbacks = it->second;
  for (size_t i = 0; i < callbacks.size(); ++i) {
    if (callbacks[i].Equals(callback))
      return;
  }
  callbacks.push_back(callback);
}

void Bus::UnlistenForServiceOwnerChange(
    const std::string& service_name,
    const GetServiceOwnerCallback& callback) {
  AssertOnOriginThread();
  DCHECK(!service_name.empty());
  DCHECK(!callback.is_null());

  GetDBusTaskRunner()->PostTask(
      FROM_HERE,
      base::Bind(&Bus::UnlistenForServiceOwnerChangeInternal,
                 this, service_name, callback));
}

void Bus::UnlistenForServiceOwnerChangeInternal(
    const std::string& service_name,
    const GetServiceOwnerCallback& callback) {
  AssertOnDBusThread();
  DCHECK(!service_name.empty());
  DCHECK(!callback.is_null());

  ServiceOwnerChangedListenerMap::iterator it =
      service_owner_changed_listener_map_.find(service_name);
  if (it == service_owner_changed_listener_map_.end())
    return;

  std::vector<GetServiceOwnerCallback>& callbacks = it->second;
  for (size_t i = 0; i < callbacks.size(); ++i) {
    if (callbacks[i].Equals(callback)) {
      callbacks.erase(callbacks.begin() + i);
      break;  // There can be only one.
    }
  }
  if (!callbacks.empty())
    return;

  // Last callback for |service_name| has been removed, remove match rule.
  const std::string name_owner_changed_match_rule =
      base::StringPrintf(kServiceNameOwnerChangeMatchRule,
                         service_name.c_str());
  ScopedDBusError error;
  RemoveMatch(name_owner_changed_match_rule, error.get());
  // And remove |service_owner_changed_listener_map_| entry.
  service_owner_changed_listener_map_.erase(it);

  if (service_owner_changed_listener_map_.empty()) {
    bool filter_removed =
        RemoveFilterFunction(Bus::OnServiceOwnerChangedFilter, this);
    DCHECK(filter_removed);
  }
}

dbus_bool_t Bus::OnAddWatch(DBusWatch* raw_watch) {
  AssertOnDBusThread();

  // watch will be deleted when raw_watch is removed in OnRemoveWatch().
  Watch* watch = new Watch(raw_watch);
  if (watch->IsReadyToBeWatched()) {
    watch->StartWatching();
  }
  ++num_pending_watches_;
  return true;
}

void Bus::OnRemoveWatch(DBusWatch* raw_watch) {
  AssertOnDBusThread();

  Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
  delete watch;
  --num_pending_watches_;
}

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);
  if (timeout->IsReadyToBeMonitored()) {
    timeout->StartMonitoring(this);
  }
  ++num_pending_timeouts_;
  return true;
}

void Bus::OnRemoveTimeout(DBusTimeout* raw_timeout) {
  AssertOnDBusThread();

  Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
  timeout->Complete();
  --num_pending_timeouts_;
}

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();

  // 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().
  GetDBusTaskRunner()->PostTask(FROM_HERE,
                                base::Bind(&Bus::ProcessAllIncomingDataIfAny,
                                           this));
}

void Bus::OnConnectionDisconnected(DBusConnection* connection) {
  AssertOnDBusThread();

  if (!on_disconnected_closure_.is_null())
    GetOriginTaskRunner()->PostTask(FROM_HERE, on_disconnected_closure_);

  if (!connection)
    return;
  DCHECK(!dbus_connection_get_is_connected(connection));

  ShutdownAndBlock();
}

void Bus::OnServiceOwnerChanged(DBusMessage* message) {
  DCHECK(message);
  AssertOnDBusThread();

  // |message| will be unrefed on exit of the function. Increment the
  // reference so we can use it in Signal::FromRawMessage() below.
  dbus_message_ref(message);
  scoped_ptr<Signal> signal(Signal::FromRawMessage(message));

  // Confirm the validity of the NameOwnerChanged signal.
  if (signal->GetMember() != kNameOwnerChangedSignal ||
      signal->GetInterface() != DBUS_INTERFACE_DBUS ||
      signal->GetSender() != DBUS_SERVICE_DBUS) {
    return;
  }

  MessageReader reader(signal.get());
  std::string service_name;
  std::string old_owner;
  std::string new_owner;
  if (!reader.PopString(&service_name) ||
      !reader.PopString(&old_owner) ||
      !reader.PopString(&new_owner)) {
    return;
  }

  ServiceOwnerChangedListenerMap::const_iterator it =
      service_owner_changed_listener_map_.find(service_name);
  if (it == service_owner_changed_listener_map_.end())
    return;

  const std::vector<GetServiceOwnerCallback>& callbacks = it->second;
  for (size_t i = 0; i < callbacks.size(); ++i) {
    GetOriginTaskRunner()->PostTask(FROM_HERE,
                                    base::Bind(callbacks[i], new_owner));
  }
}

// static
dbus_bool_t Bus::OnAddWatchThunk(DBusWatch* raw_watch, void* data) {
  Bus* self = static_cast<Bus*>(data);
  return self->OnAddWatch(raw_watch);
}

// static
void Bus::OnRemoveWatchThunk(DBusWatch* raw_watch, void* data) {
  Bus* self = static_cast<Bus*>(data);
  self->OnRemoveWatch(raw_watch);
}

// static
void Bus::OnToggleWatchThunk(DBusWatch* raw_watch, void* data) {
  Bus* self = static_cast<Bus*>(data);
  self->OnToggleWatch(raw_watch);
}

// static
dbus_bool_t Bus::OnAddTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
  Bus* self = static_cast<Bus*>(data);
  return self->OnAddTimeout(raw_timeout);
}

// static
void Bus::OnRemoveTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
  Bus* self = static_cast<Bus*>(data);
  self->OnRemoveTimeout(raw_timeout);
}

// static
void Bus::OnToggleTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
  Bus* self = static_cast<Bus*>(data);
  self->OnToggleTimeout(raw_timeout);
}

// static
void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
                                       DBusDispatchStatus status,
                                       void* data) {
  Bus* self = static_cast<Bus*>(data);
  self->OnDispatchStatusChanged(connection, status);
}

// static
DBusHandlerResult Bus::OnConnectionDisconnectedFilter(
    DBusConnection* connection,
    DBusMessage* message,
    void* data) {
  if (dbus_message_is_signal(message,
                             DBUS_INTERFACE_LOCAL,
                             kDisconnectedSignal)) {
    Bus* self = static_cast<Bus*>(data);
    self->OnConnectionDisconnected(connection);
    return DBUS_HANDLER_RESULT_HANDLED;
  }
  return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}

// static
DBusHandlerResult Bus::OnServiceOwnerChangedFilter(
    DBusConnection* connection,
    DBusMessage* message,
    void* data) {
  if (dbus_message_is_signal(message,
                             DBUS_INTERFACE_DBUS,
                             kNameOwnerChangedSignal)) {
    Bus* self = static_cast<Bus*>(data);
    self->OnServiceOwnerChanged(message);
  }
  // Always return unhandled to let others, e.g. ObjectProxies, handle the same
  // signal.
  return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}

}  // namespace dbus