dbus/bus.cc

   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.h"
  10 #include "base/message_loop_proxy.h"
  11 #include "base/stl_util.h"
  12 #include "base/threading/thread.h"
  13 #include "base/threading/thread_restrictions.h"
  14 #include "base/time.h"
  15 #include "dbus/exported_object.h"
  16 #include "dbus/object_manager.h"
  17 #include "dbus/object_path.h"
  18 #include "dbus/object_proxy.h"
  19 #include "dbus/scoped_dbus_error.h"
  20
  21 namespace dbus {
  22
  23 namespace {
  24
  25 const char kDisconnectedSignal[] = "Disconnected";
  26 const char kDisconnectedMatchRule[] =
  27     "type='signal', path='/org/freedesktop/DBus/Local',"
  28     "interface='org.freedesktop.DBus.Local', member='Disconnected'";
  29
  30 // The class is used for watching the file descriptor used for D-Bus
  31 // communication.
  32 class Watch : public base::MessagePumpLibevent::Watcher {
  33  public:
  34   explicit Watch(DBusWatch* watch)
  35       : raw_watch_(watch) {
  36     dbus_watch_set_data(raw_watch_, this, NULL);
  37   }
  38
  39   virtual ~Watch() {
  40     dbus_watch_set_data(raw_watch_, NULL, NULL);
  41   }
  42
  43   // Returns true if the underlying file descriptor is ready to be watched.
  44   bool IsReadyToBeWatched() {
  45     return dbus_watch_get_enabled(raw_watch_);
  46   }
  47
  48   // Starts watching the underlying file descriptor.
  49   void StartWatching() {
  50     const int file_descriptor = dbus_watch_get_unix_fd(raw_watch_);
  51     const int flags = dbus_watch_get_flags(raw_watch_);
  52
  53     MessageLoopForIO::Mode mode = MessageLoopForIO::WATCH_READ;
  54     if ((flags & DBUS_WATCH_READABLE) && (flags & DBUS_WATCH_WRITABLE))
  55       mode = MessageLoopForIO::WATCH_READ_WRITE;
  56     else if (flags & DBUS_WATCH_READABLE)
  57       mode = MessageLoopForIO::WATCH_READ;
  58     else if (flags & DBUS_WATCH_WRITABLE)
  59       mode = MessageLoopForIO::WATCH_WRITE;
  60     else
  61       NOTREACHED();
  62
  63     const bool persistent = true;  // Watch persistently.
  64     const bool success = MessageLoopForIO::current()->WatchFileDescriptor(
  65         file_descriptor,
  66         persistent,
  67         mode,
  68         &file_descriptor_watcher_,
  69         this);
  70     CHECK(success) << "Unable to allocate memory";
  71   }
  72
  73   // Stops watching the underlying file descriptor.
  74   void StopWatching() {
  75     file_descriptor_watcher_.StopWatchingFileDescriptor();
  76   }
  77
  78  private:
  79   // Implement MessagePumpLibevent::Watcher.
  80   virtual void OnFileCanReadWithoutBlocking(int file_descriptor) OVERRIDE {
  81     const bool success = dbus_watch_handle(raw_watch_, DBUS_WATCH_READABLE);
  82     CHECK(success) << "Unable to allocate memory";
  83   }
  84
  85   // Implement MessagePumpLibevent::Watcher.
  86   virtual void OnFileCanWriteWithoutBlocking(int file_descriptor) OVERRIDE {
  87     const bool success = dbus_watch_handle(raw_watch_, DBUS_WATCH_WRITABLE);
  88     CHECK(success) << "Unable to allocate memory";
  89   }
  90
  91   DBusWatch* raw_watch_;
  92   base::MessagePumpLibevent::FileDescriptorWatcher file_descriptor_watcher_;
  93 };
  94
  95 // The class is used for monitoring the timeout used for D-Bus method
  96 // calls.
  97 //
  98 // Unlike Watch, Timeout is a ref counted object, to ensure that |this| of
  99 // the object is is alive when HandleTimeout() is called. It's unlikely
 100 // but it may be possible that HandleTimeout() is called after
 101 // Bus::OnRemoveTimeout(). That's why we don't simply delete the object in
 102 // Bus::OnRemoveTimeout().
 103 class Timeout : public base::RefCountedThreadSafe<Timeout> {
 104  public:
 105   explicit Timeout(DBusTimeout* timeout)
 106       : raw_timeout_(timeout),
 107         monitoring_is_active_(false),
 108         is_completed(false) {
 109     dbus_timeout_set_data(raw_timeout_, this, NULL);
 110     AddRef();  // Balanced on Complete().
 111   }
 112
 113   // Returns true if the timeout is ready to be monitored.
 114   bool IsReadyToBeMonitored() {
 115     return dbus_timeout_get_enabled(raw_timeout_);
 116   }
 117
 118   // Starts monitoring the timeout.
 119   void StartMonitoring(dbus::Bus* bus) {
 120     bus->PostDelayedTaskToDBusThread(FROM_HERE,
 121                                      base::Bind(&Timeout::HandleTimeout,
 122                                                 this),
 123                                      GetInterval());
 124     monitoring_is_active_ = true;
 125   }
 126
 127   // Stops monitoring the timeout.
 128   void StopMonitoring() {
 129     // We cannot take back the delayed task we posted in
 130     // StartMonitoring(), so we just mark the monitoring is inactive now.
 131     monitoring_is_active_ = false;
 132   }
 133
 134   // Returns the interval.
 135   base::TimeDelta GetInterval() {
 136     return base::TimeDelta::FromMilliseconds(
 137         dbus_timeout_get_interval(raw_timeout_));
 138   }
 139
 140   // Cleans up the raw_timeout and marks that timeout is completed.
 141   // See the class comment above for why we are doing this.
 142   void Complete() {
 143     dbus_timeout_set_data(raw_timeout_, NULL, NULL);
 144     is_completed = true;
 145     Release();
 146   }
 147
 148  private:
 149   friend class base::RefCountedThreadSafe<Timeout>;
 150   ~Timeout() {
 151   }
 152
 153   // Handles the timeout.
 154   void HandleTimeout() {
 155     // If the timeout is marked completed, we should do nothing. This can
 156     // occur if this function is called after Bus::OnRemoveTimeout().
 157     if (is_completed)
 158       return;
 159     // Skip if monitoring is canceled.
 160     if (!monitoring_is_active_)
 161       return;
 162
 163     const bool success = dbus_timeout_handle(raw_timeout_);
 164     CHECK(success) << "Unable to allocate memory";
 165   }
 166
 167   DBusTimeout* raw_timeout_;
 168   bool monitoring_is_active_;
 169   bool is_completed;
 170 };
 171
 172 }  // namespace
 173
 174 Bus::Options::Options()
 175   : bus_type(SESSION),
 176     connection_type(PRIVATE) {
 177 }
 178
 179 Bus::Options::~Options() {
 180 }
 181
 182 Bus::Bus(const Options& options)
 183     : bus_type_(options.bus_type),
 184       connection_type_(options.connection_type),
 185       dbus_task_runner_(options.dbus_task_runner),
 186       on_shutdown_(false /* manual_reset */, false /* initially_signaled */),
 187       connection_(NULL),
 188       origin_thread_id_(base::PlatformThread::CurrentId()),
 189       async_operations_set_up_(false),
 190       shutdown_completed_(false),
 191       num_pending_watches_(0),
 192       num_pending_timeouts_(0),
 193       address_(options.address),
 194       on_disconnected_closure_(options.disconnected_callback) {
 195   // This is safe to call multiple times.
 196   dbus_threads_init_default();
 197   // The origin message loop is unnecessary if the client uses synchronous
 198   // functions only.
 199   if (MessageLoop::current())
 200     origin_task_runner_ =  MessageLoop::current()->message_loop_proxy();
 201 }
 202
 203 Bus::~Bus() {
 204   DCHECK(!connection_);
 205   DCHECK(owned_service_names_.empty());
 206   DCHECK(match_rules_added_.empty());
 207   DCHECK(filter_functions_added_.empty());
 208   DCHECK(registered_object_paths_.empty());
 209   DCHECK_EQ(0, num_pending_watches_);
 210   // TODO(satorux): This check fails occasionally in browser_tests for tests
 211   // that run very quickly. Perhaps something does not have time to clean up.
 212   // Despite the check failing, the tests seem to run fine. crosbug.com/23416
 213   // DCHECK_EQ(0, num_pending_timeouts_);
 214 }
 215
 216 ObjectProxy* Bus::GetObjectProxy(const std::string& service_name,
 217                                  const ObjectPath& object_path) {
 218   return GetObjectProxyWithOptions(service_name, object_path,
 219                                    ObjectProxy::DEFAULT_OPTIONS);
 220 }
 221
 222 ObjectProxy* Bus::GetObjectProxyWithOptions(const std::string& service_name,
 223                                             const dbus::ObjectPath& object_path,
 224                                             int options) {
 225   AssertOnOriginThread();
 226
 227   // Check if we already have the requested object proxy.
 228   const ObjectProxyTable::key_type key(service_name + object_path.value(),
 229                                        options);
 230   ObjectProxyTable::iterator iter = object_proxy_table_.find(key);
 231   if (iter != object_proxy_table_.end()) {
 232     return iter->second;
 233   }
 234
 235   scoped_refptr<ObjectProxy> object_proxy =
 236       new ObjectProxy(this, service_name, object_path, options);
 237   object_proxy_table_[key] = object_proxy;
 238
 239   return object_proxy.get();
 240 }
 241
 242 bool Bus::RemoveObjectProxy(const std::string& service_name,
 243                             const ObjectPath& object_path,
 244                             const base::Closure& callback) {
 245   return RemoveObjectProxyWithOptions(service_name, object_path,
 246                                       ObjectProxy::DEFAULT_OPTIONS,
 247                                       callback);
 248 }
 249
 250 bool Bus::RemoveObjectProxyWithOptions(const std::string& service_name,
 251                                        const dbus::ObjectPath& object_path,
 252                                        int options,
 253                                        const base::Closure& callback) {
 254   AssertOnOriginThread();
 255
 256   // Check if we have the requested object proxy.
 257   const ObjectProxyTable::key_type key(service_name + object_path.value(),
 258                                        options);
 259   ObjectProxyTable::iterator iter = object_proxy_table_.find(key);
 260   if (iter != object_proxy_table_.end()) {
 261     // Object is present. Remove it now and Detach in the DBus thread.
 262     PostTaskToDBusThread(FROM_HERE, base::Bind(
 263         &Bus::RemoveObjectProxyInternal,
 264         this, iter->second, callback));
 265
 266     object_proxy_table_.erase(iter);
 267     return true;
 268   }
 269   return false;
 270 }
 271
 272 void Bus::RemoveObjectProxyInternal(
 273     scoped_refptr<dbus::ObjectProxy> object_proxy,
 274     const base::Closure& callback) {
 275   AssertOnDBusThread();
 276
 277   object_proxy.get()->Detach();
 278
 279   PostTaskToOriginThread(FROM_HERE, callback);
 280 }
 281
 282 ExportedObject* Bus::GetExportedObject(const ObjectPath& object_path) {
 283   AssertOnOriginThread();
 284
 285   // Check if we already have the requested exported object.
 286   ExportedObjectTable::iterator iter = exported_object_table_.find(object_path);
 287   if (iter != exported_object_table_.end()) {
 288     return iter->second;
 289   }
 290
 291   scoped_refptr<ExportedObject> exported_object =
 292       new ExportedObject(this, object_path);
 293   exported_object_table_[object_path] = exported_object;
 294
 295   return exported_object.get();
 296 }
 297
 298 void Bus::UnregisterExportedObject(const ObjectPath& object_path) {
 299   AssertOnOriginThread();
 300
 301   // Remove the registered object from the table first, to allow a new
 302   // GetExportedObject() call to return a new object, rather than this one.
 303   ExportedObjectTable::iterator iter = exported_object_table_.find(object_path);
 304   if (iter == exported_object_table_.end())
 305     return;
 306
 307   scoped_refptr<ExportedObject> exported_object = iter->second;
 308   exported_object_table_.erase(iter);
 309
 310   // Post the task to perform the final unregistration to the D-Bus thread.
 311   // Since the registration also happens on the D-Bus thread in
 312   // TryRegisterObjectPath(), and the task runner we post to is a
 313   // SequencedTaskRunner, there is a guarantee that this will happen before any
 314   // future registration call.
 315   PostTaskToDBusThread(FROM_HERE,
 316                        base::Bind(&Bus::UnregisterExportedObjectInternal,
 317                                   this, exported_object));
 318 }
 319
 320 void Bus::UnregisterExportedObjectInternal(
 321     scoped_refptr<dbus::ExportedObject> exported_object) {
 322   AssertOnDBusThread();
 323
 324   exported_object->Unregister();
 325 }
 326
 327 ObjectManager* Bus::GetObjectManager(const std::string& service_name,
 328                                      const ObjectPath& object_path) {
 329   AssertOnOriginThread();
 330
 331   // Check if we already have the requested object manager.
 332   const ObjectManagerTable::key_type key(service_name + object_path.value());
 333   ObjectManagerTable::iterator iter = object_manager_table_.find(key);
 334   if (iter != object_manager_table_.end()) {
 335     return iter->second;
 336   }
 337
 338   scoped_refptr<ObjectManager> object_manager =
 339       new ObjectManager(this, service_name, object_path);
 340   object_manager_table_[key] = object_manager;
 341
 342   return object_manager.get();
 343 }
 344
 345 void Bus::RemoveObjectManager(const std::string& service_name,
 346                               const ObjectPath& object_path) {
 347   AssertOnOriginThread();
 348
 349   const ObjectManagerTable::key_type key(service_name + object_path.value());
 350   ObjectManagerTable::iterator iter = object_manager_table_.find(key);
 351   if (iter == object_manager_table_.end())
 352     return;
 353
 354   scoped_refptr<ObjectManager> object_manager = iter->second;
 355   object_manager_table_.erase(iter);
 356 }
 357
 358 void Bus::GetManagedObjects() {
 359   for (ObjectManagerTable::iterator iter = object_manager_table_.begin();
 360        iter != object_manager_table_.end(); ++iter) {
 361     iter->second->GetManagedObjects();
 362   }
 363 }
 364
 365 bool Bus::Connect() {
 366   // dbus_bus_get_private() and dbus_bus_get() are blocking calls.
 367   AssertOnDBusThread();
 368
 369   // Check if it's already initialized.
 370   if (connection_)
 371     return true;
 372
 373   ScopedDBusError error;
 374   if (bus_type_ == CUSTOM_ADDRESS) {
 375     if (connection_type_ == PRIVATE) {
 376       connection_ = dbus_connection_open_private(address_.c_str(), error.get());
 377     } else {
 378       connection_ = dbus_connection_open(address_.c_str(), error.get());
 379     }
 380   } else {
 381     const DBusBusType dbus_bus_type = static_cast<DBusBusType>(bus_type_);
 382     if (connection_type_ == PRIVATE) {
 383       connection_ = dbus_bus_get_private(dbus_bus_type, error.get());
 384     } else {
 385       connection_ = dbus_bus_get(dbus_bus_type, error.get());
 386     }
 387   }
 388   if (!connection_) {
 389     LOG(ERROR) << "Failed to connect to the bus: "
 390                << (error.is_set() ? error.message() : "");
 391     return false;
 392   }
 393
 394   if (bus_type_ == CUSTOM_ADDRESS) {
 395     // We should call dbus_bus_register here, otherwise unique name can not be
 396     // acquired. According to dbus specification, it is responsible to call
 397     // org.freedesktop.DBus.Hello method at the beging of bus connection to
 398     // acquire unique name. In the case of dbus_bus_get, dbus_bus_register is
 399     // called internally.
 400     if (!dbus_bus_register(connection_, error.get())) {
 401       LOG(ERROR) << "Failed to register the bus component: "
 402                  << (error.is_set() ? error.message() : "");
 403       return false;
 404     }
 405   }
 406   // We shouldn't exit on the disconnected signal.
 407   dbus_connection_set_exit_on_disconnect(connection_, false);
 408
 409   // Watch Disconnected signal.
 410   AddFilterFunction(Bus::OnConnectionDisconnectedFilter, this);
 411   AddMatch(kDisconnectedMatchRule, error.get());
 412
 413   return true;
 414 }
 415
 416 void Bus::ClosePrivateConnection() {
 417   // dbus_connection_close is blocking call.
 418   AssertOnDBusThread();
 419   DCHECK_EQ(PRIVATE, connection_type_)
 420       << "non-private connection should not be closed";
 421   dbus_connection_close(connection_);
 422 }
 423
 424 void Bus::ShutdownAndBlock() {
 425   AssertOnDBusThread();
 426
 427   if (shutdown_completed_)
 428     return;  // Already shutdowned, just return.
 429
 430   // Unregister the exported objects.
 431   for (ExportedObjectTable::iterator iter = exported_object_table_.begin();
 432        iter != exported_object_table_.end(); ++iter) {
 433     iter->second->Unregister();
 434   }
 435
 436   // Release all service names.
 437   for (std::set<std::string>::iterator iter = owned_service_names_.begin();
 438        iter != owned_service_names_.end();) {
 439     // This is a bit tricky but we should increment the iter here as
 440     // ReleaseOwnership() may remove |service_name| from the set.
 441     const std::string& service_name = *iter++;
 442     ReleaseOwnership(service_name);
 443   }
 444   if (!owned_service_names_.empty()) {
 445     LOG(ERROR) << "Failed to release all service names. # of services left: "
 446                << owned_service_names_.size();
 447   }
 448
 449   // Detach from the remote objects.
 450   for (ObjectProxyTable::iterator iter = object_proxy_table_.begin();
 451        iter != object_proxy_table_.end(); ++iter) {
 452     iter->second->Detach();
 453   }
 454
 455   // Release object proxies and exported objects here. We should do this
 456   // here rather than in the destructor to avoid memory leaks due to
 457   // cyclic references.
 458   object_proxy_table_.clear();
 459   exported_object_table_.clear();
 460
 461   // Private connection should be closed.
 462   if (connection_) {
 463     // Remove Disconnected watcher.
 464     ScopedDBusError error;
 465     RemoveFilterFunction(Bus::OnConnectionDisconnectedFilter, this);
 466     RemoveMatch(kDisconnectedMatchRule, error.get());
 467
 468     if (connection_type_ == PRIVATE)
 469       ClosePrivateConnection();
 470     // dbus_connection_close() won't unref.
 471     dbus_connection_unref(connection_);
 472   }
 473
 474   connection_ = NULL;
 475   shutdown_completed_ = true;
 476 }
 477
 478 void Bus::ShutdownOnDBusThreadAndBlock() {
 479   AssertOnOriginThread();
 480   DCHECK(dbus_task_runner_.get());
 481
 482   PostTaskToDBusThread(FROM_HERE, base::Bind(
 483       &Bus::ShutdownOnDBusThreadAndBlockInternal,
 484       this));
 485
 486   // http://crbug.com/125222
 487   base::ThreadRestrictions::ScopedAllowWait allow_wait;
 488
 489   // Wait until the shutdown is complete on the D-Bus thread.
 490   // The shutdown should not hang, but set timeout just in case.
 491   const int kTimeoutSecs = 3;
 492   const base::TimeDelta timeout(base::TimeDelta::FromSeconds(kTimeoutSecs));
 493   const bool signaled = on_shutdown_.TimedWait(timeout);
 494   LOG_IF(ERROR, !signaled) << "Failed to shutdown the bus";
 495 }
 496
 497 void Bus::RequestOwnership(const std::string& service_name,
 498                            OnOwnershipCallback on_ownership_callback) {
 499   AssertOnOriginThread();
 500
 501   PostTaskToDBusThread(FROM_HERE, base::Bind(
 502       &Bus::RequestOwnershipInternal,
 503       this, service_name, on_ownership_callback));
 504 }
 505
 506 void Bus::RequestOwnershipInternal(const std::string& service_name,
 507                                    OnOwnershipCallback on_ownership_callback) {
 508   AssertOnDBusThread();
 509
 510   bool success = Connect();
 511   if (success)
 512     success = RequestOwnershipAndBlock(service_name);
 513
 514   PostTaskToOriginThread(FROM_HERE,
 515                          base::Bind(on_ownership_callback,
 516                                     service_name,
 517                                     success));
 518 }
 519
 520 bool Bus::RequestOwnershipAndBlock(const std::string& service_name) {
 521   DCHECK(connection_);
 522   // dbus_bus_request_name() is a blocking call.
 523   AssertOnDBusThread();
 524
 525   // Check if we already own the service name.
 526   if (owned_service_names_.find(service_name) != owned_service_names_.end()) {
 527     return true;
 528   }
 529
 530   ScopedDBusError error;
 531   const int result = dbus_bus_request_name(connection_,
 532                                            service_name.c_str(),
 533                                            DBUS_NAME_FLAG_DO_NOT_QUEUE,
 534                                            error.get());
 535   if (result != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER) {
 536     LOG(ERROR) << "Failed to get the ownership of " << service_name << ": "
 537                << (error.is_set() ? error.message() : "");
 538     return false;
 539   }
 540   owned_service_names_.insert(service_name);
 541   return true;
 542 }
 543
 544 bool Bus::ReleaseOwnership(const std::string& service_name) {
 545   DCHECK(connection_);
 546   // dbus_bus_request_name() is a blocking call.
 547   AssertOnDBusThread();
 548
 549   // Check if we already own the service name.
 550   std::set<std::string>::iterator found =
 551       owned_service_names_.find(service_name);
 552   if (found == owned_service_names_.end()) {
 553     LOG(ERROR) << service_name << " is not owned by the bus";
 554     return false;
 555   }
 556
 557   ScopedDBusError error;
 558   const int result = dbus_bus_release_name(connection_, service_name.c_str(),
 559                                            error.get());
 560   if (result == DBUS_RELEASE_NAME_REPLY_RELEASED) {
 561     owned_service_names_.erase(found);
 562     return true;
 563   } else {
 564     LOG(ERROR) << "Failed to release the ownership of " << service_name << ": "
 565                << (error.is_set() ? error.message() : "");
 566     return false;
 567   }
 568 }
 569
 570 bool Bus::SetUpAsyncOperations() {
 571   DCHECK(connection_);
 572   AssertOnDBusThread();
 573
 574   if (async_operations_set_up_)
 575     return true;
 576
 577   // Process all the incoming data if any, so that OnDispatchStatus() will
 578   // be called when the incoming data is ready.
 579   ProcessAllIncomingDataIfAny();
 580
 581   bool success = dbus_connection_set_watch_functions(connection_,
 582                                                      &Bus::OnAddWatchThunk,
 583                                                      &Bus::OnRemoveWatchThunk,
 584                                                      &Bus::OnToggleWatchThunk,
 585                                                      this,
 586                                                      NULL);
 587   CHECK(success) << "Unable to allocate memory";
 588
 589   success = dbus_connection_set_timeout_functions(connection_,
 590                                                   &Bus::OnAddTimeoutThunk,
 591                                                   &Bus::OnRemoveTimeoutThunk,
 592                                                   &Bus::OnToggleTimeoutThunk,
 593                                                   this,
 594                                                   NULL);
 595   CHECK(success) << "Unable to allocate memory";
 596
 597   dbus_connection_set_dispatch_status_function(
 598       connection_,
 599       &Bus::OnDispatchStatusChangedThunk,
 600       this,
 601       NULL);
 602
 603   async_operations_set_up_ = true;
 604
 605   return true;
 606 }
 607
 608 DBusMessage* Bus::SendWithReplyAndBlock(DBusMessage* request,
 609                                         int timeout_ms,
 610                                         DBusError* error) {
 611   DCHECK(connection_);
 612   AssertOnDBusThread();
 613
 614   return dbus_connection_send_with_reply_and_block(
 615       connection_, request, timeout_ms, error);
 616 }
 617
 618 void Bus::SendWithReply(DBusMessage* request,
 619                         DBusPendingCall** pending_call,
 620                         int timeout_ms) {
 621   DCHECK(connection_);
 622   AssertOnDBusThread();
 623
 624   const bool success = dbus_connection_send_with_reply(
 625       connection_, request, pending_call, timeout_ms);
 626   CHECK(success) << "Unable to allocate memory";
 627 }
 628
 629 void Bus::Send(DBusMessage* request, uint32* serial) {
 630   DCHECK(connection_);
 631   AssertOnDBusThread();
 632
 633   const bool success = dbus_connection_send(connection_, request, serial);
 634   CHECK(success) << "Unable to allocate memory";
 635 }
 636
 637 bool Bus::AddFilterFunction(DBusHandleMessageFunction filter_function,
 638                             void* user_data) {
 639   DCHECK(connection_);
 640   AssertOnDBusThread();
 641
 642   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
 643       std::make_pair(filter_function, user_data);
 644   if (filter_functions_added_.find(filter_data_pair) !=
 645       filter_functions_added_.end()) {
 646     VLOG(1) << "Filter function already exists: " << filter_function
 647             << " with associated data: " << user_data;
 648     return false;
 649   }
 650
 651   const bool success = dbus_connection_add_filter(
 652       connection_, filter_function, user_data, NULL);
 653   CHECK(success) << "Unable to allocate memory";
 654   filter_functions_added_.insert(filter_data_pair);
 655   return true;
 656 }
 657
 658 bool Bus::RemoveFilterFunction(DBusHandleMessageFunction filter_function,
 659                                void* user_data) {
 660   DCHECK(connection_);
 661   AssertOnDBusThread();
 662
 663   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
 664       std::make_pair(filter_function, user_data);
 665   if (filter_functions_added_.find(filter_data_pair) ==
 666       filter_functions_added_.end()) {
 667     VLOG(1) << "Requested to remove an unknown filter function: "
 668             << filter_function
 669             << " with associated data: " << user_data;
 670     return false;
 671   }
 672
 673   dbus_connection_remove_filter(connection_, filter_function, user_data);
 674   filter_functions_added_.erase(filter_data_pair);
 675   return true;
 676 }
 677
 678 void Bus::AddMatch(const std::string& match_rule, DBusError* error) {
 679   DCHECK(connection_);
 680   AssertOnDBusThread();
 681
 682   std::map<std::string, int>::iterator iter =
 683       match_rules_added_.find(match_rule);
 684   if (iter != match_rules_added_.end()) {
 685     // The already existing rule's counter is incremented.
 686     iter->second++;
 687
 688     VLOG(1) << "Match rule already exists: " << match_rule;
 689     return;
 690   }
 691
 692   dbus_bus_add_match(connection_, match_rule.c_str(), error);
 693   match_rules_added_[match_rule] = 1;
 694 }
 695
 696 bool Bus::RemoveMatch(const std::string& match_rule, DBusError* error) {
 697   DCHECK(connection_);
 698   AssertOnDBusThread();
 699
 700   std::map<std::string, int>::iterator iter =
 701       match_rules_added_.find(match_rule);
 702   if (iter == match_rules_added_.end()) {
 703     LOG(ERROR) << "Requested to remove an unknown match rule: " << match_rule;
 704     return false;
 705   }
 706
 707   // The rule's counter is decremented and the rule is deleted when reachs 0.
 708   iter->second--;
 709   if (iter->second == 0) {
 710     dbus_bus_remove_match(connection_, match_rule.c_str(), error);
 711     match_rules_added_.erase(match_rule);
 712   }
 713   return true;
 714 }
 715
 716 bool Bus::TryRegisterObjectPath(const ObjectPath& object_path,
 717                                 const DBusObjectPathVTable* vtable,
 718                                 void* user_data,
 719                                 DBusError* error) {
 720   DCHECK(connection_);
 721   AssertOnDBusThread();
 722
 723   if (registered_object_paths_.find(object_path) !=
 724       registered_object_paths_.end()) {
 725     LOG(ERROR) << "Object path already registered: " << object_path.value();
 726     return false;
 727   }
 728
 729   const bool success = dbus_connection_try_register_object_path(
 730       connection_,
 731       object_path.value().c_str(),
 732       vtable,
 733       user_data,
 734       error);
 735   if (success)
 736     registered_object_paths_.insert(object_path);
 737   return success;
 738 }
 739
 740 void Bus::UnregisterObjectPath(const ObjectPath& object_path) {
 741   DCHECK(connection_);
 742   AssertOnDBusThread();
 743
 744   if (registered_object_paths_.find(object_path) ==
 745       registered_object_paths_.end()) {
 746     LOG(ERROR) << "Requested to unregister an unknown object path: "
 747                << object_path.value();
 748     return;
 749   }
 750
 751   const bool success = dbus_connection_unregister_object_path(
 752       connection_,
 753       object_path.value().c_str());
 754   CHECK(success) << "Unable to allocate memory";
 755   registered_object_paths_.erase(object_path);
 756 }
 757
 758 void Bus::ShutdownOnDBusThreadAndBlockInternal() {
 759   AssertOnDBusThread();
 760
 761   ShutdownAndBlock();
 762   on_shutdown_.Signal();
 763 }
 764
 765 void Bus::ProcessAllIncomingDataIfAny() {
 766   AssertOnDBusThread();
 767
 768   // As mentioned at the class comment in .h file, connection_ can be NULL.
 769   if (!connection_)
 770     return;
 771
 772   // It is safe and necessary to call dbus_connection_get_dispatch_status even
 773   // if the connection is lost. Otherwise we will miss "Disconnected" signal.
 774   // (crbug.com/174431)
 775   if (dbus_connection_get_dispatch_status(connection_) ==
 776       DBUS_DISPATCH_DATA_REMAINS) {
 777     while (dbus_connection_dispatch(connection_) ==
 778            DBUS_DISPATCH_DATA_REMAINS);
 779   }
 780 }
 781
 782 void Bus::PostTaskToOriginThread(const tracked_objects::Location& from_here,
 783                                  const base::Closure& task) {
 784   DCHECK(origin_task_runner_.get());
 785   if (!origin_task_runner_->PostTask(from_here, task)) {
 786     LOG(WARNING) << "Failed to post a task to the origin message loop";
 787   }
 788 }
 789
 790 void Bus::PostTaskToDBusThread(const tracked_objects::Location& from_here,
 791                                const base::Closure& task) {
 792   if (dbus_task_runner_.get()) {
 793     if (!dbus_task_runner_->PostTask(from_here, task)) {
 794       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
 795     }
 796   } else {
 797     DCHECK(origin_task_runner_.get());
 798     if (!origin_task_runner_->PostTask(from_here, task)) {
 799       LOG(WARNING) << "Failed to post a task to the origin message loop";
 800     }
 801   }
 802 }
 803
 804 void Bus::PostDelayedTaskToDBusThread(
 805     const tracked_objects::Location& from_here,
 806     const base::Closure& task,
 807     base::TimeDelta delay) {
 808   if (dbus_task_runner_.get()) {
 809     if (!dbus_task_runner_->PostDelayedTask(
 810             from_here, task, delay)) {
 811       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
 812     }
 813   } else {
 814     DCHECK(origin_task_runner_.get());
 815     if (!origin_task_runner_->PostDelayedTask(from_here, task, delay)) {
 816       LOG(WARNING) << "Failed to post a task to the origin message loop";
 817     }
 818   }
 819 }
 820
 821 bool Bus::HasDBusThread() {
 822   return dbus_task_runner_.get() != NULL;
 823 }
 824
 825 void Bus::AssertOnOriginThread() {
 826   DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
 827 }
 828
 829 void Bus::AssertOnDBusThread() {
 830   base::ThreadRestrictions::AssertIOAllowed();
 831
 832   if (dbus_task_runner_.get()) {
 833     DCHECK(dbus_task_runner_->RunsTasksOnCurrentThread());
 834   } else {
 835     AssertOnOriginThread();
 836   }
 837 }
 838
 839 dbus_bool_t Bus::OnAddWatch(DBusWatch* raw_watch) {
 840   AssertOnDBusThread();
 841
 842   // watch will be deleted when raw_watch is removed in OnRemoveWatch().
 843   Watch* watch = new Watch(raw_watch);
 844   if (watch->IsReadyToBeWatched()) {
 845     watch->StartWatching();
 846   }
 847   ++num_pending_watches_;
 848   return true;
 849 }
 850
 851 void Bus::OnRemoveWatch(DBusWatch* raw_watch) {
 852   AssertOnDBusThread();
 853
 854   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
 855   delete watch;
 856   --num_pending_watches_;
 857 }
 858
 859 void Bus::OnToggleWatch(DBusWatch* raw_watch) {
 860   AssertOnDBusThread();
 861
 862   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
 863   if (watch->IsReadyToBeWatched()) {
 864     watch->StartWatching();
 865   } else {
 866     // It's safe to call this if StartWatching() wasn't called, per
 867     // message_pump_libevent.h.
 868     watch->StopWatching();
 869   }
 870 }
 871
 872 dbus_bool_t Bus::OnAddTimeout(DBusTimeout* raw_timeout) {
 873   AssertOnDBusThread();
 874
 875   // timeout will be deleted when raw_timeout is removed in
 876   // OnRemoveTimeoutThunk().
 877   Timeout* timeout = new Timeout(raw_timeout);
 878   if (timeout->IsReadyToBeMonitored()) {
 879     timeout->StartMonitoring(this);
 880   }
 881   ++num_pending_timeouts_;
 882   return true;
 883 }
 884
 885 void Bus::OnRemoveTimeout(DBusTimeout* raw_timeout) {
 886   AssertOnDBusThread();
 887
 888   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
 889   timeout->Complete();
 890   --num_pending_timeouts_;
 891 }
 892
 893 void Bus::OnToggleTimeout(DBusTimeout* raw_timeout) {
 894   AssertOnDBusThread();
 895
 896   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
 897   if (timeout->IsReadyToBeMonitored()) {
 898     timeout->StartMonitoring(this);
 899   } else {
 900     timeout->StopMonitoring();
 901   }
 902 }
 903
 904 void Bus::OnDispatchStatusChanged(DBusConnection* connection,
 905                                   DBusDispatchStatus status) {
 906   DCHECK_EQ(connection, connection_);
 907   AssertOnDBusThread();
 908
 909   // We cannot call ProcessAllIncomingDataIfAny() here, as calling
 910   // dbus_connection_dispatch() inside DBusDispatchStatusFunction is
 911   // prohibited by the D-Bus library. Hence, we post a task here instead.
 912   // See comments for dbus_connection_set_dispatch_status_function().
 913   PostTaskToDBusThread(FROM_HERE,
 914                        base::Bind(&Bus::ProcessAllIncomingDataIfAny,
 915                                   this));
 916 }
 917
 918 void Bus::OnConnectionDisconnected(DBusConnection* connection) {
 919   AssertOnDBusThread();
 920
 921   if (!on_disconnected_closure_.is_null())
 922     PostTaskToOriginThread(FROM_HERE, on_disconnected_closure_);
 923
 924   if (!connection)
 925     return;
 926   DCHECK(!dbus_connection_get_is_connected(connection));
 927
 928   ShutdownAndBlock();
 929 }
 930
 931 dbus_bool_t Bus::OnAddWatchThunk(DBusWatch* raw_watch, void* data) {
 932   Bus* self = static_cast<Bus*>(data);
 933   return self->OnAddWatch(raw_watch);
 934 }
 935
 936 void Bus::OnRemoveWatchThunk(DBusWatch* raw_watch, void* data) {
 937   Bus* self = static_cast<Bus*>(data);
 938   self->OnRemoveWatch(raw_watch);
 939 }
 940
 941 void Bus::OnToggleWatchThunk(DBusWatch* raw_watch, void* data) {
 942   Bus* self = static_cast<Bus*>(data);
 943   self->OnToggleWatch(raw_watch);
 944 }
 945
 946 dbus_bool_t Bus::OnAddTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 947   Bus* self = static_cast<Bus*>(data);
 948   return self->OnAddTimeout(raw_timeout);
 949 }
 950
 951 void Bus::OnRemoveTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 952   Bus* self = static_cast<Bus*>(data);
 953   self->OnRemoveTimeout(raw_timeout);
 954 }
 955
 956 void Bus::OnToggleTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 957   Bus* self = static_cast<Bus*>(data);
 958   self->OnToggleTimeout(raw_timeout);
 959 }
 960
 961 void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
 962                                        DBusDispatchStatus status,
 963                                        void* data) {
 964   Bus* self = static_cast<Bus*>(data);
 965   self->OnDispatchStatusChanged(connection, status);
 966 }
 967
 968 DBusHandlerResult Bus::OnConnectionDisconnectedFilter(
 969     DBusConnection* connection,
 970     DBusMessage* message,
 971     void* data) {
 972   if (dbus_message_is_signal(message,
 973                              DBUS_INTERFACE_LOCAL,
 974                              kDisconnectedSignal)) {
 975     Bus* self = static_cast<Bus*>(data);
 976     self->AssertOnDBusThread();
 977     self->OnConnectionDisconnected(connection);
 978     return DBUS_HANDLER_RESULT_HANDLED;
 979   }
 980   return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
 981 }
 982
 983 }  // namespace dbus