[glib/gdbus] Allow subscribing to signals from any sender

[David Zeuthen <davidz src gnome org>]

commit 1a0a8700d6de0f36818ba4b3923330f28f6dc87f
Author: David Zeuthen <davidz redhat com>
Date:   Thu Apr 30 12:35:26 2009 -0400

    Allow subscribing to signals from any sender
    
     - use locking to protect the data structures used for managing signal
       subscribers (invoking callbacks without holding the lock)
    
     - add test cases for signal subscription
---
 gdbus/gdbusconnection.c  |   82 +++++++++++++++++++++------
 gdbus/tests/connection.c |  137 ++++++++++++++++++++++++++++++++++++++++++++++
 gdbus/tests/sessionbus.c |    2 +-
 3 files changed, 201 insertions(+), 20 deletions(-)

diff --git a/gdbus/gdbusconnection.c b/gdbus/gdbusconnection.c
index ec528da..4fca538 100644
--- a/gdbus/gdbusconnection.c
+++ b/gdbus/gdbusconnection.c
@@ -1589,7 +1589,7 @@ is_signal_data_for_name_lost_or_acquired (SignalData *signal_data)
 /**
  * g_dbus_connection_dbus_1_signal_subscribe:
  * @connection: A #GDBusConnection.
- * @sender: Sender name to match on. Must be either <literal>org.freedesktop.DBus</literal> (for listening to signals from the message bus daemon) or a unique name.
+ * @sender: Sender name to match on. Must be either <literal>org.freedesktop.DBus</literal> (for listening to signals from the message bus daemon) or a unique name or %NULL to listen from all senders.
  * @interface_name: D-Bus interface name to match on or %NULL or to match on all interfaces.
  * @member: D-Bus signal name to match on or %NULL to match on all signals.
  * @object_path: Object path to match on or %NULL to match on all object paths.
@@ -1644,12 +1644,17 @@ g_dbus_connection_dbus_1_signal_subscribe (GDBusConnection     *connection,
 
   g_return_val_if_fail (G_IS_DBUS_CONNECTION (connection), 0);
   g_return_val_if_fail (g_dbus_connection_get_is_open (connection), 0);
-  g_return_val_if_fail (sender != NULL && (strcmp (sender, DBUS_SERVICE_DBUS) == 0 || sender[0] == ':'), 0);
+  g_return_val_if_fail (sender == NULL || ((strcmp (sender, DBUS_SERVICE_DBUS) == 0 || sender[0] == ':')), 0);
   g_return_val_if_fail (callback != NULL, 0);
   /* TODO: check that passed in data is well-formed */
 
+  G_LOCK (connection_lock);
+
   rule = args_to_rule (sender, interface_name, member, object_path, arg0);
 
+  if (sender == NULL)
+    sender = "";
+
   subscriber.callback = callback;
   subscriber.user_data = user_data;
   subscriber.id = _global_subscriber_id++; /* TODO: overflow etc. */
@@ -1701,6 +1706,8 @@ g_dbus_connection_dbus_1_signal_subscribe (GDBusConnection     *connection,
     }
   g_ptr_array_add (signal_data_array, signal_data);
 
+  G_UNLOCK (connection_lock);
+
   return subscriber.id;
 }
 
