Linux/CrOS: Retry connecting to mtpd.

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/stringprintf.h"
 #include "base/threading/thread.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/time.h"
 #include "dbus/exported_object.h"
+#include "dbus/message.h"
 #include "dbus/object_manager.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 NameOwnerChanged member in org.freedesktop.DBus
+const char kNameOwnerChangedSignal[] = "NameOwnerChanged";
+
+// The match rule used to filter for changes to a given service name owner.
+const char kServiceNameOwnerChangeMatchRule[] =
+    "type='signal',interface='org.freedesktop.DBus',"
+    "member='NameOwnerChanged',path='/org/freedesktop/DBus',"
+    "sender='org.freedesktop.DBus',arg0='%s'";
+
 // The class is used for watching the file descriptor used for D-Bus
 // communication.
 class Watch : public base::MessagePumpLibevent::Watcher {
  public:
   explicit Watch(DBusWatch* watch)
       : raw_watch_(watch) {
     dbus_watch_set_data(raw_watch_, this, NULL);
   }
 
   virtual ~Watch() {
@@ skipping 861 matching lines @@
 void Bus::GetServiceOwnerInternal(const std::string& service_name,
                                   const GetServiceOwnerCallback& callback) {
   AssertOnDBusThread();
 
   std::string service_owner;
   if (Connect())
     service_owner = GetServiceOwnerAndBlock(service_name, SUPPRESS_ERRORS);
   PostTaskToOriginThread(FROM_HERE, base::Bind(callback, service_owner));
 }
 
+void Bus::ListenForServiceOwnerChange(
+    const std::string& service_name,
+    const GetServiceOwnerCallback& callback) {
+  AssertOnOriginThread();
+  DCHECK(!service_name.empty());
+  DCHECK(!callback.is_null());
+
+  PostTaskToDBusThread(FROM_HERE,
+                       base::Bind(&Bus::ListenForServiceOwnerChangeInternal,
+                                  this, service_name, callback));
+}
+
+void Bus::ListenForServiceOwnerChangeInternal(
+    const std::string& service_name,
+    const GetServiceOwnerCallback& callback) {
+  AssertOnDBusThread();
+  DCHECK(!service_name.empty());
+  DCHECK(!callback.is_null());
+
+  if (!Connect() || !SetUpAsyncOperations())
+    return;
+
+  if (service_owner_changed_listener_map_.empty()) {
+    bool filter_added =
+        AddFilterFunction(Bus::OnServiceOwnerChangedFilter, this);
+    DCHECK(filter_added);
+  }
+
+  ServiceOwnerChangedListenerMap::iterator it =
+      service_owner_changed_listener_map_.find(service_name);
+  if (it == service_owner_changed_listener_map_.end()) {
+    // Add a match rule for the new service name.
+    const std::string name_owner_changed_match_rule =
+        base::StringPrintf(kServiceNameOwnerChangeMatchRule,
+                           service_name.c_str());
+    ScopedDBusError error;
+    AddMatch(name_owner_changed_match_rule, error.get());
+    if (error.is_set()) {
+      LOG(ERROR) << "Failed to add match rule for " << service_name
+                 << ". Got " << error.name() << ": " << error.message();
+      return;
+    }
+
+    service_owner_changed_listener_map_[service_name].push_back(callback);
+    return;
+  }
+
+  // Check if the callback has already been added.
+  std::vector<GetServiceOwnerCallback>& callbacks = it->second;
+  for (size_t i = 0; i < callbacks.size(); ++i) {
+    if (callbacks[i].Equals(callback))
+      return;
+  }
+  callbacks.push_back(callback);
+}
+
+void Bus::UnlistenForServiceOwnerChange(
+    const std::string& service_name,
+    const GetServiceOwnerCallback& callback) {
+  AssertOnOriginThread();
+  DCHECK(!service_name.empty());
+  DCHECK(!callback.is_null());
+
+  PostTaskToDBusThread(FROM_HERE,
+                       base::Bind(&Bus::UnlistenForServiceOwnerChangeInternal,
+                                  this, service_name, callback));
+}
+
+void Bus::UnlistenForServiceOwnerChangeInternal(
+    const std::string& service_name,
+    const GetServiceOwnerCallback& callback) {
+  AssertOnDBusThread();
+  DCHECK(!service_name.empty());
+  DCHECK(!callback.is_null());
+
+  ServiceOwnerChangedListenerMap::iterator it =
+      service_owner_changed_listener_map_.find(service_name);
+  if (it == service_owner_changed_listener_map_.end())
+    return;
+
+  std::vector<GetServiceOwnerCallback>& callbacks = it->second;
+  for (size_t i = 0; i < callbacks.size(); ++i) {
+    if (callbacks[i].Equals(callback)) {
+      callbacks.erase(callbacks.begin() + i);
+      break;  // There can be only one.
+    }
+  }
+  if (!callbacks.empty())
+    return;
+
+  // Last callback for |service_name| has been removed, remove match rule.
+  const std::string name_owner_changed_match_rule =
+      base::StringPrintf(kServiceNameOwnerChangeMatchRule,
+                         service_name.c_str());
+  ScopedDBusError error;
+  RemoveMatch(name_owner_changed_match_rule, error.get());
+  // And remove |service_owner_changed_listener_map_| entry.
+  service_owner_changed_listener_map_.erase(it);
+
+  if (service_owner_changed_listener_map_.empty()) {
+    bool filter_removed =
+        RemoveFilterFunction(Bus::OnServiceOwnerChangedFilter, this);
+    DCHECK(filter_removed);
+  }
+}
+
 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);
   if (watch->IsReadyToBeWatched()) {
     watch->StartWatching();
   }
   ++num_pending_watches_;
   return true;
@@ skipping 72 matching lines @@
   if (!on_disconnected_closure_.is_null())
     PostTaskToOriginThread(FROM_HERE, on_disconnected_closure_);
 
   if (!connection)
     return;
   DCHECK(!dbus_connection_get_is_connected(connection));
 
   ShutdownAndBlock();
 }
 
