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   ~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) {
  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) {
  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                                      GetIntervalInMs());
 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 in milliseconds.
 132   int GetIntervalInMs() {
 133     return dbus_timeout_get_interval(raw_timeout_);
 134   }
 135
 136   // Cleans up the raw_timeout and marks that timeout is completed.
 137   // See the class comment above for why we are doing this.
 138   void Complete() {
 139     dbus_timeout_set_data(raw_timeout_, NULL, NULL);
 140     is_completed = true;
 141     Release();
 142   }
 143
 144  private:
 145   friend class base::RefCountedThreadSafe<Timeout>;
 146   ~Timeout() {
 147   }
 148
 149   // Handles the timeout.
 150   void HandleTimeout() {
 151     // If the timeout is marked completed, we should do nothing. This can
 152     // occur if this function is called after Bus::OnRemoveTimeout().
 153     if (is_completed)
 154       return;
 155     // Skip if monitoring is canceled.
 156     if (!monitoring_is_active_)
 157       return;
 158
 159     const bool success = dbus_timeout_handle(raw_timeout_);
 160     CHECK(success) << "Unable to allocate memory";
 161   }
 162
 163   DBusTimeout* raw_timeout_;
 164   bool monitoring_is_active_;
 165   bool is_completed;
 166 };
 167
 168 }  // namespace
 169
 170 Bus::Options::Options()
 171   : bus_type(SESSION),
 172     connection_type(PRIVATE) {
 173 }
 174
 175 Bus::Options::~Options() {
 176 }
 177
 178 Bus::Bus(const Options& options)
 179     : bus_type_(options.bus_type),
 180       connection_type_(options.connection_type),
 181       dbus_thread_message_loop_proxy_(options.dbus_thread_message_loop_proxy),
 182       on_shutdown_(false /* manual_reset */, false /* initially_signaled */),
 183       connection_(NULL),
 184       origin_thread_id_(base::PlatformThread::CurrentId()),
 185       async_operations_set_up_(false),
 186       shutdown_completed_(false),
 187       num_pending_watches_(0),
 188       num_pending_timeouts_(0) {
 189   // This is safe to call multiple times.
 190   dbus_threads_init_default();
 191   // The origin message loop is unnecessary if the client uses synchronous
 192   // functions only.
 193   if (MessageLoop::current())
 194     origin_message_loop_proxy_ =  MessageLoop::current()->message_loop_proxy();
 195 }
 196
 197 Bus::~Bus() {
 198   DCHECK(!connection_);
 199   DCHECK(owned_service_names_.empty());
 200   DCHECK(match_rules_added_.empty());
 201   DCHECK(filter_functions_added_.empty());
 202   DCHECK(registered_object_paths_.empty());
 203   DCHECK_EQ(0, num_pending_watches_);
 204   // TODO(satorux): This check fails occasionally in browser_tests for tests
 205   // that run very quickly. Perhaps something does not have time to clean up.
 206   // Despite the check failing, the tests seem to run fine. crosbug.com/23416
 207   // DCHECK_EQ(0, num_pending_timeouts_);
 208 }
 209
 210 ObjectProxy* Bus::GetObjectProxy(const std::string& service_name,
 211                                  const ObjectPath& object_path) {
 212   return GetObjectProxyWithOptions(service_name, object_path,
 213                                    ObjectProxy::DEFAULT_OPTIONS);
 214 }
 215
 216 ObjectProxy* Bus::GetObjectProxyWithOptions(const std::string& service_name,
 217                                             const dbus::ObjectPath& object_path,
 218                                             int options) {
 219   AssertOnOriginThread();
 220
 221   // Check if we already have the requested object proxy.
 222   const ObjectProxyTable::key_type key(service_name + object_path.value(),
 223                                        options);
 224   ObjectProxyTable::iterator iter = object_proxy_table_.find(key);
 225   if (iter != object_proxy_table_.end()) {
 226     return iter->second;
 227   }
 228
 229   scoped_refptr<ObjectProxy> object_proxy =
 230       new ObjectProxy(this, service_name, object_path, options);
 231   object_proxy_table_[key] = object_proxy;
 232
 233   return object_proxy.get();
 234 }
 235
 236 ExportedObject* Bus::GetExportedObject(const std::string& service_name,
 237                                        const ObjectPath& object_path) {
 238   AssertOnOriginThread();
 239
 240   // Check if we already have the requested exported object.
 241   const std::string key = service_name + object_path.value();
 242   ExportedObjectTable::iterator iter = exported_object_table_.find(key);
 243   if (iter != exported_object_table_.end()) {
 244     return iter->second;
 245   }
 246
 247   scoped_refptr<ExportedObject> exported_object =
 248       new ExportedObject(this, service_name, object_path);
 249   exported_object_table_[key] = exported_object;
 250
 251   return exported_object.get();
 252 }
 253
 254 bool Bus::Connect() {
 255   // dbus_bus_get_private() and dbus_bus_get() are blocking calls.
 256   AssertOnDBusThread();
 257
 258   // Check if it's already initialized.
 259   if (connection_)
 260     return true;
 261
 262   ScopedDBusError error;
 263   const DBusBusType dbus_bus_type = static_cast<DBusBusType>(bus_type_);
 264   if (connection_type_ == PRIVATE) {
 265     connection_ = dbus_bus_get_private(dbus_bus_type, error.get());
 266   } else {
 267     connection_ = dbus_bus_get(dbus_bus_type, error.get());
 268   }
 269   if (!connection_) {
 270     LOG(ERROR) << "Failed to connect to the bus: "
 271                << (dbus_error_is_set(error.get()) ? error.message() : "");
 272     return false;
 273   }
 274   // We shouldn't exit on the disconnected signal.
 275   dbus_connection_set_exit_on_disconnect(connection_, false);
 276
 277   return true;
 278 }
 279
 280 void Bus::ShutdownAndBlock() {
 281   AssertOnDBusThread();
 282
 283   // Unregister the exported objects.
 284   for (ExportedObjectTable::iterator iter = exported_object_table_.begin();
 285        iter != exported_object_table_.end(); ++iter) {
 286     iter->second->Unregister();
 287   }
 288
 289   // Release all service names.
 290   for (std::set<std::string>::iterator iter = owned_service_names_.begin();
 291        iter != owned_service_names_.end();) {
 292     // This is a bit tricky but we should increment the iter here as
 293     // ReleaseOwnership() may remove |service_name| from the set.
 294     const std::string& service_name = *iter++;
 295     ReleaseOwnership(service_name);
 296   }
 297   if (!owned_service_names_.empty()) {
 298     LOG(ERROR) << "Failed to release all service names. # of services left: "
 299                << owned_service_names_.size();
 300   }
 301
 302   // Detach from the remote objects.
 303   for (ObjectProxyTable::iterator iter = object_proxy_table_.begin();
 304        iter != object_proxy_table_.end(); ++iter) {
 305     iter->second->Detach();
 306   }
 307
 308   // Release object proxies and exported objects here. We should do this
 309   // here rather than in the destructor to avoid memory leaks due to
 310   // cyclic references.
 311   object_proxy_table_.clear();
 312   exported_object_table_.clear();
 313
 314   // Private connection should be closed.
 315   if (connection_) {
 316     if (connection_type_ == PRIVATE)
 317       dbus_connection_close(connection_);
 318     // dbus_connection_close() won't unref.
 319     dbus_connection_unref(connection_);
 320   }
 321
 322   connection_ = NULL;
 323   shutdown_completed_ = true;
 324 }
 325
 326 void Bus::ShutdownOnDBusThreadAndBlock() {
 327   AssertOnOriginThread();
 328   DCHECK(dbus_thread_message_loop_proxy_.get());
 329
 330   PostTaskToDBusThread(FROM_HERE, base::Bind(
 331       &Bus::ShutdownOnDBusThreadAndBlockInternal,
 332       this));
 333
 334   // Wait until the shutdown is complete on the D-Bus thread.
 335   // The shutdown should not hang, but set timeout just in case.
 336   const int kTimeoutSecs = 3;
 337   const base::TimeDelta timeout(base::TimeDelta::FromSeconds(kTimeoutSecs));
 338   const bool signaled = on_shutdown_.TimedWait(timeout);
 339   LOG_IF(ERROR, !signaled) << "Failed to shutdown the bus";
 340 }
 341
 342 bool Bus::RequestOwnership(const std::string& service_name) {
 343   DCHECK(connection_);
 344   // dbus_bus_request_name() is a blocking call.
 345   AssertOnDBusThread();
 346
 347   // Check if we already own the service name.
 348   if (owned_service_names_.find(service_name) != owned_service_names_.end()) {
 349     return true;
 350   }
 351
 352   ScopedDBusError error;
 353   const int result = dbus_bus_request_name(connection_,
 354                                            service_name.c_str(),
 355                                            DBUS_NAME_FLAG_DO_NOT_QUEUE,
 356                                            error.get());
 357   if (result != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER) {
 358     LOG(ERROR) << "Failed to get the ownership of " << service_name << ": "
 359                << (dbus_error_is_set(error.get()) ? error.message() : "");
 360     return false;
 361   }
 362   owned_service_names_.insert(service_name);
 363   return true;
 364 }
 365
 366 bool Bus::ReleaseOwnership(const std::string& service_name) {
 367   DCHECK(connection_);
 368   // dbus_bus_request_name() is a blocking call.
 369   AssertOnDBusThread();
 370
 371   // Check if we already own the service name.
 372   std::set<std::string>::iterator found =
 373       owned_service_names_.find(service_name);
 374   if (found == owned_service_names_.end()) {
 375     LOG(ERROR) << service_name << " is not owned by the bus";
 376     return false;
 377   }
 378
 379   ScopedDBusError error;
 380   const int result = dbus_bus_release_name(connection_, service_name.c_str(),
 381                                            error.get());
 382   if (result == DBUS_RELEASE_NAME_REPLY_RELEASED) {
 383     owned_service_names_.erase(found);
 384     return true;
 385   } else {
 386     LOG(ERROR) << "Failed to release the ownership of " << service_name << ": "
 387                << (error.is_set() ? error.message() : "");
 388     return false;
 389   }
 390 }
 391
 392 bool Bus::SetUpAsyncOperations() {
 393   DCHECK(connection_);
 394   AssertOnDBusThread();
 395
 396   if (async_operations_set_up_)
 397     return true;
 398
 399   // Process all the incoming data if any, so that OnDispatchStatus() will
 400   // be called when the incoming data is ready.
 401   ProcessAllIncomingDataIfAny();
 402
 403   bool success = dbus_connection_set_watch_functions(connection_,
 404                                                      &Bus::OnAddWatchThunk,
 405                                                      &Bus::OnRemoveWatchThunk,
 406                                                      &Bus::OnToggleWatchThunk,
 407                                                      this,
 408                                                      NULL);
 409   CHECK(success) << "Unable to allocate memory";
 410
 411   success = dbus_connection_set_timeout_functions(connection_,
 412                                                   &Bus::OnAddTimeoutThunk,
 413                                                   &Bus::OnRemoveTimeoutThunk,
 414                                                   &Bus::OnToggleTimeoutThunk,
 415                                                   this,
 416                                                   NULL);
 417   CHECK(success) << "Unable to allocate memory";
 418
 419   dbus_connection_set_dispatch_status_function(
 420       connection_,
 421       &Bus::OnDispatchStatusChangedThunk,
 422       this,
 423       NULL);
 424
 425   async_operations_set_up_ = true;
 426
 427   return true;
 428 }
 429
 430 DBusMessage* Bus::SendWithReplyAndBlock(DBusMessage* request,
 431                                         int timeout_ms,
 432                                         DBusError* error) {
 433   DCHECK(connection_);
 434   AssertOnDBusThread();
 435
 436   return dbus_connection_send_with_reply_and_block(
 437       connection_, request, timeout_ms, error);
 438 }
 439
 440 void Bus::SendWithReply(DBusMessage* request,
 441                         DBusPendingCall** pending_call,
 442                         int timeout_ms) {
 443   DCHECK(connection_);
 444   AssertOnDBusThread();
 445
 446   const bool success = dbus_connection_send_with_reply(
 447       connection_, request, pending_call, timeout_ms);
 448   CHECK(success) << "Unable to allocate memory";
 449 }
 450
 451 void Bus::Send(DBusMessage* request, uint32* serial) {
 452   DCHECK(connection_);
 453   AssertOnDBusThread();
 454
 455   const bool success = dbus_connection_send(connection_, request, serial);
 456   CHECK(success) << "Unable to allocate memory";
 457 }
 458
 459 bool Bus::AddFilterFunction(DBusHandleMessageFunction filter_function,
 460                             void* user_data) {
 461   DCHECK(connection_);
 462   AssertOnDBusThread();
 463
 464   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
 465       std::make_pair(filter_function, user_data);
 466   if (filter_functions_added_.find(filter_data_pair) !=
 467       filter_functions_added_.end()) {
 468     VLOG(1) << "Filter function already exists: " << filter_function
 469             << " with associated data: " << user_data;
 470     return false;
 471   }
 472
 473   const bool success = dbus_connection_add_filter(
 474       connection_, filter_function, user_data, NULL);
 475   CHECK(success) << "Unable to allocate memory";
 476   filter_functions_added_.insert(filter_data_pair);
 477   return true;
 478 }
 479
 480 bool Bus::RemoveFilterFunction(DBusHandleMessageFunction filter_function,
 481                                void* user_data) {
 482   DCHECK(connection_);
 483   AssertOnDBusThread();
 484
 485   std::pair<DBusHandleMessageFunction, void*> filter_data_pair =
 486       std::make_pair(filter_function, user_data);
 487   if (filter_functions_added_.find(filter_data_pair) ==
 488       filter_functions_added_.end()) {
 489     VLOG(1) << "Requested to remove an unknown filter function: "
 490             << filter_function
 491             << " with associated data: " << user_data;
 492     return false;
 493   }
 494
 495   dbus_connection_remove_filter(connection_, filter_function, user_data);
 496   filter_functions_added_.erase(filter_data_pair);
 497   return true;
 498 }
 499
 500 void Bus::AddMatch(const std::string& match_rule, DBusError* error) {
 501   DCHECK(connection_);
 502   AssertOnDBusThread();
 503
 504   if (match_rules_added_.find(match_rule) != match_rules_added_.end()) {
 505     VLOG(1) << "Match rule already exists: " << match_rule;
 506     return;
 507   }
 508
 509   dbus_bus_add_match(connection_, match_rule.c_str(), error);
 510   match_rules_added_.insert(match_rule);
 511 }
 512
 513 void Bus::RemoveMatch(const std::string& match_rule, DBusError* error) {
 514   DCHECK(connection_);
 515   AssertOnDBusThread();
 516
 517   if (match_rules_added_.find(match_rule) == match_rules_added_.end()) {
 518     LOG(ERROR) << "Requested to remove an unknown match rule: " << match_rule;
 519     return;
 520   }
 521
 522   dbus_bus_remove_match(connection_, match_rule.c_str(), error);
 523   match_rules_added_.erase(match_rule);
 524 }
 525
 526 bool Bus::TryRegisterObjectPath(const ObjectPath& object_path,
 527                                 const DBusObjectPathVTable* vtable,
 528                                 void* user_data,
 529                                 DBusError* error) {
 530   DCHECK(connection_);
 531   AssertOnDBusThread();
 532
 533   if (registered_object_paths_.find(object_path) !=
 534       registered_object_paths_.end()) {
 535     LOG(ERROR) << "Object path already registered: " << object_path.value();
 536     return false;
 537   }
 538
 539   const bool success = dbus_connection_try_register_object_path(
 540       connection_,
 541       object_path.value().c_str(),
 542       vtable,
 543       user_data,
 544       error);
 545   if (success)
 546     registered_object_paths_.insert(object_path);
 547   return success;
 548 }
 549
 550 void Bus::UnregisterObjectPath(const ObjectPath& object_path) {
 551   DCHECK(connection_);
 552   AssertOnDBusThread();
 553
 554   if (registered_object_paths_.find(object_path) ==
 555       registered_object_paths_.end()) {
 556     LOG(ERROR) << "Requested to unregister an unknown object path: "
 557                << object_path.value();
 558     return;
 559   }
 560
 561   const bool success = dbus_connection_unregister_object_path(
 562       connection_,
 563       object_path.value().c_str());
 564   CHECK(success) << "Unable to allocate memory";
 565   registered_object_paths_.erase(object_path);
 566 }
 567
 568 void Bus::ShutdownOnDBusThreadAndBlockInternal() {
 569   AssertOnDBusThread();
 570
 571   ShutdownAndBlock();
 572   on_shutdown_.Signal();
 573 }
 574
 575 void Bus::ProcessAllIncomingDataIfAny() {
 576   AssertOnDBusThread();
 577
 578   // As mentioned at the class comment in .h file, connection_ can be NULL.
 579   if (!connection_ || !dbus_connection_get_is_connected(connection_))
 580     return;
 581
 582   if (dbus_connection_get_dispatch_status(connection_) ==
 583       DBUS_DISPATCH_DATA_REMAINS) {
 584     while (dbus_connection_dispatch(connection_) ==
 585            DBUS_DISPATCH_DATA_REMAINS);
 586   }
 587 }
 588
 589 void Bus::PostTaskToOriginThread(const tracked_objects::Location& from_here,
 590                                  const base::Closure& task) {
 591   DCHECK(origin_message_loop_proxy_.get());
 592   if (!origin_message_loop_proxy_->PostTask(from_here, task)) {
 593     LOG(WARNING) << "Failed to post a task to the origin message loop";
 594   }
 595 }
 596
 597 void Bus::PostTaskToDBusThread(const tracked_objects::Location& from_here,
 598                                const base::Closure& task) {
 599   if (dbus_thread_message_loop_proxy_.get()) {
 600     if (!dbus_thread_message_loop_proxy_->PostTask(from_here, task)) {
 601       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
 602     }
 603   } else {
 604     DCHECK(origin_message_loop_proxy_.get());
 605     if (!origin_message_loop_proxy_->PostTask(from_here, task)) {
 606       LOG(WARNING) << "Failed to post a task to the origin message loop";
 607     }
 608   }
 609 }
 610
 611 void Bus::PostDelayedTaskToDBusThread(
 612     const tracked_objects::Location& from_here,
 613     const base::Closure& task,
 614     int delay_ms) {
 615   if (dbus_thread_message_loop_proxy_.get()) {
 616     if (!dbus_thread_message_loop_proxy_->PostDelayedTask(
 617             from_here, task, delay_ms)) {
 618       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
 619     }
 620   } else {
 621     DCHECK(origin_message_loop_proxy_.get());
 622     if (!origin_message_loop_proxy_->PostDelayedTask(
 623             from_here, task, delay_ms)) {
 624       LOG(WARNING) << "Failed to post a task to the origin message loop";
 625     }
 626   }
 627 }
 628
 629 bool Bus::HasDBusThread() {
 630   return dbus_thread_message_loop_proxy_.get() != NULL;
 631 }
 632
 633 void Bus::AssertOnOriginThread() {
 634   DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
 635 }
 636
 637 void Bus::AssertOnDBusThread() {
 638   base::ThreadRestrictions::AssertIOAllowed();
 639
 640   if (dbus_thread_message_loop_proxy_.get()) {
 641     DCHECK(dbus_thread_message_loop_proxy_->BelongsToCurrentThread());
 642   } else {
 643     AssertOnOriginThread();
 644   }
 645 }
 646
 647 dbus_bool_t Bus::OnAddWatch(DBusWatch* raw_watch) {
 648   AssertOnDBusThread();
 649
 650   // watch will be deleted when raw_watch is removed in OnRemoveWatch().
 651   Watch* watch = new Watch(raw_watch);
 652   if (watch->IsReadyToBeWatched()) {
 653     watch->StartWatching();
 654   }
 655   ++num_pending_watches_;
 656   return true;
 657 }
 658
 659 void Bus::OnRemoveWatch(DBusWatch* raw_watch) {
 660   AssertOnDBusThread();
 661
 662   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
 663   delete watch;
 664   --num_pending_watches_;
 665 }
 666
 667 void Bus::OnToggleWatch(DBusWatch* raw_watch) {
 668   AssertOnDBusThread();
 669
 670   Watch* watch = static_cast<Watch*>(dbus_watch_get_data(raw_watch));
 671   if (watch->IsReadyToBeWatched()) {
 672     watch->StartWatching();
 673   } else {
 674     // It's safe to call this if StartWatching() wasn't called, per
 675     // message_pump_libevent.h.
 676     watch->StopWatching();
 677   }
 678 }
 679
 680 dbus_bool_t Bus::OnAddTimeout(DBusTimeout* raw_timeout) {
 681   AssertOnDBusThread();
 682
 683   // timeout will be deleted when raw_timeout is removed in
 684   // OnRemoveTimeoutThunk().
 685   Timeout* timeout = new Timeout(raw_timeout);
 686   if (timeout->IsReadyToBeMonitored()) {
 687     timeout->StartMonitoring(this);
 688   }
 689   ++num_pending_timeouts_;
 690   return true;
 691 }
 692
 693 void Bus::OnRemoveTimeout(DBusTimeout* raw_timeout) {
 694   AssertOnDBusThread();
 695
 696   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
 697   timeout->Complete();
 698   --num_pending_timeouts_;
 699 }
 700
 701 void Bus::OnToggleTimeout(DBusTimeout* raw_timeout) {
 702   AssertOnDBusThread();
 703
 704   Timeout* timeout = static_cast<Timeout*>(dbus_timeout_get_data(raw_timeout));
 705   if (timeout->IsReadyToBeMonitored()) {
 706     timeout->StartMonitoring(this);
 707   } else {
 708     timeout->StopMonitoring();
 709   }
 710 }
 711
 712 void Bus::OnDispatchStatusChanged(DBusConnection* connection,
 713                                   DBusDispatchStatus status) {
 714   DCHECK_EQ(connection, connection_);
 715   AssertOnDBusThread();
 716
 717   if (!dbus_connection_get_is_connected(connection))
 718     return;
 719
 720   // We cannot call ProcessAllIncomingDataIfAny() here, as calling
 721   // dbus_connection_dispatch() inside DBusDispatchStatusFunction is
 722   // prohibited by the D-Bus library. Hence, we post a task here instead.
 723   // See comments for dbus_connection_set_dispatch_status_function().
 724   PostTaskToDBusThread(FROM_HERE,
 725                        base::Bind(&Bus::ProcessAllIncomingDataIfAny,
 726                                   this));
 727 }
 728
 729 dbus_bool_t Bus::OnAddWatchThunk(DBusWatch* raw_watch, void* data) {
 730   Bus* self = static_cast<Bus*>(data);
 731   return self->OnAddWatch(raw_watch);
 732 }
 733
 734 void Bus::OnRemoveWatchThunk(DBusWatch* raw_watch, void* data) {
 735   Bus* self = static_cast<Bus*>(data);
 736   return self->OnRemoveWatch(raw_watch);
 737 }
 738
 739 void Bus::OnToggleWatchThunk(DBusWatch* raw_watch, void* data) {
 740   Bus* self = static_cast<Bus*>(data);
 741   return self->OnToggleWatch(raw_watch);
 742 }
 743
 744 dbus_bool_t Bus::OnAddTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 745   Bus* self = static_cast<Bus*>(data);
 746   return self->OnAddTimeout(raw_timeout);
 747 }
 748
 749 void Bus::OnRemoveTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 750   Bus* self = static_cast<Bus*>(data);
 751   return self->OnRemoveTimeout(raw_timeout);
 752 }
 753
 754 void Bus::OnToggleTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
 755   Bus* self = static_cast<Bus*>(data);
 756   return self->OnToggleTimeout(raw_timeout);
 757 }
 758
 759 void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
 760                                        DBusDispatchStatus status,
 761                                        void* data) {
 762   Bus* self = static_cast<Bus*>(data);
 763   return self->OnDispatchStatusChanged(connection, status);
 764 }
 765
 766 }  // namespace dbus