Home | History | Annotate | Download | only in dbus 
      1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
      2 // Use of this source code is governed by a BSD-style license that can be
      3 // found in the LICENSE file.
      4 
      5 #include "dbus/bus.h"
      6 
      7 #include "base/bind.h"
      8 #include "base/logging.h"
      9 #include "base/message_loop/message_loop.h"
     10 #include "base/message_loop/message_loop_proxy.h"
     11 #include "base/stl_util.h"
     12 #include "base/strings/stringprintf.h"
     13 #include "base/threading/thread.h"
     14 #include "base/threading/thread_restrictions.h"
     15 #include "base/time/time.h"
     16 #include "dbus/exported_object.h"
     17 #include "dbus/message.h"
     18 #include "dbus/object_manager.h"
     19 #include "dbus/object_path.h"
     20 #include "dbus/object_proxy.h"
     21 #include "dbus/scoped_dbus_error.h"
     22 
     23 namespace dbus {
     24 
     25 namespace {
     26 
     27 const char kDisconnectedSignal[] = "Disconnected";
     28 const char kDisconnectedMatchRule[] =
     29     "type='signal', path='/org/freedesktop/DBus/Local',"
     30     "interface='org.freedesktop.DBus.Local', member='Disconnected'";
     31 
     32 // The NameOwnerChanged member in org.freedesktop.DBus
     33 const char kNameOwnerChangedSignal[] = "NameOwnerChanged";
     34 
     35 // The match rule used to filter for changes to a given service name owner.
     36 const char kServiceNameOwnerChangeMatchRule[] =
     37     "type='signal',interface='org.freedesktop.DBus',"
     38     "member='NameOwnerChanged',path='/org/freedesktop/DBus',"
     39     "sender='org.freedesktop.DBus',arg0='%s'";
     40 
     41 // The class is used for watching the file descriptor used for D-Bus
     42 // communication.
     43 class Watch : public base::MessagePumpLibevent::Watcher {
     44  public:
     45   explicit Watch(DBusWatch* watch)
     46       : raw_watch_(watch) {
     47     dbus_watch_set_data(raw_watch_, this, NULL);
     48   }
     49 
     50   virtual ~Watch() {
     51     dbus_watch_set_data(raw_watch_, NULL, NULL);
     52   }
     53 
     54   // Returns true if the underlying file descriptor is ready to be watched.
     55   bool IsReadyToBeWatched() {
     56     return dbus_watch_get_enabled(raw_watch_);
     57   }
     58 
     59   // Starts watching the underlying file descriptor.
     60   void StartWatching() {
     61     const int file_descriptor = dbus_watch_get_unix_fd(raw_watch_);
     62     const int flags = dbus_watch_get_flags(raw_watch_);
     63 
     64     base::MessageLoopForIO::Mode mode = base::MessageLoopForIO::WATCH_READ;
     65     if ((flags & DBUS_WATCH_READABLE) && (flags & DBUS_WATCH_WRITABLE))
     66       mode = base::MessageLoopForIO::WATCH_READ_WRITE;
     67     else if (flags & DBUS_WATCH_READABLE)
     68       mode = base::MessageLoopForIO::WATCH_READ;
     69     else if (flags & DBUS_WATCH_WRITABLE)
     70       mode = base::MessageLoopForIO::WATCH_WRITE;
     71     else
     72       NOTREACHED();
     73 
     74     const bool persistent = true;  // Watch persistently.
     75     const bool success = base::MessageLoopForIO::current()->WatchFileDescriptor(
     76         file_descriptor, persistent, mode, &file_descriptor_watcher_, this);
     77     CHECK(success) << "Unable to allocate memory";
     78   }
     79 
     80   // Stops watching the underlying file descriptor.
     81   void StopWatching() {
     82     file_descriptor_watcher_.StopWatchingFileDescriptor();
     83   }
     84 
     85  private:
     86   // Implement MessagePumpLibevent::Watcher.
     87   virtual void OnFileCanReadWithoutBlocking(int file_descriptor) OVERRIDE {
     88     const bool success = dbus_watch_handle(raw_watch_, DBUS_WATCH_READABLE);
     89     CHECK(success) << "Unable to allocate memory";
     90   }
     91 
     92   // Implement MessagePumpLibevent::Watcher.
     93   virtual void OnFileCanWriteWithoutBlocking(int file_descriptor) OVERRIDE {
     94     const bool success = dbus_watch_handle(raw_watch_, DBUS_WATCH_WRITABLE);
     95     CHECK(success) << "Unable to allocate memory";
     96   }
     97 
     98   DBusWatch* raw_watch_;
     99   base::MessagePumpLibevent::FileDescriptorWatcher file_descriptor_watcher_;
    100 };
    101 
    102 // The class is used for monitoring the timeout used for D-Bus method
    103 // calls.
    104 //
    105 // Unlike Watch, Timeout is a ref counted object, to ensure that |this| of
    106 // the object is is alive when HandleTimeout() is called. It's unlikely
    107 // but it may be possible that HandleTimeout() is called after
    108 // Bus::OnRemoveTimeout(). That's why we don't simply delete the object in
    109 // Bus::OnRemoveTimeout().
    110 class Timeout : public base::RefCountedThreadSafe<Timeout> {
    111  public:
    112   explicit Timeout(DBusTimeout* timeout)
    113       : raw_timeout_(timeout),
    114         monitoring_is_active_(false),
    115         is_completed(false) {
    116     dbus_timeout_set_data(raw_timeout_, this, NULL);
    117     AddRef();  // Balanced on Complete().
    118   }
    119 
    120   // Returns true if the timeout is ready to be monitored.
    121   bool IsReadyToBeMonitored() {
    122     return dbus_timeout_get_enabled(raw_timeout_);
    123   }
    124 
    125   // Starts monitoring the timeout.
    126   void StartMonitoring(Bus* bus) {
    127     bus->GetDBusTaskRunner()->PostDelayedTask(
    128         FROM_HERE,
    129         base::Bind(&Timeout::HandleTimeout, this),
    130         GetInterval());
    131     monitoring_is_active_ = true;
    132   }
    133 
    134   // Stops monitoring the timeout.
    135   void StopMonitoring() {
    136     // We cannot take back the delayed task we posted in
    137     // StartMonitoring(), so we just mark the monitoring is inactive now.
    138     monitoring_is_active_ = false;
    139   }
    140 
    141   // Returns the interval.
    142   base::TimeDelta GetInterval() {
    143     return base::TimeDelta::FromMilliseconds(
    144         dbus_timeout_get_interval(raw_timeout_));
    145   }
    146 
    147   // Cleans up the raw_timeout and marks that timeout is completed.
    148   // See the class comment above for why we are doing this.
    149   void Complete() {
    150     dbus_timeout_set_data(raw_timeout_, NULL, NULL);
    151     is_completed = true;
    152     Release();
    153   }
    154 
    155  private:
    156   friend class base::RefCountedThreadSafe<Timeout>;
    157   ~Timeout() {
    158   }
    159 
    160   // Handles the timeout.
    161   void HandleTimeout() {
    162     // If the timeout is marked completed, we should do nothing. This can
    163     // occur if this function is called after Bus::OnRemoveTimeout().
    164     if (is_completed)
    165       return;
    166     // Skip if monitoring is canceled.
    167     if (!monitoring_is_active_)
    168       return;
    169 
    170     const bool success = dbus_timeout_handle(raw_timeout_);
    171     CHECK(success) << "Unable to allocate memory";
    172   }
    173 
    174   DBusTimeout* raw_timeout_;
    175   bool monitoring_is_active_;
    176   bool is_completed;
    177 };
    178 
    179 }  // namespace
    180 
    181 Bus::Options::Options()
    182   : bus_type(SESSION),
    183     connection_type(PRIVATE) {
    184 }
    185 
    186 Bus::Options::~Options() {
    187 }
    188 
    189 Bus::Bus(const Options& options)
    190     : bus_type_(options.bus_type),
    191       connection_type_(options.connection_type),
    192       dbus_task_runner_(options.dbus_task_runner),
    193       on_shutdown_(false /* manual_reset */, false /* initially_signaled */),
    194       connection_(NULL),
    195       origin_thread_id_(base::PlatformThread::CurrentId()),
    196       async_operations_set_up_(false),
    197       shutdown_completed_(false),
    198       num_pending_watches_(0),
    199       num_pending_timeouts_(0),
    200       address_(options.address),
    201       on_disconnected_closure_(options.disconnected_callback) {
    202   // This is safe to call multiple times.
    203   dbus_threads_init_default();
    204   // The origin message loop is unnecessary if the client uses synchronous
    205   // functions only.
    206   if (base::MessageLoop::current())
    207     origin_task_runner_ = base::MessageLoop::current()->message_loop_proxy();
    208 }
    209 
    210 Bus::~Bus() {
    211   DCHECK(!connection_);
    212   DCHECK(owned_service_names_.empty());
    213   DCHECK(match_rules_added_.empty());
    214   DCHECK(filter_functions_added_.empty());
    215   DCHECK(registered_object_paths_.empty());
    216   DCHECK_EQ(0, num_pending_watches_);
    217   // TODO(satorux): This check fails occasionally in browser_tests for tests
    218   // that run very quickly. Perhaps something does not have time to clean up.
    219   // Despite the check failing, the tests seem to run fine. crosbug.com/23416
    220   // DCHECK_EQ(0, num_pending_timeouts_);
    221 }
    222 
    223 ObjectProxy* Bus::GetObjectProxy(const std::string& service_name,
    224                                  const ObjectPath& object_path) {
    225   return GetObjectProxyWithOptions(service_name, object_path,
    226                                    ObjectProxy::DEFAULT_OPTIONS);
    227 }
    228 
    229 ObjectProxy* Bus::GetObjectProxyWithOptions(const std::string& service_name,
    230                                             const ObjectPath& object_path,
    231                                             int options) {
    232   AssertOnOriginThread();
    233 
    234   // Check if we already have the requested object proxy.
    235   const ObjectProxyTable::key_type key(service_name + object_path.value(),
    236                                        options);
    237   ObjectProxyTable::iterator iter = object_proxy_table_.find(key);
    238   if (iter != object_proxy_table_.end()) {
    239     return iter->second.get();
    240   }
    241 
    242   scoped_refptr<ObjectProxy> object_proxy =
    243       new ObjectProxy(this, service_name, object_path, options);
    244   object_proxy_table_[key] = object_proxy;
    245 
    246   return object_proxy.get();
    247 }
    248 
    249 bool Bus::RemoveObjectProxy(const std::string& service_name,
    250                             const ObjectPath& object_path,
    251                             const base::Closure& callback) {
    252   return RemoveObjectProxyWithOptions(service_name, object_path,
    253                                       ObjectProxy::DEFAULT_OPTIONS,
    254                                       callback);
    255 }
    256 
    257 bool Bus::RemoveObjectProxyWithOptions(const std::string& service_name,
    258                                        const ObjectPath& object_path,
    259                                        int options,
    260                                        const base::Closure& callback) {
    261   AssertOnOriginThread();
    262 
    263   // Check if we have the requested object proxy.
    264   const ObjectProxyTable::key_type key(service_name + object_path.value(),
    265                                        options);
    266   ObjectProxyTable::iterator iter = object_proxy_table_.find(key);
    267   if (iter != object_proxy_table_.end()) {
    268     scoped_refptr<ObjectProxy> object_proxy = iter->second;
    269     object_proxy_table_.erase(iter);
    270     // Object is present. Remove it now and Detach on the DBus thread.
    271     GetDBusTaskRunner()->PostTask(
    272         FROM_HERE,
    273         base::Bind(&Bus::RemoveObjectProxyInternal,
    274                    this, object_proxy, callback));
    275     return true;
    276   }
    277   return false;
    278 }
    279 
    280 void Bus::RemoveObjectProxyInternal(scoped_refptr<ObjectProxy> object_proxy,
    281                                     const base::Closure& callback) {
    282   AssertOnDBusThread();
    283 
    284   object_proxy.get()->Detach();
    285 
    286   GetOriginTaskRunner()->PostTask(FROM_HERE, callback);
    287 }
    288 
    289 ExportedObject* Bus::GetExportedObject(const ObjectPath& object_path) {
    290   AssertOnOriginThread();
    291 
    292   // Check if we already have the requested exported object.
    293   ExportedObjectTable::iterator iter = exported_object_table_.find(object_path);
    294   if (iter != exported_object_table_.end()) {
    295     return iter->second.get();
    296   }
    297 
    298   scoped_refptr<ExportedObject> exported_object =
    299       new ExportedObject(this, object_path);
    300   exported_object_table_[object_path] = exported_object;
    301 
    302   return exported_object.get();
    303 }
    304 
    305 void Bus::UnregisterExportedObject(const ObjectPath& object_path) {
    306   AssertOnOriginThread();
    307 
    308   // Remove the registered object from the table first, to allow a new
    309   // GetExportedObject() call to return a new object, rather than this one.
    310   ExportedObjectTable::iterator iter = exported_object_table_.find(object_path);
    311   if (iter == exported_object_table_.end())
    312     return;
    313 
    314   scoped_refptr<ExportedObject> exported_object = iter->second;
    315   exported_object_table_.erase(iter);
    316 
    317   // Post the task to perform the final unregistration to the D-Bus thread.
    318   // Since the registration also happens on the D-Bus thread in
    319   // TryRegisterObjectPath(), and the task runner we post to is a
    320   // SequencedTaskRunner, there is a guarantee that this will happen before any
    321   // future registration call.
    322   GetDBusTaskRunner()->PostTask(
    323       FROM_HERE,
    324       base::Bind(&Bus::UnregisterExportedObjectInternal,
    325                  this, exported_object));
    326 }
    327 
    328 void Bus::UnregisterExportedObjectInternal(
    329     scoped_refptr<ExportedObject> exported_object) {
    330   AssertOnDBusThread();
    331 
    332   exported_object->Unregister();
    333 }
    334 
    335 ObjectManager* Bus::GetObjectManager(const std::string& service_name,
    336                                      const ObjectPath& object_path) {
    337   AssertOnOriginThread();
    338 
    339   // Check if we already have the requested object manager.
    340   const ObjectManagerTable::key_type key(service_name + object_path.value());
    341   ObjectManagerTable::iterator iter = object_manager_table_.find(key);
    342   if (iter != object_manager_table_.end()) {
    343     return iter->second.get();
    344   }
    345 
    346   scoped_refptr<ObjectManager> object_manager =
    347       new ObjectManager(this, service_name, object_path);
    348   object_manager_table_[key] = object_manager;
    349 
    350   return object_manager.get();
    351 }
    352 
    353 bool Bus::RemoveObjectManager(const std::string& service_name,
    354                               const ObjectPath& object_path,
    355                               const base::Closure& callback) {
    356   AssertOnOriginThread();
    357   DCHECK(!callback.is_null());
    358 
    359   const ObjectManagerTable::key_type key(service_name + object_path.value());
    360   ObjectManagerTable::iterator iter = object_manager_table_.find(key);
    361   if (iter == object_manager_table_.end())
    362     return false;
    363 
    364   // ObjectManager is present. Remove it now and CleanUp on the DBus thread.
    365   scoped_refptr<ObjectManager> object_manager = iter->second;
    366   object_manager_table_.erase(iter);
    367 
    368   GetDBusTaskRunner()->PostTask(
    369       FROM_HERE,
    370       base::Bind(&Bus::RemoveObjectManagerInternal,
    371                  this, object_manager, callback));
    372 
    373   return true;
    374 }
    375 
    376 void Bus::RemoveObjectManagerInternal(
    377       scoped_refptr<dbus::ObjectManager> object_manager,
    378       const base::Closure& callback) {
    379   AssertOnDBusThread();
    380   DCHECK(object_manager.get());
    381 
    382   object_manager->CleanUp();
    383 
    384   // The ObjectManager has to be deleted on the origin thread since it was
    385   // created there.
    386   GetOriginTaskRunner()->PostTask(
    387       FROM_HERE,
    388       base::Bind(&Bus::RemoveObjectManagerInternalHelper,
    389                  this, object_manager, callback));
    390 }
    391 
    392 void Bus::RemoveObjectManagerInternalHelper(
    393       scoped_refptr<dbus::ObjectManager> object_manager,
    394       const base::Closure& callback) {
    395   AssertOnOriginThread();
    396   DCHECK(object_manager.get());
    397 
    398   // Release the object manager and run the callback.
    399   object_manager = NULL;
    400   callback.Run();
    401 }
    402 
    403 void Bus::GetManagedObjects() {
    404   for (ObjectManagerTable::iterator iter = object_manager_table_.begin();
    405        iter != object_manager_table_.end(); ++iter) {
    406     iter->second->GetManagedObjects();
    407   }
    408 }
    409 
    410 bool Bus::Connect() {
    411   // dbus_bus_get_private() and dbus_bus_get() are blocking calls.
    412   AssertOnDBusThread();
    413 
    414   // Check if it's already initialized.
    415   if (connection_)
    416     return true;
    417 
    418   ScopedDBusError error;
    419   if (bus_type_ == CUSTOM_ADDRESS) {
    420     if (connection_type_ == PRIVATE) {
    421       connection_ = dbus_connection_open_private(address_.c_str(), error.get());
    422     } else {
    423       connection_ = dbus_connection_open(address_.c_str(), error.get());
    424     }
    425   } else {
    426     const DBusBusType dbus_bus_type = static_cast<DBusBusType>(bus_type_);
    427     if (connection_type_ == PRIVATE) {
    428       connection_ = dbus_bus_get_private(dbus_bus_type, error.get());
    429     } else {
    430       connection_ = dbus_bus_get(dbus_bus_type, error.get());
    431     }
    432   }
    433   if (!connection_) {
    434     LOG(ERROR) << "Failed to connect to the bus: "
    435                << (error.is_set() ? error.message() : "");
    436     return false;
    437   }
    438 
    439   if (bus_type_ == CUSTOM_ADDRESS) {
    440     // We should call dbus_bus_register here, otherwise unique name can not be
    441     // acquired. According to dbus specification, it is responsible to call
    442     // org.freedesktop.DBus.Hello method at the beging of bus connection to
    443     // acquire unique name. In the case of dbus_bus_get, dbus_bus_register is
    444     // called internally.
    445     if (!dbus_bus_register(connection_, error.get())) {
    446       LOG(ERROR) << "Failed to register the bus component: "
    447                  << (error.is_set() ? error.message() : "");
    448       return false;
    449     }
    450   }
    451   // We shouldn't exit on the disconnected signal.
    452   dbus_connection_set_exit_on_disconnect(connection_, false);
    453 
    454   // Watch Disconnected signal.
    455   AddFilterFunction(Bus::OnConnectionDisconnectedFilter, this);
    456   AddMatch(kDisconnectedMatchRule, error.get());
    457 
    458   return true;
    459 }
    460 
    461 void Bus::ClosePrivateConnection() {
    462   // dbus_connection_close is blocking call.
    463   AssertOnDBusThread();
    464   DCHECK_EQ(PRIVATE, connection_type_)
    465       << "non-private connection should not be closed";
    466   dbus_connection_close(connection_);
    467 }
    468 
    469 void Bus::ShutdownAndBlock() {
    470   AssertOnDBusThread();
    471 
    472   if (shutdown_completed_)
    473     return;  // Already shutdowned, just return.
    474 
    475   // Unregister the exported objects.
    476   for (ExportedObjectTable::iterator iter = exported_object_table_.begin();
    477        iter != exported_object_table_.end(); ++iter) {
    478     iter->second->Unregister();
    479   }
    480 
    481   // Release all service names.
    482   for (std::set<std::string>::iterator iter = owned_service_names_.begin();
    483        iter != owned_service_names_.end();) {
    484     // This is a bit tricky but we should increment the iter here as
    485     // ReleaseOwnership() may remove |service_name| from the set.
    486     const std::string& service_name = *iter++;
    487     ReleaseOwnership(service_name);
    488   }
    489   if (!owned_service_names_.empty()) {
    490     LOG(ERROR) << "Failed to release all service names. # of services left: "
    491                << owned_service_names_.size();
    492   }
    493 
    494   // Detach from the remote objects.
    495   for (ObjectProxyTable::iterator iter = object_proxy_table_.begin();
    496        iter != object_proxy_table_.end(); ++iter) {
    497     iter->second->Detach();
    498   }
    499 
    500   // Clean up the object managers.
    501   for (ObjectManagerTable::iterator iter = object_manager_table_.begin();
    502        iter != object_manager_table_.end(); ++iter) {
    503     iter->second->CleanUp();
    504   }
    505 
    506   // Release object proxies and exported objects here. We should do this
    507   // here rather than in the destructor to avoid memory leaks due to
    508   // cyclic references.
    509   object_proxy_table_.clear();
    510   exported_object_table_.clear();
    511 
    512   // Private connection should be closed.
    513   if (connection_) {
    514     // Remove Disconnected watcher.
    515     ScopedDBusError error;
    516     RemoveFilterFunction(Bus::OnConnectionDisconnectedFilter, this);
    517     RemoveMatch(kDisconnectedMatchRule, error.get());
    518 
    519     if (connection_type_ == PRIVATE)
    520       ClosePrivateConnection();
    521     // dbus_connection_close() won't unref.
    522     dbus_connection_unref(connection_);
    523   }
    524 
    525   connection_ = NULL;
    526   shutdown_completed_ = true;
    527 }
    528 
    529 void Bus::ShutdownOnDBusThreadAndBlock() {
    530   AssertOnOriginThread();
    531   DCHECK(dbus_task_runner_.get());
    532 
    533   GetDBusTaskRunner()->PostTask(
    534       FROM_HERE,
    535       base::Bind(&Bus::ShutdownOnDBusThreadAndBlockInternal, this));
    536 
    537   // http://crbug.com/125222
    538   base::ThreadRestrictions::ScopedAllowWait allow_wait;
    539 
    540   // Wait until the shutdown is complete on the D-Bus thread.
    541   // The shutdown should not hang, but set timeout just in case.
    542   const int kTimeoutSecs = 3;
    543   const base::TimeDelta timeout(base::TimeDelta::FromSeconds(kTimeoutSecs));
    544   const bool signaled = on_shutdown_.TimedWait(timeout);
    545   LOG_IF(ERROR, !signaled) << "Failed to shutdown the bus";
    546 }
    547 
    548 void Bus::RequestOwnership(const std::string& service_name,
    549                            ServiceOwnershipOptions options,
    550                            OnOwnershipCallback on_ownership_callback) {
    551   AssertOnOriginThread();
    552 
    553   GetDBusTaskRunner()->PostTask(
    554       FROM_HERE,
    555       base::Bind(&Bus::RequestOwnershipInternal,
    556                  this, service_name, options, on_ownership_callback));
    557 }
    558 
    559 void Bus::RequestOwnershipInternal(const std::string& service_name,
    560                                    ServiceOwnershipOptions options,
    561                                    OnOwnershipCallback on_ownership_callback) {
    562   AssertOnDBusThread();
    563 
    564   bool success = Connect();
    565   if (success)
    566     success = RequestOwnershipAndBlock(service_name, options);
    567 
    568   GetOriginTaskRunner()->PostTask(FROM_HERE,
    569                                   base::Bind(on_ownership_callback,
    570                                              service_name,
    571                                              success));
    572 }
    573 
    574 bool Bus::RequestOwnershipAndBlock(const std::string& service_name,
    575                                    ServiceOwnershipOptions options) {
    576   DCHECK(connection_);
    577   // dbus_bus_request_name() is a blocking call.
    578   AssertOnDBusThread();
    579 
    580   // Check if we already own the service name.
    581   if (owned_service_names_.find(service_name) != owned_service_names_.end()) {
    582     return true;
    583   }
    584 
    585   ScopedDBusError error;
    586   const int result = dbus_bus_request_name(connection_,
    587                                            service_name.c_str(),
    588                                            options,
    589                                            error.get());
    590   if (result != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER) {
    591     LOG(ERROR) << "Failed to get the ownership of " << service_name << ": "
    592                << (error.is_set() ? error.message() : "");
    593     return false;
    594   }
    595   owned_service_names_.insert(service_name);
    596   return true;
    597 }
    598 
    599 bool Bus::ReleaseOwnership(const std::string& service_name) {
    600   DCHECK(connection_);
    601   // dbus_bus_request_name() is a blocking call.
    602   AssertOnDBusThread();
    603 
    604   // Check if we already own the service name.
    605   std::set<std::string>::iterator found =
    606       owned_service_names_.find(service_name);
    607   if (found == owned_service_names_.end()) {
    608     LOG(ERROR) << service_name << " is not owned by the bus";
    609     return false;
    610   }
    611 
    612   ScopedDBusError error;
    613   const int result = dbus_bus_release_name(connection_, service_name.c_str(),
    614                                            error.get());
    615   if (result == DBUS_RELEASE_NAME_REPLY_RELEASED) {
    616     owned_service_names_.erase(found);
    617     return true;
    618   } else {
    619     LOG(ERROR) << "Failed to release the ownership of " << service_name << ": "
    620                << (error.is_set() ? error.message() : "")
    621                << ", result code: " << result;
    622     return false;
    623   }
    624 }
    625 
    626 bool Bus::SetUpAsyncOperations() {
    627   DCHECK(connection_);
    628   AssertOnDBusThread();
    629 
    630   if (async_operations_set_up_)
    631     return true;
    632 
    633   // Process all the incoming data if any, so that OnDispatchStatus() will
    634   // be called when the incoming data is ready.
    635   ProcessAllIncomingDataIfAny();
    636 
    637   bool success = dbus_connection_set_watch_functions(connection_,
    638                                                      &Bus::OnAddWatchThunk,
    639                                                      &Bus::OnRemoveWatchThunk,
    640                                                      &Bus::OnToggleWatchThunk,
    641                                                      this,
    642                                                      NULL);
    643   CHECK(success) << "Unable to allocate memory";
    644 
    645   success = dbus_connection_set_timeout_functions(connection_,
    646                                                   &Bus::OnAddTimeoutThunk,
    647                                                   &Bus::OnRemoveTimeoutThunk,
    648                                                   &Bus::OnToggleTimeoutThunk,
    649                                                   this,
    650                                                   NULL);
    651   CHECK(success) << "Unable to allocate memory";
    652 
    653   dbus_connection_set_dispatch_status_function(
    654       connection_,
    655       &Bus::OnDispatchStatusChangedThunk,
    656       this,
    657       NULL);
    658 
    659   async_operations_set_up_ = true;
    660 
    661   return true;
    662 }
    663 
    664 DBusMessage* Bus::SendWithReplyAndBlock(DBusMessage* request,
    665                                         int timeout_ms,
    666                                         DBusError* error) {
    667   DCHECK(connection_);
    668   AssertOnDBusThread();
    669 
    670   return dbus_connection_send_with_reply_and_block(
    671       connection_, request, timeout_ms, error);
    672 }
    673 
    674 void Bus::SendWithReply(DBusMessage* request,
    675                         DBusPendingCall** pending_call,
    676                         int timeout_ms) {
    677   DCHECK(connection_);
    678   AssertOnDBusThread();
    679 
    680   const bool success = dbus_connection_send_with_reply(
    681       connection_, request, pending_call, timeout_ms);
    682   CHECK(success) << "Unable to allocate memory";
    683 }
    684 
    685 void Bus::Send(DBusMessage* request, uint32* serial) {
    686   DCHECK(connection_);
    687   AssertOnDBusThread();
    688 
    689   const bool success = dbus_connection_send(connection_, request, serial);
    690   CHECK(success) << "Unable to allocate memory";
    691 }
    692 
    693 bool Bus::AddFilterFunction(DBusHandleMessageFunction filter_function,
    694                             void* user_data) {
    695   DCHECK(connection_);
    696   AssertOnDBusThread();
    697 
    698   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
    699       std::make_pair(filter_function, user_data);
    700   if (filter_functions_added_.find(filter_data_pair) !=
    701       filter_functions_added_.end()) {
    702     VLOG(1) << "Filter function already exists: " << filter_function
    703             << " with associated data: " << user_data;
    704     return false;
    705   }
    706 
    707   const bool success = dbus_connection_add_filter(
    708       connection_, filter_function, user_data, NULL);
    709   CHECK(success) << "Unable to allocate memory";
    710   filter_functions_added_.insert(filter_data_pair);
    711   return true;
    712 }
    713 
    714 bool Bus::RemoveFilterFunction(DBusHandleMessageFunction filter_function,
    715                                void* user_data) {
    716   DCHECK(connection_);
    717   AssertOnDBusThread();
    718 
    719   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
    720       std::make_pair(filter_function, user_data);
    721   if (filter_functions_added_.find(filter_data_pair) ==
    722       filter_functions_added_.end()) {
    723     VLOG(1) << "Requested to remove an unknown filter function: "
    724             << filter_function
    725             << " with associated data: " << user_data;
    726     return false;
    727   }
    728 
    729   dbus_connection_remove_filter(connection_, filter_function, user_data);
    730   filter_functions_added_.erase(filter_data_pair);
    731   return true;
    732 }
    733 
    734 void Bus::AddMatch(const std::string& match_rule, DBusError* error) {
    735   DCHECK(connection_);
    736   AssertOnDBusThread();
    737 
    738   std::map<std::string, int>::iterator iter =
    739       match_rules_added_.find(match_rule);
    740   if (iter != match_rules_added_.end()) {
    741     // The already existing rule's counter is incremented.
    742     iter->second++;
    743 
    744     VLOG(1) << "Match rule already exists: " << match_rule;
    745     return;
    746   }
    747 
    748   dbus_bus_add_match(connection_, match_rule.c_str(), error);
    749   match_rules_added_[match_rule] = 1;
    750 }
    751 
    752 bool Bus::RemoveMatch(const std::string& match_rule, DBusError* error) {
    753   DCHECK(connection_);
    754   AssertOnDBusThread();
    755 
    756   std::map<std::string, int>::iterator iter =
    757       match_rules_added_.find(match_rule);
    758   if (iter == match_rules_added_.end()) {
    759     LOG(ERROR) << "Requested to remove an unknown match rule: " << match_rule;
    760     return false;
    761   }
    762 
    763   // The rule's counter is decremented and the rule is deleted when reachs 0.
    764   iter->second--;
    765   if (iter->second == 0) {
    766     dbus_bus_remove_match(connection_, match_rule.c_str(), error);
    767     match_rules_added_.erase(match_rule);
    768   }
    769   return true;
    770 }
    771 
    772 bool Bus::TryRegisterObjectPath(const ObjectPath& object_path,
    773                                 const DBusObjectPathVTable* vtable,
    774                                 void* user_data,
    775                                 DBusError* error) {
    776   DCHECK(connection_);
    777   AssertOnDBusThread();
    778 
    779   if (registered_object_paths_.find(object_path) !=
    780       registered_object_paths_.end()) {
    781     LOG(ERROR) << "Object path already registered: " << object_path.value();
    782     return false;
    783   }
    784 
    785   const bool success = dbus_connection_try_register_object_path(
    786       connection_,
    787       object_path.value().c_str(),
    788       vtable,
    789       user_data,
    790       error);
    791   if (success)
    792     registered_object_paths_.insert(object_path);
    793   return success;
    794 }
    795 
    796 void Bus::UnregisterObjectPath(const ObjectPath& object_path) {
    797   DCHECK(connection_);
    798   AssertOnDBusThread();
    799 
    800   if (registered_object_paths_.find(object_path) ==
    801       registered_object_paths_.end()) {
    802     LOG(ERROR) << "Requested to unregister an unknown object path: "
    803                << object_path.value();
    804     return;
    805   }
    806 
    807   const bool success = dbus_connection_unregister_object_path(
    808       connection_,
    809       object_path.value().c_str());
    810   CHECK(success) << "Unable to allocate memory";
    811   registered_object_paths_.erase(object_path);
    812 }
    813 
    814 void Bus::ShutdownOnDBusThreadAndBlockInternal() {
    815   AssertOnDBusThread();
    816 
    817   ShutdownAndBlock();
    818   on_shutdown_.Signal();
    819 }
    820 
    821 void Bus::ProcessAllIncomingDataIfAny() {
    822   AssertOnDBusThread();
    823 
    824   // As mentioned at the class comment in .h file, connection_ can be NULL.
    825   if (!connection_)
    826     return;
    827 
    828   // It is safe and necessary to call dbus_connection_get_dispatch_status even
    829   // if the connection is lost. Otherwise we will miss "Disconnected" signal.
    830   // (crbug.com/174431)
    831   if (dbus_connection_get_dispatch_status(connection_) ==
    832       DBUS_DISPATCH_DATA_REMAINS) {
    833     while (dbus_connection_dispatch(connection_) ==
    834            DBUS_DISPATCH_DATA_REMAINS) {
    835     }
    836   }
    837 }
    838 
    839 base::TaskRunner* Bus::GetDBusTaskRunner() {
    840   if (dbus_task_runner_.get())
    841     return dbus_task_runner_.get();
    842   else
    843     return GetOriginTaskRunner();
    844 }
    845 
    846 base::TaskRunner* Bus::GetOriginTaskRunner() {
    847   DCHECK(origin_task_runner_.get());
    848   return origin_task_runner_.get();
    849 }
    850 
    851 bool Bus::HasDBusThread() {
    852   return dbus_task_runner_.get() != NULL;
    853 }
    854 
    855 void Bus::AssertOnOriginThread() {
    856   DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
    857 }
    858 
    859 void Bus::AssertOnDBusThread() {
    860   base::ThreadRestrictions::AssertIOAllowed();
    861 
    862   if (dbus_task_runner_.get()) {
    863     DCHECK(dbus_task_runner_->RunsTasksOnCurrentThread());
    864   } else {
    865     AssertOnOriginThread();
    866   }
    867 }
    868 
    869 std::string Bus::GetServiceOwnerAndBlock(const std::string& service_name,
    870                                          GetServiceOwnerOption options) {
    871   AssertOnDBusThread();
    872 
    873   MethodCall get_name_owner_call("org.freedesktop.DBus", "GetNameOwner");
    874   MessageWriter writer(&get_name_owner_call);
    875   writer.AppendString(service_name);
    876   VLOG(1) << "Method call: " << get_name_owner_call.ToString();
    877 
    878   const ObjectPath obj_path("/org/freedesktop/DBus");
    879   if (!get_name_owner_call.SetDestination("org.freedesktop.DBus") ||
    880       !get_name_owner_call.SetPath(obj_path)) {
    881     if (options == REPORT_ERRORS)
    882       LOG(ERROR) << "Failed to get name owner.";
    883     return "";
    884   }
    885 
    886   ScopedDBusError error;
    887   DBusMessage* response_message =
    888       SendWithReplyAndBlock(get_name_owner_call.raw_message(),
    889                             ObjectProxy::TIMEOUT_USE_DEFAULT,
    890                             error.get());
    891   if (!response_message) {
    892     if (options == REPORT_ERRORS) {
    893       LOG(ERROR) << "Failed to get name owner. Got " << error.name() << ": "
    894                  << error.message();
    895     }
    896     return "";
    897   }
    898 
    899   scoped_ptr<Response> response(Response::FromRawMessage(response_message));
    900   MessageReader reader(response.get());
    901 
    902   std::string service_owner;
    903   if (!reader.PopString(&service_owner))
    904     service_owner.clear();
    905   return service_owner;
    906 }
    907 
    908 void Bus::GetServiceOwner(const std::string& service_name,
    909                           const GetServiceOwnerCallback& callback) {
    910   AssertOnOriginThread();
    911 
    912   GetDBusTaskRunner()->PostTask(
    913       FROM_HERE,
    914       base::Bind(&Bus::GetServiceOwnerInternal, this, service_name, callback));
    915 }
    916 
    917 void Bus::GetServiceOwnerInternal(const std::string& service_name,
    918                                   const GetServiceOwnerCallback& callback) {
    919   AssertOnDBusThread();
    920 
    921   std::string service_owner;
    922   if (Connect())
    923     service_owner = GetServiceOwnerAndBlock(service_name, SUPPRESS_ERRORS);
    924   GetOriginTaskRunner()->PostTask(FROM_HERE,
    925                                   base::Bind(callback, service_owner));
    926 }
    927 
    928 void Bus::ListenForServiceOwnerChange(
    929     const std::string& service_name,
    930     const GetServiceOwnerCallback& callback) {
    931   AssertOnOriginThread();
    932   DCHECK(!service_name.empty());
    933   DCHECK(!callback.is_null());
    934 
    935   GetDBusTaskRunner()->PostTask(
    936       FROM_HERE,
    937       base::Bind(&Bus::ListenForServiceOwnerChangeInternal,
    938                  this, service_name, callback));
    939 }
    940 
    941 void Bus::ListenForServiceOwnerChangeInternal(
    942     const std::string& service_name,
    943     const GetServiceOwnerCallback& callback) {
    944   AssertOnDBusThread();
    945   DCHECK(!service_name.empty());
    946   DCHECK(!callback.is_null());
    947 
    948   if (!Connect() || !SetUpAsyncOperations())
    949     return;
    950 
    951   if (service_owner_changed_listener_map_.empty()) {
    952     bool filter_added =
    953         AddFilterFunction(Bus::OnServiceOwnerChangedFilter, this);
    954     DCHECK(filter_added);
    955   }
    956 
    957   ServiceOwnerChangedListenerMap::iterator it =
    958       service_owner_changed_listener_map_.find(service_name);
    959   if (it == service_owner_changed_listener_map_.end()) {
    960     // Add a match rule for the new service name.
    961     const std::string name_owner_changed_match_rule =
    962         base::StringPrintf(kServiceNameOwnerChangeMatchRule,
    963                            service_name.c_str());
    964     ScopedDBusError error;
    965     AddMatch(name_owner_changed_match_rule, error.get());
    966     if (error.is_set()) {
    967       LOG(ERROR) << "Failed to add match rule for " << service_name
    968                  << ". Got " << error.name() << ": " << error.message();
    969       return;
    970     }
    971 
    972     service_owner_changed_listener_map_[service_name].push_back(callback);
    973     return;
    974   }
    975 
    976   // Check if the callback has already been added.
    977   std::vector<GetServiceOwnerCallback>& callbacks = it->second;
    978   for (size_t i = 0; i < callbacks.size(); ++i) {
    979     if (callbacks[i].Equals(callback))
    980       return;
    981   }
    982   callbacks.push_back(callback);
    983 }
    984 
    985 void Bus::UnlistenForServiceOwnerChange(
    986     const std::string& service_name,
    987     const GetServiceOwnerCallback& callback) {
    988   AssertOnOriginThread();
    989   DCHECK(!service_name.empty());
    990   DCHECK(!callback.is_null());
    991 
    992   GetDBusTaskRunner()->PostTask(
    993       FROM_HERE,
    994       base::Bind(&Bus::UnlistenForServiceOwnerChangeInternal,
    995                  this, service_name, callback));
    996 }
    997 
    998 void Bus::UnlistenForServiceOwnerChangeInternal(
    999     const std::string& service_name,
   1000     const GetServiceOwnerCallback& callback) {
   1001   AssertOnDBusThread();
   1002   DCHECK(!service_name.empty());
   1003   DCHECK(!callback.is_null());
   1004 
   1005   ServiceOwnerChangedListenerMap::iterator it =
   1006       service_owner_changed_listener_map_.find(service_name);
   1007   if (it == service_owner_changed_listener_map_.end())
   1008     return;
   1009 
   1010   std::vector<GetServiceOwnerCallback>& callbacks = it->second;
   1011   for (size_t i = 0; i < callbacks.size(); ++i) {
   1012     if (callbacks[i].Equals(callback)) {
   1013       callbacks.erase(callbacks.begin() + i);
   1014       break;  // There can be only one.
   1015     }
   1016   }
   1017   if (!callbacks.empty())
   1018     return;
   1019 
   1020   // Last callback for |service_name| has been removed, remove match rule.
   1021   const std::string name_owner_changed_match_rule =
   1022       base::StringPrintf(kServiceNameOwnerChangeMatchRule,
   1023                          service_name.c_str());
   1024   ScopedDBusError error;
   1025   RemoveMatch(name_owner_changed_match_rule, error.get());
   1026   // And remove |service_owner_changed_listener_map_| entry.
   1027   service_owner_changed_listener_map_.erase(it);
   1028 
   1029   if (service_owner_changed_listener_map_.empty()) {
   1030     bool filter_removed =
   1031         RemoveFilterFunction(Bus::OnServiceOwnerChangedFilter, this);
   1032     DCHECK(filter_removed);
   1033   }
   1034 }
   1035 
   1036 dbus_bool_t Bus::OnAddWatch(DBusWatch* raw_watch) {
   1037   AssertOnDBusThread();
   1038 
   1039   // watch will be deleted when raw_watch is removed in OnRemoveWatch().
   1040   Watch* watch = new Watch(raw_watch);
   1041   if (watch->IsReadyToBeWatched()) {
   1042     watch->StartWatching();
   1043   }
   1044   ++num_pending_watches_;
   1045   return true;
   1046 }
   1047 
   1048 void Bus::OnRemoveWatch(DBusWatch* raw_watch) {
   1049   AssertOnDBusThread();
   1050 
   1051   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
   1052   delete watch;
   1053   --num_pending_watches_;
   1054 }
   1055 
   1056 void Bus::OnToggleWatch(DBusWatch* raw_watch) {
   1057   AssertOnDBusThread();
   1058 
   1059   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
   1060   if (watch->IsReadyToBeWatched()) {
   1061     watch->StartWatching();
   1062   } else {
   1063     // It's safe to call this if StartWatching() wasn't called, per
   1064     // message_pump_libevent.h.
   1065     watch->StopWatching();
   1066   }
   1067 }
   1068 
   1069 dbus_bool_t Bus::OnAddTimeout(DBusTimeout* raw_timeout) {
   1070   AssertOnDBusThread();
   1071 
   1072   // timeout will be deleted when raw_timeout is removed in
   1073   // OnRemoveTimeoutThunk().
   1074   Timeout* timeout = new Timeout(raw_timeout);
   1075   if (timeout->IsReadyToBeMonitored()) {
   1076     timeout->StartMonitoring(this);
   1077   }
   1078   ++num_pending_timeouts_;
   1079   return true;
   1080 }
   1081 
   1082 void Bus::OnRemoveTimeout(DBusTimeout* raw_timeout) {
   1083   AssertOnDBusThread();
   1084 
   1085   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
   1086   timeout->Complete();
   1087   --num_pending_timeouts_;
   1088 }
   1089 
   1090 void Bus::OnToggleTimeout(DBusTimeout* raw_timeout) {
   1091   AssertOnDBusThread();
   1092 
   1093   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
   1094   if (timeout->IsReadyToBeMonitored()) {
   1095     timeout->StartMonitoring(this);
   1096   } else {
   1097     timeout->StopMonitoring();
   1098   }
   1099 }
   1100 
   1101 void Bus::OnDispatchStatusChanged(DBusConnection* connection,
   1102                                   DBusDispatchStatus status) {
   1103   DCHECK_EQ(connection, connection_);
   1104   AssertOnDBusThread();
   1105 
   1106   // We cannot call ProcessAllIncomingDataIfAny() here, as calling
   1107   // dbus_connection_dispatch() inside DBusDispatchStatusFunction is
   1108   // prohibited by the D-Bus library. Hence, we post a task here instead.
   1109   // See comments for dbus_connection_set_dispatch_status_function().
   1110   GetDBusTaskRunner()->PostTask(FROM_HERE,
   1111                                 base::Bind(&Bus::ProcessAllIncomingDataIfAny,
   1112                                            this));
   1113 }
   1114 
   1115 void Bus::OnConnectionDisconnected(DBusConnection* connection) {
   1116   AssertOnDBusThread();
   1117 
   1118   if (!on_disconnected_closure_.is_null())
   1119     GetOriginTaskRunner()->PostTask(FROM_HERE, on_disconnected_closure_);
   1120 
   1121   if (!connection)
   1122     return;
   1123   DCHECK(!dbus_connection_get_is_connected(connection));
   1124 
   1125   ShutdownAndBlock();
   1126 }
   1127 
   1128 void Bus::OnServiceOwnerChanged(DBusMessage* message) {
   1129   DCHECK(message);
   1130   AssertOnDBusThread();
   1131 
   1132   // |message| will be unrefed on exit of the function. Increment the
   1133   // reference so we can use it in Signal::FromRawMessage() below.
   1134   dbus_message_ref(message);
   1135   scoped_ptr<Signal> signal(Signal::FromRawMessage(message));
   1136 
   1137   // Confirm the validity of the NameOwnerChanged signal.
   1138   if (signal->GetMember() != kNameOwnerChangedSignal ||
   1139       signal->GetInterface() != DBUS_INTERFACE_DBUS ||
   1140       signal->GetSender() != DBUS_SERVICE_DBUS) {
   1141     return;
   1142   }
   1143 
   1144   MessageReader reader(signal.get());
   1145   std::string service_name;
   1146   std::string old_owner;
   1147   std::string new_owner;
   1148   if (!reader.PopString(&service_name) ||
   1149       !reader.PopString(&old_owner) ||
   1150       !reader.PopString(&new_owner)) {
   1151     return;
   1152   }
   1153 
   1154   ServiceOwnerChangedListenerMap::const_iterator it =
   1155       service_owner_changed_listener_map_.find(service_name);
   1156   if (it == service_owner_changed_listener_map_.end())
   1157     return;
   1158 
   1159   const std::vector<GetServiceOwnerCallback>& callbacks = it->second;
   1160   for (size_t i = 0; i < callbacks.size(); ++i) {
   1161     GetOriginTaskRunner()->PostTask(FROM_HERE,
   1162                                     base::Bind(callbacks[i], new_owner));
   1163   }
   1164 }
   1165 
   1166 // static
   1167 dbus_bool_t Bus::OnAddWatchThunk(DBusWatch* raw_watch, void* data) {
   1168   Bus* self = static_cast<Bus*>(data);
   1169   return self->OnAddWatch(raw_watch);
   1170 }
   1171 
   1172 // static
   1173 void Bus::OnRemoveWatchThunk(DBusWatch* raw_watch, void* data) {
   1174   Bus* self = static_cast<Bus*>(data);
   1175   self->OnRemoveWatch(raw_watch);
   1176 }
   1177 
   1178 // static
   1179 void Bus::OnToggleWatchThunk(DBusWatch* raw_watch, void* data) {
   1180   Bus* self = static_cast<Bus*>(data);
   1181   self->OnToggleWatch(raw_watch);
   1182 }
   1183 
   1184 // static
   1185 dbus_bool_t Bus::OnAddTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
   1186   Bus* self = static_cast<Bus*>(data);
   1187   return self->OnAddTimeout(raw_timeout);
   1188 }
   1189 
   1190 // static
   1191 void Bus::OnRemoveTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
   1192   Bus* self = static_cast<Bus*>(data);
   1193   self->OnRemoveTimeout(raw_timeout);
   1194 }
   1195 
   1196 // static
   1197 void Bus::OnToggleTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
   1198   Bus* self = static_cast<Bus*>(data);
   1199   self->OnToggleTimeout(raw_timeout);
   1200 }
   1201 
   1202 // static
   1203 void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
   1204                                        DBusDispatchStatus status,
   1205                                        void* data) {
   1206   Bus* self = static_cast<Bus*>(data);
   1207   self->OnDispatchStatusChanged(connection, status);
   1208 }
   1209 
   1210 // static
   1211 DBusHandlerResult Bus::OnConnectionDisconnectedFilter(
   1212     DBusConnection* connection,
   1213     DBusMessage* message,
   1214     void* data) {
   1215   if (dbus_message_is_signal(message,
   1216                              DBUS_INTERFACE_LOCAL,
   1217                              kDisconnectedSignal)) {
   1218     Bus* self = static_cast<Bus*>(data);
   1219     self->OnConnectionDisconnected(connection);
   1220     return DBUS_HANDLER_RESULT_HANDLED;
   1221   }
   1222   return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
   1223 }
   1224 
   1225 // static
   1226 DBusHandlerResult Bus::OnServiceOwnerChangedFilter(
   1227     DBusConnection* connection,
   1228     DBusMessage* message,
   1229     void* data) {
   1230   if (dbus_message_is_signal(message,
   1231                              DBUS_INTERFACE_DBUS,
   1232                              kNameOwnerChangedSignal)) {
   1233     Bus* self = static_cast<Bus*>(data);
   1234     self->OnServiceOwnerChanged(message);
   1235   }
   1236   // Always return unhandled to let others, e.g. ObjectProxies, handle the same
   1237   // signal.
   1238   return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
   1239 }
   1240 
   1241 }  // namespace dbus
   1242