@@ -1725,6 +1732,8 @@ g_dbus_connection_dbus_1_signal_unsubscribe (GDBusConnection    *connection,
   GPtrArray *signal_data_array;
   guint n;
 
+  G_LOCK (connection_lock);
+
   signal_data = g_hash_table_lookup (connection->priv->map_id_to_signal_data,
                                      GUINT_TO_POINTER (subscription_id));
   if (signal_data == NULL)
@@ -1768,28 +1777,19 @@ g_dbus_connection_dbus_1_signal_unsubscribe (GDBusConnection    *connection,
   g_assert_not_reached ();
 
  out:
+  G_UNLOCK (connection_lock);
   ;
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
 
-/* do not call with any locks held */
 static void
-distribute_signals (GDBusConnection *connection,
-                    DBusMessage     *message)
+add_callbacks (GPtrArray   *signal_data_array,
+               GArray      *callbacks,
+               DBusMessage *message)
 {
-  const gchar *sender;
-  GPtrArray *signal_data_array;
   guint n, m;
 
-  sender = dbus_message_get_sender (message);
-  if (sender == NULL)
-    goto out;
-
-  signal_data_array = g_hash_table_lookup (connection->priv->map_sender_to_signal_data_array, sender);
-  if (signal_data_array == NULL)
-    goto out;
-
   /* TODO: if this is slow, then we can change signal_data_array into
    *       map_object_path_to_signal_data_array or something.
    */
@@ -1825,14 +1825,58 @@ distribute_signals (GDBusConnection *connection,
       for (m = 0; m < signal_data->subscribers->len; m++)
         {
           SignalSubscriber *subscriber;
-
           subscriber = &(g_array_index (signal_data->subscribers, SignalSubscriber, m));
 
-          subscriber->callback (connection,
-                                message,
-                                subscriber->user_data);
+          g_array_append_val (callbacks, *subscriber);
         }
     }
+}
+
+/* do not call with any locks held */
+static void
+distribute_signals (GDBusConnection *connection,
+                    DBusMessage     *message)
+{
+  const gchar *sender;
+  GPtrArray *signal_data_array;
+  GArray *callbacks;
+  guint n;
+
+  sender = dbus_message_get_sender (message);
+  if (sender == NULL)
+    goto out;
+
+  G_LOCK (connection_lock);
+
+  /* collect callbacks with lock held, invoke them without holding lock */
+  callbacks = g_array_new (FALSE, FALSE, sizeof (SignalSubscriber));
+
+  /* collect subcsribers that match on sender */
+  signal_data_array = g_hash_table_lookup (connection->priv->map_sender_to_signal_data_array, sender);
+  if (signal_data_array != NULL) {
+    add_callbacks (signal_data_array, callbacks, message);
+  }
+
+  /* collect subcsribers not matching on sender */
+  signal_data_array = g_hash_table_lookup (connection->priv->map_sender_to_signal_data_array, "");
+  if (signal_data_array != NULL) {
+    add_callbacks (signal_data_array, callbacks, message);
+  }
+
+  G_UNLOCK (connection_lock);
+
+  for (n = 0; n < callbacks->len; n++)
+    {
+      SignalSubscriber *subscriber;
+
+      subscriber = &(g_array_index (callbacks, SignalSubscriber, n));
+
+      subscriber->callback (connection,
+                            message,
+                            subscriber->user_data);
+    }
+
+  g_array_free (callbacks, TRUE);
 
 out:
   ;
diff --git a/gdbus/tests/connection.c b/gdbus/tests/connection.c
index 60267d1..966bfff 100644
--- a/gdbus/tests/connection.c
+++ b/gdbus/tests/connection.c
@@ -335,6 +335,142 @@ test_connection_send (void)
 }
 
 /* ---------------------------------------------------------------------------------------------------- */
+/* Connection signal tests */
+/* ---------------------------------------------------------------------------------------------------- */
+
+static void
+test_connection_signal_handler (GDBusConnection *connection,
+                                DBusMessage     *message,
+                                gpointer         user_data)
+{
+  gint *counter = user_data;
+  *counter += 1;
+  g_main_loop_quit (loop);
+}
+
+static gboolean
+test_connection_signal_quit_mainloop (gpointer user_data)
+{
+  g_main_loop_quit (loop);
+  return FALSE;
+}
+
+static void
+test_connection_signals (void)
+{
+  GDBusConnection *c1;
+  GDBusConnection *c2;
+  GDBusConnection *c3;
+  guint s1;
+  guint s2;
+  gint count_s1;
+  gint count_s2;
+  DBusMessage *m;
+
+  /**
+   * Bring up three separate connections
+   */
+  session_bus_up ();
+
+  /* if running with dbus-monitor, it claims the name :1.0 - so if we don't run with the monitor
+   * emulate this
+   */
+  if (g_getenv ("G_DBUS_MONITOR") == NULL)
+    {
+      c1 = g_dbus_connection_bus_get (G_BUS_TYPE_SESSION);
+      _g_assert_signal_received (c1, "opened");
+      g_assert (g_dbus_connection_get_is_open (c1));
+      g_assert (g_dbus_connection_get_is_initialized (c1));
+      g_object_unref (c1);
+    }
+  c1 = g_dbus_connection_bus_get (G_BUS_TYPE_SESSION);
+  _g_assert_signal_received (c1, "opened");
+  g_assert (g_dbus_connection_get_is_open (c1));
+  g_assert (g_dbus_connection_get_is_initialized (c1));
+  g_assert_cmpstr (g_dbus_connection_get_unique_name (c1), ==, ":1.1");
+  c2 = g_dbus_connection_bus_get_private (G_BUS_TYPE_SESSION);
+  _g_assert_signal_received (c2, "opened");
+  g_assert (g_dbus_connection_get_is_open (c2));
+  g_assert (g_dbus_connection_get_is_initialized (c2));
+  g_assert_cmpstr (g_dbus_connection_get_unique_name (c2), ==, ":1.2");
+  c3 = g_dbus_connection_bus_get_private (G_BUS_TYPE_SESSION);
+  _g_assert_signal_received (c3, "opened");
+  g_assert (g_dbus_connection_get_is_open (c3));
+  g_assert (g_dbus_connection_get_is_initialized (c3));
+  g_assert_cmpstr (g_dbus_connection_get_unique_name (c3), ==, ":1.3");
+
+  /**
+   * Install two signal handlers for the first connection
+   *
+   *  - Listen to the signal "Foo" from :1.2 (e.g. c2)
+   *  - Listen to the signal "Foo" from anyone (e.g. both c2 and c3)
+   *
+   * and the conut how many times this signal handler was invoked.
+   */
+  s1 = g_dbus_connection_dbus_1_signal_subscribe (c1,
+                                                  ":1.2",
+                                                  NULL,
+                                                  "Foo",
+                                                  NULL,
+                                                  NULL,
+                                                  test_connection_signal_handler,
+                                                  &count_s1);
+  s2 = g_dbus_connection_dbus_1_signal_subscribe (c1,
+                                                  NULL,
+                                                  NULL,
+                                                  "Foo",
+                                                  NULL,
+                                                  NULL,
+                                                  test_connection_signal_handler,
+                                                  &count_s2);
+  g_assert (s2 != 0);
+
+  count_s1 = 0;
+  count_s2 = 0;
+
+  m = dbus_message_new_signal ("/",
+                               "org.gtk.GDBus.ExampleInterface",
+                               "Foo");
+
+  /**
+   * Make c2 emit "Foo" - we should catch it twice
+   */
+  m = dbus_message_new_signal ("/",
+                               "org.gtk.GDBus.ExampleInterface",
+                               "Foo");
+  g_dbus_connection_send_dbus_1_message (c2, m);
+  while (!(count_s1 == 1 && count_s2 == 1))
+    g_main_loop_run (loop);
+  g_assert_cmpint (count_s1, ==, 1);
+  g_assert_cmpint (count_s2, ==, 1);
+
+  /**
+   * Make c3 emit "Foo" - we should catch it only once
+   */
+  g_dbus_connection_send_dbus_1_message (c3, m);
+  while (!(count_s1 == 1 && count_s2 == 2))
+    g_main_loop_run (loop);
+  g_assert_cmpint (count_s1, ==, 1);
+  g_assert_cmpint (count_s2, ==, 2);
+
+  /**
+   * Tool around it the mainloop to avoid race conditions
+   */
+  g_timeout_add (500, test_connection_signal_quit_mainloop, NULL);
+  g_main_loop_run (loop);
+  g_assert_cmpint (count_s1, ==, 1);
+  g_assert_cmpint (count_s2, ==, 2);
+
+  dbus_message_unref (m);
+  g_dbus_connection_dbus_1_signal_unsubscribe (c1, s1);
+  g_dbus_connection_dbus_1_signal_unsubscribe (c1, s2);
+  g_object_unref (c1);
+  g_object_unref (c2);
+  g_object_unref (c3);
+  session_bus_down ();
+}
+
+/* ---------------------------------------------------------------------------------------------------- */
 
 int
 main (int   argc,
@@ -354,5 +490,6 @@ main (int   argc,
 
   g_test_add_func ("/gdbus/connection-life-cycle", test_connection_life_cycle);
   g_test_add_func ("/gdbus/connection-send", test_connection_send);
+  g_test_add_func ("/gdbus/connection-signals", test_connection_signals);
   return g_test_run();
 }
diff --git a/gdbus/tests/sessionbus.c b/gdbus/tests/sessionbus.c
index 57ffc20..6bdafc1 100644
--- a/gdbus/tests/sessionbus.c
+++ b/gdbus/tests/sessionbus.c
@@ -246,7 +246,7 @@ session_bus_up_with_address (const gchar *given_address)
   write (pipe_fds[1], buf, strlen (buf));
 
   /* start dbus-monitor */
-  if (g_getenv ("GDBUS_MONITOR") != NULL)
+  if (g_getenv ("G_DBUS_MONITOR") != NULL)
     {
       g_spawn_command_line_async ("dbus-monitor --session", NULL);
       usleep (500 * 1000);