DBus: Use TaskRunners instead of MessageLoopProxies.

dbus/bus.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "dbus/bus.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/message_loop_proxy.h"
 #include "base/stl_util.h"
 #include "base/threading/thread.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/time.h"
 #include "dbus/exported_object.h"
-#include "dbus/object_path.h"
 #include "dbus/object_proxy.h"
 #include "dbus/scoped_dbus_error.h"
 
 namespace dbus {
 
 namespace {
 
 const char kDisconnectedSignal[] = "Disconnected";
 const char kDisconnectedMatchRule[] =
     "type='signal', path='/org/freedesktop/DBus/Local',"
     "interface='org.freedesktop.DBus.Local', member='Disconnected'";
 
 // The class is used for watching the file descriptor used for D-Bus
 // communication.
 class Watch : public base::MessagePumpLibevent::Watcher {
  public:
-  Watch(DBusWatch* watch)
+  explicit Watch(DBusWatch* watch)
       : raw_watch_(watch) {
     dbus_watch_set_data(raw_watch_, this, NULL);
   }
 
   virtual ~Watch() {
     dbus_watch_set_data(raw_watch_, NULL, NULL);
   }
 
   // Returns true if the underlying file descriptor is ready to be watched.
   bool IsReadyToBeWatched() {
@@ skipping 50 matching lines @@
 // The class is used for monitoring the timeout used for D-Bus method
 // calls.
 //
 // Unlike Watch, Timeout is a ref counted object, to ensure that |this| of
 // the object is is alive when HandleTimeout() is called. It's unlikely
 // but it may be possible that HandleTimeout() is called after
 // Bus::OnRemoveTimeout(). That's why we don't simply delete the object in
 // Bus::OnRemoveTimeout().
 class Timeout : public base::RefCountedThreadSafe<Timeout> {
  public:
-  Timeout(DBusTimeout* timeout)
+  explicit Timeout(DBusTimeout* timeout)
       : raw_timeout_(timeout),
         monitoring_is_active_(false),
         is_completed(false) {
     dbus_timeout_set_data(raw_timeout_, this, NULL);
     AddRef();  // Balanced on Complete().
   }
 
   // Returns true if the timeout is ready to be monitored.
   bool IsReadyToBeMonitored() {
     return dbus_timeout_get_enabled(raw_timeout_);
@@ skipping 59 matching lines @@
   : bus_type(SESSION),
     connection_type(PRIVATE) {
 }
 
 Bus::Options::~Options() {
 }
 
 Bus::Bus(const Options& options)
     : bus_type_(options.bus_type),
       connection_type_(options.connection_type),
-      dbus_thread_message_loop_proxy_(options.dbus_thread_message_loop_proxy),
+      dbus_task_runner_(options.dbus_task_runner),
       on_shutdown_(false /* manual_reset */, false /* initially_signaled */),
       connection_(NULL),
       origin_thread_id_(base::PlatformThread::CurrentId()),
       async_operations_set_up_(false),
       shutdown_completed_(false),
       num_pending_watches_(0),
       num_pending_timeouts_(0),
       address_(options.address),
       on_disconnected_closure_(options.disconnected_callback) {
   // This is safe to call multiple times.
   dbus_threads_init_default();
   // The origin message loop is unnecessary if the client uses synchronous
   // functions only.
   if (MessageLoop::current())
-    origin_message_loop_proxy_ =  MessageLoop::current()->message_loop_proxy();
+    origin_task_runner_ =  MessageLoop::current()->message_loop_proxy();
 }
 
 Bus::~Bus() {
   DCHECK(!connection_);
   DCHECK(owned_service_names_.empty());
   DCHECK(match_rules_added_.empty());
   DCHECK(filter_functions_added_.empty());
   DCHECK(registered_object_paths_.empty());
   DCHECK_EQ(0, num_pending_watches_);
   // TODO(satorux): This check fails occasionally in browser_tests for tests
@@ skipping 91 matching lines @@
   // GetExportedObject() call to return a new object, rather than this one.
   ExportedObjectTable::iterator iter = exported_object_table_.find(object_path);
   if (iter == exported_object_table_.end())
     return;
 
   scoped_refptr<ExportedObject> exported_object = iter->second;
   exported_object_table_.erase(iter);
 
   // Post the task to perform the final unregistration to the D-Bus thread.
   // Since the registration also happens on the D-Bus thread in
-  // TryRegisterObjectPath(), and the message loop proxy we post to is a
-  // MessageLoopProxy which inherits from SequencedTaskRunner, there is a
-  // guarantee that this will happen before any future registration call.
-  PostTaskToDBusThread(FROM_HERE, base::Bind(
-      &Bus::UnregisterExportedObjectInternal,
-      this, exported_object));
+  // TryRegisterObjectPath(), and the task runner we post to is a
+  // SequencedTaskRunner, there is a guarantee that this will happen before any
+  // future registration call.
+  PostTaskToDBusThread(FROM_HERE,
+                       base::Bind(&Bus::UnregisterExportedObjectInternal,
+                                  this, exported_object));
 }
 
 void Bus::UnregisterExportedObjectInternal(
     scoped_refptr<dbus::ExportedObject> exported_object) {
   AssertOnDBusThread();
 
   exported_object->Unregister();
 }
 
 bool Bus::Connect() {
@@ skipping 104 matching lines @@
     // dbus_connection_close() won't unref.
     dbus_connection_unref(connection_);
   }
 
   connection_ = NULL;
   shutdown_completed_ = true;
 }
 
 void Bus::ShutdownOnDBusThreadAndBlock() {
   AssertOnOriginThread();
-  DCHECK(dbus_thread_message_loop_proxy_.get());
+  DCHECK(dbus_task_runner_.get());
 
   PostTaskToDBusThread(FROM_HERE, base::Bind(
       &Bus::ShutdownOnDBusThreadAndBlockInternal,
       this));
 
   // http://crbug.com/125222
   base::ThreadRestrictions::ScopedAllowWait allow_wait;
 
   // Wait until the shutdown is complete on the D-Bus thread.
   // The shutdown should not hang, but set timeout just in case.
@@ skipping 283 matching lines @@
   // (crbug.com/174431)
   if (dbus_connection_get_dispatch_status(connection_) ==
       DBUS_DISPATCH_DATA_REMAINS) {
     while (dbus_connection_dispatch(connection_) ==
            DBUS_DISPATCH_DATA_REMAINS);
   }
 }
 
 void Bus::PostTaskToOriginThread(const tracked_objects::Location& from_here,
                                  const base::Closure& task) {
-  DCHECK(origin_message_loop_proxy_.get());
-  if (!origin_message_loop_proxy_->PostTask(from_here, task)) {
+  DCHECK(origin_task_runner_.get());
+  if (!origin_task_runner_->PostTask(from_here, task)) {
     LOG(WARNING) << "Failed to post a task to the origin message loop";
   }
 }
 
 void Bus::PostTaskToDBusThread(const tracked_objects::Location& from_here,
                                const base::Closure& task) {
-  if (dbus_thread_message_loop_proxy_.get()) {
-    if (!dbus_thread_message_loop_proxy_->PostTask(from_here, task)) {
+  if (dbus_task_runner_.get()) {
+    if (!dbus_task_runner_->PostTask(from_here, task)) {
       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
     }
   } else {
-    DCHECK(origin_message_loop_proxy_.get());
-    if (!origin_message_loop_proxy_->PostTask(from_here, task)) {
+    DCHECK(origin_task_runner_.get());
+    if (!origin_task_runner_->PostTask(from_here, task)) {
       LOG(WARNING) << "Failed to post a task to the origin message loop";
     }
   }
 }
 
 void Bus::PostDelayedTaskToDBusThread(
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     base::TimeDelta delay) {
-  if (dbus_thread_message_loop_proxy_.get()) {
-    if (!dbus_thread_message_loop_proxy_->PostDelayedTask(
+  if (dbus_task_runner_.get()) {
+    if (!dbus_task_runner_->PostDelayedTask(
             from_here, task, delay)) {
       LOG(WARNING) << "Failed to post a task to the D-Bus thread message loop";
     }
   } else {
-    DCHECK(origin_message_loop_proxy_.get());
-    if (!origin_message_loop_proxy_->PostDelayedTask(
-            from_here, task, delay)) {
+    DCHECK(origin_task_runner_.get());
+    if (!origin_task_runner_->PostDelayedTask(from_here, task, delay)) {
       LOG(WARNING) << "Failed to post a task to the origin message loop";
     }
   }
 }
 
 bool Bus::HasDBusThread() {
-  return dbus_thread_message_loop_proxy_.get() != NULL;
+  return dbus_task_runner_.get() != NULL;
 }
 
 void Bus::AssertOnOriginThread() {
   DCHECK_EQ(origin_thread_id_, base::PlatformThread::CurrentId());
 }
 
 void Bus::AssertOnDBusThread() {
   base::ThreadRestrictions::AssertIOAllowed();
 
-  if (dbus_thread_message_loop_proxy_.get()) {
-    DCHECK(dbus_thread_message_loop_proxy_->BelongsToCurrentThread());
+  if (dbus_task_runner_.get()) {
+    DCHECK(dbus_task_runner_->RunsTasksOnCurrentThread());
   } else {
     AssertOnOriginThread();
   }
 }
 
 dbus_bool_t Bus::OnAddWatch(DBusWatch* raw_watch) {
   AssertOnDBusThread();
 
   // watch will be deleted when raw_watch is removed in OnRemoveWatch().
   Watch* watch = new Watch(raw_watch);
@@ skipping 115 matching lines @@
 }
 
 void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
                                        DBusDispatchStatus status,
                                        void* data) {
   Bus* self = static_cast<Bus*>(data);
   self->OnDispatchStatusChanged(connection, status);
 }
 
 DBusHandlerResult Bus::OnConnectionDisconnectedFilter(
-    DBusConnection *connection,
-    DBusMessage *message,
-    void *data) {
+    DBusConnection* connection,
+    DBusMessage* message,
+    void* data) {
   if (dbus_message_is_signal(message,
                              DBUS_INTERFACE_LOCAL,
                              kDisconnectedSignal)) {
     Bus* self = static_cast<Bus*>(data);
     self->AssertOnDBusThread();
     self->OnConnectionDisconnected(connection);
     return DBUS_HANDLER_RESULT_HANDLED;
   }
   return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
 }
 
 }  // namespace dbus