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