+void Bus::OnServiceOwnerChanged(DBusMessage* message) {
+  DCHECK(message);
+  AssertOnDBusThread();
+
+  // |message| will be unrefed on exit of the function. Increment the
+  // reference so we can use it in Signal::FromRawMessage() below.
+  dbus_message_ref(message);
+  scoped_ptr<Signal> signal(Signal::FromRawMessage(message));
+
+  // Confirm the validity of the NameOwnerChanged signal.
+  if (signal->GetMember() != kNameOwnerChangedSignal ||
+      signal->GetInterface() != DBUS_INTERFACE_DBUS ||
+      signal->GetSender() != DBUS_SERVICE_DBUS) {
+    return;
+  }
+
+  MessageReader reader(signal.get());
+  std::string service_name;
+  std::string old_owner;
+  std::string new_owner;
+  if (!reader.PopString(&service_name) ||
+      !reader.PopString(&old_owner) ||
+      !reader.PopString(&new_owner)) {
+    return;
+  }
+
+  ServiceOwnerChangedListenerMap::const_iterator it =
+      service_owner_changed_listener_map_.find(service_name);
+  if (it == service_owner_changed_listener_map_.end())
+    return;
+
+  const std::vector<GetServiceOwnerCallback>& callbacks = it->second;
+  for (size_t i = 0; i < callbacks.size(); ++i) {
+    PostTaskToOriginThread(FROM_HERE,
+                           base::Bind(callbacks[i], new_owner));
+  }
+}
+
+// static
 dbus_bool_t Bus::OnAddWatchThunk(DBusWatch* raw_watch, void* data) {
   Bus* self = static_cast<Bus*>(data);
   return self->OnAddWatch(raw_watch);
 }
 
+// static
 void Bus::OnRemoveWatchThunk(DBusWatch* raw_watch, void* data) {
   Bus* self = static_cast<Bus*>(data);
   self->OnRemoveWatch(raw_watch);
 }
 
+// static
 void Bus::OnToggleWatchThunk(DBusWatch* raw_watch, void* data) {
   Bus* self = static_cast<Bus*>(data);
   self->OnToggleWatch(raw_watch);
 }
 
+// static
 dbus_bool_t Bus::OnAddTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
   Bus* self = static_cast<Bus*>(data);
   return self->OnAddTimeout(raw_timeout);
 }
 
+// static
 void Bus::OnRemoveTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
   Bus* self = static_cast<Bus*>(data);
   self->OnRemoveTimeout(raw_timeout);
 }
 
+// static
 void Bus::OnToggleTimeoutThunk(DBusTimeout* raw_timeout, void* data) {
   Bus* self = static_cast<Bus*>(data);
   self->OnToggleTimeout(raw_timeout);
 }
 
+// static
 void Bus::OnDispatchStatusChangedThunk(DBusConnection* connection,
                                        DBusDispatchStatus status,
                                        void* data) {
   Bus* self = static_cast<Bus*>(data);
   self->OnDispatchStatusChanged(connection, status);
 }
 
+// static
 DBusHandlerResult Bus::OnConnectionDisconnectedFilter(
     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;
 }
 
+// static
+DBusHandlerResult Bus::OnServiceOwnerChangedFilter(
+    DBusConnection* connection,
+    DBusMessage* message,
+    void* data) {
+  if (dbus_message_is_signal(message,
+                             DBUS_INTERFACE_DBUS,
+                             kNameOwnerChangedSignal)) {
+    Bus* self = static_cast<Bus*>(data);
+    self->OnServiceOwnerChanged(message);
+  }
+  // Always return unhandled to let others, e.g. ObjectProxies, handle the same
+  // signal.
+  return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
+}
+
 }  // namespace dbus