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