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