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                << (dbus_error_is_set(error.get()) ? error.message() : "");
 309     return false;
 310   }
 311   // We shouldn't exit on the disconnected signal.
 312   dbus_connection_set_exit_on_disconnect(connection_, false);
 313
 314   return true;
 315 }
 316
 317 void Bus::ShutdownAndBlock() {
 318   AssertOnDBusThread();
 319
 320   // Unregister the exported objects.
 321   for (ExportedObjectTable::iterator iter = exported_object_table_.begin();
 322        iter != exported_object_table_.end(); ++iter) {
 323     iter->second->Unregister();
 324   }
 325
 326   // Release all service names.
 327   for (std::set<std::string>::iterator iter = owned_service_names_.begin();
 328        iter != owned_service_names_.end();) {
 329     // This is a bit tricky but we should increment the iter here as
 330     // ReleaseOwnership() may remove |service_name| from the set.
 331     const std::string& service_name = *iter++;
 332     ReleaseOwnership(service_name);
 333   }
 334   if (!owned_service_names_.empty()) {
 335     LOG(ERROR) << "Failed to release all service names. # of services left: "
 336                << owned_service_names_.size();
 337   }
 338
 339   // Detach from the remote objects.
 340   for (ObjectProxyTable::iterator iter = object_proxy_table_.begin();
 341        iter != object_proxy_table_.end(); ++iter) {
 342     iter->second->Detach();
 343   }
 344
 345   // Release object proxies and exported objects here. We should do this
 346   // here rather than in the destructor to avoid memory leaks due to
 347   // cyclic references.
 348   object_proxy_table_.clear();
 349   exported_object_table_.clear();
 350
 351   // Private connection should be closed.
 352   if (connection_) {
 353     if (connection_type_ == PRIVATE)
 354       dbus_connection_close(connection_);
 355     // dbus_connection_close() won't unref.
 356     dbus_connection_unref(connection_);
 357   }
 358
 359   connection_ = NULL;
 360   shutdown_completed_ = true;
 361 }
 362
 363 void Bus::ShutdownOnDBusThreadAndBlock() {
 364   AssertOnOriginThread();
 365   DCHECK(dbus_thread_message_loop_proxy_.get());
 366
 367   PostTaskToDBusThread(FROM_HERE, base::Bind(
 368       &Bus::ShutdownOnDBusThreadAndBlockInternal,
 369       this));
 370
 371   // http://crbug.com/125222
 372   base::ThreadRestrictions::ScopedAllowWait allow_wait;
 373
 374   // Wait until the shutdown is complete on the D-Bus thread.
 375   // The shutdown should not hang, but set timeout just in case.
 376   const int kTimeoutSecs = 3;
 377   const base::TimeDelta timeout(base::TimeDelta::FromSeconds(kTimeoutSecs));
 378   const bool signaled = on_shutdown_.TimedWait(timeout);
 379   LOG_IF(ERROR, !signaled) << "Failed to shutdown the bus";
 380 }
 381
 382 void Bus::RequestOwnership(const std::string& service_name,
 383                            OnOwnershipCallback on_ownership_callback) {
 384   AssertOnOriginThread();
 385
 386   PostTaskToDBusThread(FROM_HERE, base::Bind(
 387       &Bus::RequestOwnershipInternal,
 388       this, service_name, on_ownership_callback));
 389 }
 390
 391 void Bus::RequestOwnershipInternal(const std::string& service_name,
 392                                    OnOwnershipCallback on_ownership_callback) {
 393   AssertOnDBusThread();
 394
 395   bool success = Connect();
 396   if (success)
 397     success = RequestOwnershipAndBlock(service_name);
 398
 399   PostTaskToOriginThread(FROM_HERE,
 400                          base::Bind(&Bus::OnOwnership,
 401                                     this,
 402                                     on_ownership_callback,
 403                                     service_name,
 404                                     success));
 405 }
 406
 407 void Bus::OnOwnership(OnOwnershipCallback on_ownership_callback,
 408                       const std::string& service_name,
 409                       bool success) {
 410   AssertOnOriginThread();
 411
 412   on_ownership_callback.Run(service_name, success);
 413 }
 414
 415 bool Bus::RequestOwnershipAndBlock(const std::string& service_name) {
 416   DCHECK(connection_);
 417   // dbus_bus_request_name() is a blocking call.
 418   AssertOnDBusThread();
 419
 420   // Check if we already own the service name.
 421   if (owned_service_names_.find(service_name) != owned_service_names_.end()) {
 422     return true;
 423   }
 424
 425   ScopedDBusError error;
 426   const int result = dbus_bus_request_name(connection_,
 427                                            service_name.c_str(),
 428                                            DBUS_NAME_FLAG_DO_NOT_QUEUE,
 429                                            error.get());
 430   if (result != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER) {
 431     LOG(ERROR) << "Failed to get the ownership of " << service_name << ": "
 432                << (dbus_error_is_set(error.get()) ? error.message() : "");
 433     return false;
 434   }
 435   owned_service_names_.insert(service_name);
 436   return true;
 437 }
 438
 439 bool Bus::ReleaseOwnership(const std::string& service_name) {
 440   DCHECK(connection_);
 441   // dbus_bus_request_name() is a blocking call.
 442   AssertOnDBusThread();
 443
 444   // Check if we already own the service name.
 445   std::set<std::string>::iterator found =
 446       owned_service_names_.find(service_name);
 447   if (found == owned_service_names_.end()) {
 448     LOG(ERROR) << service_name << " is not owned by the bus";
 449     return false;
 450   }
 451
 452   ScopedDBusError error;
 453   const int result = dbus_bus_release_name(connection_, service_name.c_str(),
 454                                            error.get());
 455   if (result == DBUS_RELEASE_NAME_REPLY_RELEASED) {
 456     owned_service_names_.erase(found);
 457     return true;
 458   } else {
 459     LOG(ERROR) << "Failed to release the ownership of " << service_name << ": "
 460                << (error.is_set() ? error.message() : "");
 461     return false;
 462   }
 463 }
 464
 465 bool Bus::SetUpAsyncOperations() {
 466   DCHECK(connection_);
 467   AssertOnDBusThread();
 468
 469   if (async_operations_set_up_)
 470     return true;
 471
 472   // Process all the incoming data if any, so that OnDispatchStatus() will
 473   // be called when the incoming data is ready.
 474   ProcessAllIncomingDataIfAny();
 475
 476   bool success = dbus_connection_set_watch_functions(connection_,
 477                                                      &Bus::OnAddWatchThunk,
 478                                                      &Bus::OnRemoveWatchThunk,
 479                                                      &Bus::OnToggleWatchThunk,
 480                                                      this,
 481                                                      NULL);
 482   CHECK(success) << "Unable to allocate memory";
 483
 484   success = dbus_connection_set_timeout_functions(connection_,
 485                                                   &Bus::OnAddTimeoutThunk,
 486                                                   &Bus::OnRemoveTimeoutThunk,
 487                                                   &Bus::OnToggleTimeoutThunk,
 488                                                   this,
 489                                                   NULL);
 490   CHECK(success) << "Unable to allocate memory";
 491
 492   dbus_connection_set_dispatch_status_function(
 493       connection_,
 494       &Bus::OnDispatchStatusChangedThunk,
 495       this,
 496       NULL);
 497
 498   async_operations_set_up_ = true;
 499
 500   return true;
 501 }
 502
 503 DBusMessage* Bus::SendWithReplyAndBlock(DBusMessage* request,
 504                                         int timeout_ms,
 505                                         DBusError* error) {
 506   DCHECK(connection_);
 507   AssertOnDBusThread();
 508
 509   return dbus_connection_send_with_reply_and_block(
 510       connection_, request, timeout_ms, error);
 511 }
 512
 513 void Bus::SendWithReply(DBusMessage* request,
 514                         DBusPendingCall** pending_call,
 515                         int timeout_ms) {
 516   DCHECK(connection_);
 517   AssertOnDBusThread();
 518
 519   const bool success = dbus_connection_send_with_reply(
 520       connection_, request, pending_call, timeout_ms);
 521   CHECK(success) << "Unable to allocate memory";
 522 }
 523
 524 void Bus::Send(DBusMessage* request, uint32* serial) {
 525   DCHECK(connection_);
 526   AssertOnDBusThread();
 527
 528   const bool success = dbus_connection_send(connection_, request, serial);
 529   CHECK(success) << "Unable to allocate memory";
 530 }
 531
 532 bool Bus::AddFilterFunction(DBusHandleMessageFunction filter_function,
 533                             void* user_data) {
 534   DCHECK(connection_);
 535   AssertOnDBusThread();
 536
 537   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
 538       std::make_pair(filter_function, user_data);
 539   if (filter_functions_added_.find(filter_data_pair) !=
 540       filter_functions_added_.end()) {
 541     VLOG(1) << "Filter function already exists: " << filter_function
 542             << " with associated data: " << user_data;
 543     return false;
 544   }
 545
 546   const bool success = dbus_connection_add_filter(
 547       connection_, filter_function, user_data, NULL);
 548   CHECK(success) << "Unable to allocate memory";
 549   filter_functions_added_.insert(filter_data_pair);
 550   return true;
 551 }
 552
 553 bool Bus::RemoveFilterFunction(DBusHandleMessageFunction filter_function,
 554                                void* user_data) {
 555   DCHECK(connection_);
 556   AssertOnDBusThread();
 557
 558   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
 559       std::make_pair(filter_function, user_data);
 560   if (filter_functions_added_.find(filter_data_pair) ==
 561       filter_functions_added_.end()) {
 562     VLOG(1) << "Requested to remove an unknown filter function: "
 563             << filter_function
 564             << " with associated data: " << user_data;
 565     return false;
 566   }
 567
 568   dbus_connection_remove_filter(connection_, filter_function, user_data);
 569   filter_functions_added_.erase(filter_data_pair);
 570   return true;
 571 }
 572
 573 void Bus::AddMatch(const std::string& match_rule, DBusError* error) {
 574   DCHECK(connection_);
 575   AssertOnDBusThread();
 576
 577   if (match_rules_added_.find(match_rule) != match_rules_added_.end()) {
 578     VLOG(1) << "Match rule already exists: " << match_rule;
 579     return;
 580   }
 581
 582   dbus_bus_add_match(connection_, match_rule.c_str(), error);
 583   match_rules_added_.insert(match_rule);
 584 }
 585
 586 void Bus::RemoveMatch(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     LOG(ERROR) << "Requested to remove an unknown match rule: " << match_rule;
 592     return;
 593   }
 594
 595   dbus_bus_remove_match(connection_, match_rule.c_str(), error);
 596   match_rules_added_.erase(match_rule);
 597 }
 598
 599 bool Bus::TryRegisterObjectPath(const ObjectPath& object_path,
 600                                 const DBusObjectPathVTable* vtable,
 601                                 void* user_data,
 602                                 DBusError* error) {
 603   DCHECK(connection_);
 604   AssertOnDBusThread();
 605
 606   if (registered_object_paths_.find(object_path) !=
 607       registered_object_paths_.end()) {
 608     LOG(ERROR) << "Object path already registered: " << object_path.value();
 609     return false;
 610   }
 611
 612   const bool success = dbus_connection_try_register_object_path(
 613       connection_,
 614       object_path.value().c_str(),
 615       vtable,
 616       user_data,
 617       error);
 618   if (success)
 619     registered_object_paths_.insert(object_path);
 620   return success;
 621 }
 622
 623 void Bus::UnregisterObjectPath(const ObjectPath& object_path) {
 624   DCHECK(connection_);
 625   AssertOnDBusThread();
 626
 627   if (registered_object_paths_.find(object_path) ==
 628       registered_object_paths_.end()) {
 629     LOG(ERROR) << "Requested to unregister an unknown object path: "
 630                << object_path.value();
 631     return;
 632   }
 633
 634   const bool success = dbus_connection_unregister_object_path(
 635       connection_,
 636       object_path.value().c_str());
 637   CHECK(success) << "Unable to allocate memory";
 638   registered_object_paths_.erase(object_path);
 639 }
 640
 641 void Bus::ShutdownOnDBusThreadAndBlockInternal() {
 642   AssertOnDBusThread();
 643
 644   ShutdownAndBlock();
 645   on_shutdown_.Signal();
 646 }
 647
 648 void Bus::ProcessAllIncomingDataIfAny() {
 649   AssertOnDBusThread();
 650
 651   // As mentioned at the class comment in .h file, connection_ can be NULL.
 652   if (!connection_ || !dbus_connection_get_is_connected(connection_))
 653     return;
 654
 655   if (dbus_connection_get_dispatch_status(connection_) ==
 656       DBUS_DISPATCH_DATA_REMAINS) {
 657     while (dbus_connection_dispatch(connection_) ==
 658            DBUS_DISPATCH_DATA_REMAINS);
 659   }
 660 }
 661
 662 void Bus::PostTaskToOriginThread(const tracked_objects::Location& from_here,
 663                                  const base::Closure& task) {
 664   DCHECK(origin_message_loop_proxy_.get());
 665   if (!origin_message_loop_proxy_->PostTask(from_here, task)) {
 666     LOG(WARNING) << "Failed to post a task to the origin message loop";
 667   }
 668 }
 669
 670 void Bus::PostTaskToDBusThread(const tracked_objects::Location& from_here,
 671                                const base::Closure& task) {
 672   if (dbus_thread_message_loop_proxy_.get()) {
 673     if (!dbus_thread_message_loop_proxy_->PostTask(from_here, task)) {
 674       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
 675     }
 676   } else {
 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
 684 void Bus::PostDelayedTaskToDBusThread(
 685     const tracked_objects::Location& from_here,
 686     const base::Closure& task,
 687     base::TimeDelta delay) {
 688   if (dbus_thread_message_loop_proxy_.get()) {
 689     if (!dbus_thread_message_loop_proxy_->PostDelayedTask(
 690             from_here, task, delay)) {
 691       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
 692     }
 693   } else {
 694     DCHECK(origin_message_loop_proxy_.get());
 695     if (!origin_message_loop_proxy_->PostDelayedTask(
 696             from_here, task, delay)) {
 697       LOG(WARNING) << "Failed to post a task to the origin message loop";
 698     }
 699   }
 700 }
 701
 702 bool Bus::HasDBusThread() {
 703   return dbus_thread_message_loop_proxy_.get() != NULL;
 704 }
 705
 706 void Bus::AssertOnOriginThread() {
 707   DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
 708 }
 709
 710 void Bus::AssertOnDBusThread() {
 711   base::ThreadRestrictions::AssertIOAllowed();
 712
 713   if (dbus_thread_message_loop_proxy_.get()) {
 714     DCHECK(dbus_thread_message_loop_proxy_->BelongsToCurrentThread());
 715   } else {
 716     AssertOnOriginThread();
 717   }
 718 }
 719
 720 dbus_bool_t Bus::OnAddWatch(DBusWatch* raw_watch) {
 721   AssertOnDBusThread();
 722
 723   // watch will be deleted when raw_watch is removed in OnRemoveWatch().
 724   Watch* watch = new Watch(raw_watch);
 725   if (watch->IsReadyToBeWatched()) {
 726     watch->StartWatching();
 727   }
 728   ++num_pending_watches_;
 729   return true;
 730 }
 731
 732 void Bus::OnRemoveWatch(DBusWatch* raw_watch) {
 733   AssertOnDBusThread();
 734
 735   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
 736   delete watch;
 737   --num_pending_watches_;
 738 }
 739
 740 void Bus::OnToggleWatch(DBusWatch* raw_watch) {
 741   AssertOnDBusThread();
 742
 743   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
 744   if (watch->IsReadyToBeWatched()) {
 745     watch->StartWatching();
 746   } else {
 747     // It's safe to call this if StartWatching() wasn't called, per
 748     // message_pump_libevent.h.
 749     watch->StopWatching();
 750   }
 751 }
 752
 753 dbus_bool_t Bus::OnAddTimeout(DBusTimeout* raw_timeout) {
 754   AssertOnDBusThread();
 755
 756   // timeout will be deleted when raw_timeout is removed in
 757   // OnRemoveTimeoutThunk().
 758   Timeout* timeout = new Timeout(raw_timeout);
 759   if (timeout->IsReadyToBeMonitored()) {
 760     timeout->StartMonitoring(this);
 761   }
 762   ++num_pending_timeouts_;
 763   return true;
 764 }
 765
 766 void Bus::OnRemoveTimeout(DBusTimeout* raw_timeout) {
 767   AssertOnDBusThread();
 768
 769   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
 770   timeout->Complete();
 771   --num_pending_timeouts_;
 772 }
 773
 774 void Bus::OnToggleTimeout(DBusTimeout* raw_timeout) {
 775   AssertOnDBusThread();
 776
 777   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
 778   if (timeout->IsReadyToBeMonitored()) {
 779     timeout->StartMonitoring(this);
 780   } else {
 781     timeout->StopMonitoring();
 782   }
 783 }
 784
 785 void Bus::OnDispatchStatusChanged(DBusConnection* connection,
 786                                   DBusDispatchStatus status) {
 787   DCHECK_EQ(connection, connection_);
 788   AssertOnDBusThread();
 789
 790   if (!dbus_connection_get_is_connected(connection))
 791     return;
 792
 793   // We cannot call ProcessAllIncomingDataIfAny() here, as calling
 794   // dbus_connection_dispatch() inside DBusDispatchStatusFunction is
 795   // prohibited by the D-Bus library. Hence, we post a task here instead.
 796   // See comments for dbus_connection_set_dispatch_status_function().
 797   PostTaskToDBusThread(FROM_HERE,
 798                        base::Bind(&Bus::ProcessAllIncomingDataIfAny,
 799                                   this));
 800 }
 801
 802 dbus_bool_t Bus::OnAddWatchThunk(DBusWatch* raw_watch, void* data) {
 803   Bus* self = static_cast<Bus*>(data);
 804   return self->OnAddWatch(raw_watch);
 805 }
 806
 807 void Bus::OnRemoveWatchThunk(DBusWatch* raw_watch, void* data) {
 808   Bus* self = static_cast<Bus*>(data);
 809   self->OnRemoveWatch(raw_watch);
 810 }
 811
 812 void Bus::OnToggleWatchThunk(DBusWatch* raw_watch, void* data) {
 813   Bus* self = static_cast<Bus*>(data);
 814   self->OnToggleWatch(raw_watch);
 815 }
 816
 817 dbus_bool_t Bus::OnAddTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 818   Bus* self = static_cast<Bus*>(data);
 819   return self->OnAddTimeout(raw_timeout);
 820 }
 821
 822 void Bus::OnRemoveTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 823   Bus* self = static_cast<Bus*>(data);
 824   self->OnRemoveTimeout(raw_timeout);
 825 }
 826
 827 void Bus::OnToggleTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 828   Bus* self = static_cast<Bus*>(data);
 829   self->OnToggleTimeout(raw_timeout);
 830 }
 831
 832 void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
 833                                        DBusDispatchStatus status,
 834                                        void* data) {
 835   Bus* self = static_cast<Bus*>(data);
 836   self->OnDispatchStatusChanged(connection, status);
 837 }
 838
 839 }  // namespace dbus