Multi process notifications on linux (dbus)

chrome/browser/multi_process_notification_linux.cc

 // Copyright (c) 2011 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 "chrome/browser/multi_process_notification.h"
 
+#include <dbus/dbus.h>
+#include <set>
+
+#include "base/basictypes.h"
+#include "base/file_path.h"
 #include "base/logging.h"
+#include "base/message_loop_proxy.h"
+#include "base/path_service.h"
+#include "base/stringprintf.h"
+#include "base/threading/simple_thread.h"
+#include "chrome/common/chrome_paths.h"
 
 namespace multi_process_notification {
 
-bool Post(const std::string& name, Domain domain) {
-  // TODO(dmaclach): Implement
-  NOTIMPLEMENTED();
-  return false;
-}
+const char* kNotifyPath = "/modules/notify";
+const char* kNotifyInterface = "org.chromium.Notify";
+const char* kNotifySignalName = "Notify";
+
+// A simple thread to run the DBus main loop.
+class ListenerThread : public base::SimpleThread {
+ public:
+  ListenerThread();
+
+  bool Init();
+
+  bool AddListener(ListenerImpl* listner);
+  bool RemoveListener(ListenerImpl* listner);
+
+  // SimpleThread overrides
+  virtual void Run();
+
+ private:
+  DBusConnection* connection_;
+
+  Lock listeners_lock_;
+  std::set<ListenerImpl*> listeners_;
+};
 
 class ListenerImpl {
  public:
   ListenerImpl(const std::string& name,
                Domain domain,
                Listener::Delegate* delegate);
+  virtual ~ListenerImpl();
 
   bool Start(MessageLoop* io_loop_to_listen_on);
+  void OnListen(const std::string& name);
 
   std::string name() const { return name_; }
   Domain domain() const { return domain_; }
 
  private:
+  void StartListener();
+
   std::string name_;
   Domain domain_;
   Listener::Delegate* delegate_;
 
+  DBusError error_;
+  DBusConnection* connection_;
+
+  static ListenerThread* g_thread_;
+
+  scoped_refptr<base::MessageLoopProxy> message_loop_proxy_;
+
   DISALLOW_COPY_AND_ASSIGN(ListenerImpl);
 };
 
+ListenerThread* ListenerImpl::g_thread_ = NULL;
+
+std::string AddPrefixToNotification(const std::string& name,
+                                    Domain domain) {
+  std::string prefix;
+  switch (domain) {
+    case multi_process_notification::ProfileDomain: {
+      FilePath user_data_dir;
+      if (!PathService::Get(chrome::DIR_USER_DATA, &user_data_dir)) {
+        NOTREACHED();
+      }
+      prefix = StringPrintf("user.%u.%s.", getuid(),
+                            user_data_dir.value().c_str());
+      break;
+    }
+    case multi_process_notification::UserDomain:
+      prefix = StringPrintf("user.%u.", getuid());
+      break;
+    case multi_process_notification::SystemDomain:
+      prefix = "";
+      break;
+  }
+  return prefix + name;
+}
+
+bool Post(const std::string& name, Domain domain) {
+  bool success = true;

[dmac, 2011/01/21 19:20:01]

move success decl down near where it is used.

[garykac, 2011/01/21 21:46:19]

Done.

+
+  DBusError error;
+  dbus_error_init(&error);

[dmac, 2011/01/21 19:20:01]

check error here?

[garykac, 2011/01/21 21:46:19]

This is initializing the DBusError struct, so there's no error to check.

+
+  // Get a connection to the DBus.
+  DBusConnection* connection = dbus_bus_get(DBUS_BUS_SESSION, &error);
+  if (dbus_error_is_set(&error)) {
+    LOG(ERROR) << "Failed to create initial dbus connection: " << error.message;
+    dbus_error_free(&error);
+    return false;
+  }
+  if (!connection) {
+    LOG(ERROR) << "Failed to create initial dbus connection";
+    return false;
+  }
+
+  // Construct the fullname for the signal by adding domain info as needed.
+  std::string fullname = AddPrefixToNotification(name, domain);
+
+  // Create the Notify signal.
+  DBusMessage* message = dbus_message_new_signal(kNotifyPath, kNotifyInterface,

[dmac, 2011/01/21 19:20:01]

create a wrapper class for this that does the unref for us. I think if you
search the code base you will find that there is already one defined.

[garykac, 2011/01/21 21:46:19]

There's no generic 'scoped_ptr' wrapper for these types and this file contains
the only occurrences of DbusMessage in Chrome.

wifi_data_provider_linux.cc has something that could be adapted if needed.

+                                                 kNotifySignalName);
+  if (!message) {
+    LOG(ERROR) << "Failed to create message for signal.";
+    success = false;
+  }
+
+  // Add the user-defined signal name as an argument to the Notify signal.
+  if (success) {
+    DBusMessageIter args;
+    const char* cname = fullname.c_str();
+    dbus_message_iter_init_append(message, &args);
+    if (!dbus_message_iter_append_basic(&args,
+                                        DBUS_TYPE_STRING, &cname)) {
+      LOG(ERROR) << "Failed to set signal name.";

[dmac, 2011/01/21 19:20:01]

maybe pass the name in along with the error?

[garykac, 2011/01/21 21:46:19]

Done.

+      success = false;
+    }
+  }
+
+  // Actually send the signal.
+  if (success) {
+    dbus_uint32_t serial = 0;
+    if (!dbus_connection_send(connection, message, &serial)) {
+      LOG(ERROR) << "Unable to send message.";
+      success = false;
+    }
+  }
+
+  if (success) {
+    dbus_connection_flush(connection);
+  }
+  dbus_message_unref(message);
+
+  return success;
+}
+
+ListenerThread::ListenerThread()
+    : base::SimpleThread("ListenerThread"),
+      connection_(NULL) {
+}
+
+bool ListenerThread::Init() {
+  DBusError error;
+  dbus_error_init(&error);

[dmac, 2011/01/21 19:20:01]

create a wrapper class for dbus_error handling as well? Maybe there is one of
these that is shared too?

[garykac, 2011/01/21 21:46:19]

wifi_data_provider_linux.cc uses CheckError, and I considered doing something
similar.
The downside of this approach is that they have a single generic error message
and rely on error.message for context. I wanted to provide more background info
for the error.

+
+  // Get a connection to the DBus.
+  connection_ = dbus_bus_get(DBUS_BUS_SESSION, &error);
+  if (dbus_error_is_set(&error)) {
+    LOG(ERROR) << "Failed to create initial dbus connection: "
+               << error.message;
+    dbus_error_free(&error);
+    return false;
+  }
+  if (!connection_) {
+    LOG(ERROR) << "Failed to create initial dbus connection";
+    return false;
+  }
+
+  // Create matching rule for our signal type.
+  std::string match_rule = StringPrintf("type='signal',interface='%s'",
+                                        kNotifyInterface);
+  dbus_bus_add_match(connection_, match_rule.c_str(), &error);
+  dbus_connection_flush(connection_);
+  if (dbus_error_is_set(&error)) {
+    LOG(ERROR) << "Failed to add match rule: "
+               << error.message;
+    dbus_error_free(&error);
+    return false;
+  }
+
+  return true;
+}
+
+bool ListenerThread::AddListener(ListenerImpl* listener) {
+  base::AutoLock autolock(listeners_lock_);
+  listeners_.insert(listener);
+  return true;
+}
+
+bool ListenerThread::RemoveListener(ListenerImpl* listener) {
+  base::AutoLock autolock(listeners_lock_);
+  listeners_.erase(listener);
+  return true;
+}
+
+void ListenerThread::Run() {
+  bool done = false;
+
+  while (!done) {
+    // Get next available message.
+    // Using -1 for a timeout to make this a blocking call blocks all dbus
+    // calls on this connection. Thus, we use a timeout of 0 and a sleep().
+    // TODO(garykac): Ugh. Fix this to use file descriptors.
+    dbus_connection_read_write(connection_, 0);
+
+    DBusMessage* message = dbus_connection_pop_message(connection_);
+    if (!message) {
+      sleep(1);  // Yuck!
+      continue;
+    }
+
+    // Process all queued up messages.
+    while (message) {
+      if (dbus_message_is_signal(message,
+                                 kNotifyInterface, kNotifySignalName)) {
+        // Get user-defined name from the signal arguments.
+        DBusMessageIter args;
+        if (!dbus_message_iter_init(message, &args)) {
+          LOG(ERROR) << "Params missing from dbus signal";
+        } else if (dbus_message_iter_get_arg_type(&args) != DBUS_TYPE_STRING) {
+          LOG(ERROR) << "Dbus signal param is not string type";
+        } else {
+          char* name;
+          dbus_message_iter_get_basic(&args, &name);
+          {  // Scope for lock
+            base::AutoLock autolock(listeners_lock_);
+            std::set<ListenerImpl*>::iterator it;
+            // Note that this doesn't scale well to a large number of listeners.
+            // We should only have a couple active, so this isn't a problem.
+            for (it=listeners_.begin(); it!=listeners_.end(); it++) {
+              (*it)->OnListen(name);
+            }
+          }
+        }
+      }
+
+      dbus_message_unref(message);
+      message = dbus_connection_pop_message(connection_);
+    }
+  }
+}
+
 ListenerImpl::ListenerImpl(const std::string& name,
                            Domain domain,
                            Listener::Delegate* delegate)
-    : name_(name), domain_(domain), delegate_(delegate) {
+    : name_(name),
+      domain_(domain),
+      delegate_(delegate),
+      connection_(NULL) {
+}
+
+ListenerImpl::~ListenerImpl() {
+  if (g_thread_) {
+    g_thread_->RemoveListener(this);
+  }
 }
 
 bool ListenerImpl::Start(MessageLoop* io_loop_to_listen_on) {
-  // TODO(dmaclach): Implement
-  NOTIMPLEMENTED();
-  return false;
+  if (io_loop_to_listen_on->type() != MessageLoop::TYPE_IO) {
+    DLOG(ERROR) << "io_loop_to_listen_on must be TYPE_IO";
+    return false;
+  }
+  // Start the listener on the IO thread.
+  message_loop_proxy_ = base::MessageLoopProxy::CreateForCurrentThread();
+  Task* task = NewRunnableMethod(this, &ListenerImpl::StartListener);
+  io_loop_to_listen_on->PostTask(FROM_HERE, task);
+  return true;
+}
+
+void ListenerImpl::StartListener() {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  bool success = true;
+
+  if (!g_thread_) {

[dmac, 2011/01/21 19:20:01]

need a lock around this thread creation. Multiple threads can call startListener
so you would have a race condition.

[garykac, 2011/01/21 21:46:19]

Yikes!  Done.

+    g_thread_ = new ListenerThread();
+    success = g_thread_->Init();
+    if (success) {
+      g_thread_->Start();
+    }
+  }
+
+  g_thread_->AddListener(this);
+
+  // Send initialization success/fail status to delegate.
+  Task* task = new Listener::ListenerStartedTask(name_, domain_, delegate_,
+                                                 success);
+  CHECK(message_loop_proxy_->PostTask(FROM_HERE, task));
+}
+
+void ListenerImpl::OnListen(const std::string& name) {
+  std::string myname = AddPrefixToNotification(name_, domain_);
+  // Ignore the signal unless it matches our name.
+  if (name == myname) {
+    Task* task =
+        new Listener::NotificationReceivedTask(name_, domain_, delegate_);
+    CHECK(message_loop_proxy_->PostTask(FROM_HERE, task));
+  }
 }
 
 Listener::Listener(const std::string& name,
                    Domain domain,
                    Listener::Delegate* delegate)
     : impl_(new ListenerImpl(name, domain, delegate)) {
 }
 
 Listener::~Listener() {
 }
 
 bool Listener::Start(MessageLoop* io_loop_to_listen_on) {
   return impl_->Start(io_loop_to_listen_on);
 }
 
 std::string Listener::name() const {
   return impl_->name();
 }
 
 Domain Listener::domain() const {
   return impl_->domain();
 }
 
 }  // namespace multi_process_notification
+
+DISABLE_RUNNABLE_METHOD_REFCOUNT(multi_process_notification::ListenerImpl);