gst/gstbus.c: Make GstBusSource work with non-default main contexts (#562170).

Original commit message from CVS:
* gst/gstbus.c: (gst_bus_dispose), (gst_bus_get_property),
(gst_bus_wakeup_main_context), (gst_bus_set_main_context),
(gst_bus_post), (gst_bus_source_prepare), (gst_bus_source_finalize),
(gst_bus_create_watch):
Make GstBusSource work with non-default main contexts (#562170).
* tests/check/gst/gstbus.c: (message_func_eos), (message_func_app),
(test_watch), (test_watch_with_custom_context), (gst_bus_suite):
Add test case for GstBusSource with a non-default main context.
* tests/check/libs/.cvsignore:
Ignore more.

ChangeLog

@@ -1,3 +1,18 @@
+2008-12-27  Tim-Philipp Müller  <tim.muller at collabora co uk>
+
+	* gst/gstbus.c: (gst_bus_dispose), (gst_bus_get_property),
+	  (gst_bus_wakeup_main_context), (gst_bus_set_main_context),
+	  (gst_bus_post), (gst_bus_source_prepare), (gst_bus_source_finalize),
+	  (gst_bus_create_watch):
+	  Make GstBusSource work with non-default main contexts (#562170).
+
+	* tests/check/gst/gstbus.c: (message_func_eos), (message_func_app),
+	  (test_watch), (test_watch_with_custom_context), (gst_bus_suite):
+	  Add test case for GstBusSource with a non-default main context.
+
+	* tests/check/libs/.cvsignore:
+	  Ignore more.
+
 2008-12-27  Tim-Philipp Müller  <tim.muller at collabora co uk>
 
 	* gst/gstregistrybinary.c: (unpack_element), (unpack_const_string),

gst/gstbus.c

@@ -95,13 +95,11 @@ static void gst_bus_set_property (GObject * object, guint prop_id,
     const GValue * value, GParamSpec * pspec);
 static void gst_bus_get_property (GObject * object, guint prop_id,
     GValue * value, GParamSpec * pspec);
+static void gst_bus_set_main_context (GstBus * bus, GMainContext * ctx);
 
 static GstObjectClass *parent_class = NULL;
 static guint gst_bus_signals[LAST_SIGNAL] = { 0 };
 
-/* the context we wakeup when we posted a message on the bus */
-static GMainContext *main_context;
-
 struct _GstBusPrivate
 {
   guint num_sync_message_emitters;
@@ -109,6 +107,8 @@ struct _GstBusPrivate
   GCond *queue_cond;
 
   GSource *watch_id;
+
+  GMainContext *main_context;
 };
 
 GType
@@ -217,8 +217,6 @@ gst_bus_class_init (GstBusClass * klass)
       G_STRUCT_OFFSET (GstBusClass, message), NULL, NULL,
       marshal_VOID__MINIOBJECT, G_TYPE_NONE, 1, GST_TYPE_MESSAGE);
 
-  main_context = g_main_context_default ();
-
   g_type_class_add_private (klass, sizeof (GstBusPrivate));
 }
 
@@ -259,6 +257,11 @@ gst_bus_dispose (GObject * object)
     bus->priv->queue_cond = NULL;
   }
 
+  if (bus->priv->main_context) {
+    g_main_context_unref (bus->priv->main_context);
+    bus->priv->main_context = NULL;
+  }
+
   G_OBJECT_CLASS (parent_class)->dispose (object);
 }
 
@@ -292,6 +295,31 @@ gst_bus_get_property (GObject * object, guint prop_id,
   }
 }
 
+static void
+gst_bus_wakeup_main_context (GstBus * bus)
+{
+  GST_OBJECT_LOCK (bus);
+  g_main_context_wakeup (bus->priv->main_context);
+  GST_OBJECT_UNLOCK (bus);
+}
+
+static void
+gst_bus_set_main_context (GstBus * bus, GMainContext * ctx)
+{
+  GST_OBJECT_LOCK (bus);
+
+  if (bus->priv->main_context != NULL) {
+    g_main_context_unref (bus->priv->main_context);
+    bus->priv->main_context = NULL;
+  }
+
+  if (ctx != NULL) {
+    bus->priv->main_context = g_main_context_ref (ctx);
+  }
+
+  GST_OBJECT_UNLOCK (bus);
+}
+
 /**
  * gst_bus_new:
  *
@@ -373,8 +401,7 @@ gst_bus_post (GstBus * bus, GstMessage * message)
       g_mutex_unlock (bus->queue_lock);
       GST_DEBUG_OBJECT (bus, "[msg %p] pushed on async queue", message);
 
-      /* FIXME cannot assume sources are only in the default context */
-      g_main_context_wakeup (main_context);
+      gst_bus_wakeup_main_context (bus);
 
       break;
     case GST_BUS_ASYNC:
@@ -398,8 +425,7 @@ gst_bus_post (GstBus * bus, GstMessage * message)
       g_cond_broadcast (bus->priv->queue_cond);
       g_mutex_unlock (bus->queue_lock);
 
-      /* FIXME cannot assume sources are only in the default context */
-      g_main_context_wakeup (main_context);
+      gst_bus_wakeup_main_context (bus);
 
       /* now block till the message is freed */
       g_cond_wait (cond, lock);
@@ -733,6 +759,7 @@ no_replace:
  */
 typedef struct
 {
+  gboolean inited;
   GSource source;
   GstBus *bus;
 } GstBusSource;
@@ -742,6 +769,14 @@ gst_bus_source_prepare (GSource * source, gint * timeout)
 {
   GstBusSource *bsrc = (GstBusSource *) source;
 
+  /* we do this here now that we know that we're attached to a main context
+   * (we don't support detaching a source from a main context and then
+   * re-attaching it to a different main context) */
+  if (G_UNLIKELY (!bsrc->inited)) {
+    gst_bus_set_main_context (bsrc->bus, g_source_get_context (source));
+    bsrc->inited = TRUE;
+  }
+
   *timeout = -1;
   return gst_bus_have_pending (bsrc->bus);
 }
@@ -813,6 +848,7 @@ gst_bus_source_finalize (GSource * source)
     bus->priv->watch_id = NULL;
   GST_OBJECT_UNLOCK (bus);
 
+  gst_bus_set_main_context (bsource->bus, NULL);
   gst_object_unref (bsource->bus);
   bsource->bus = NULL;
 }
@@ -843,8 +879,7 @@ gst_bus_create_watch (GstBus * bus)
 
   source = (GstBusSource *) g_source_new (&gst_bus_source_funcs,
       sizeof (GstBusSource));
-  gst_object_ref (bus);
-  source->bus = bus;
+  source->bus = gst_object_ref (bus);
 
   return (GSource *) source;
 }

tests/check/gst/gstbus.c

@@ -101,7 +101,7 @@ GST_START_TEST (test_hammer_bus)
 GST_END_TEST;
 
 static gboolean
-message_func_eos (GstBus * bus, GstMessage * message, gpointer data)
+message_func_eos (GstBus * bus, GstMessage * message, guint * p_counter)
 {
   const GstStructure *s;
   gint i;
@@ -114,11 +114,14 @@ message_func_eos (GstBus * bus, GstMessage * message, gpointer data)
   if (!gst_structure_get_int (s, "msg_id", &i))
     g_critical ("Invalid message");
 
+  if (p_counter != NULL)
+    *p_counter += 1;
+
   return i != 9;
 }
 
 static gboolean
-message_func_app (GstBus * bus, GstMessage * message, gpointer data)
+message_func_app (GstBus * bus, GstMessage * message, guint * p_counter)
 {
   const GstStructure *s;
   gint i;
@@ -132,6 +135,9 @@ message_func_app (GstBus * bus, GstMessage * message, gpointer data)
   if (!gst_structure_get_int (s, "msg_id", &i))
     g_critical ("Invalid message");
 
+  if (p_counter != NULL)
+    *p_counter += 1;
+
   return i != 9;
 }
 
@@ -158,6 +164,8 @@ send_messages (gpointer data)
  * respective callbacks. */
 GST_START_TEST (test_watch)
 {
+  guint num_eos = 0;
+  guint num_app = 0;
   guint id;
 
   test_bus = gst_bus_new ();
@@ -167,14 +175,17 @@ GST_START_TEST (test_watch)
   id = gst_bus_add_watch (test_bus, gst_bus_async_signal_func, NULL);
   fail_if (id == 0);
   g_signal_connect (test_bus, "message::eos", (GCallback) message_func_eos,
-      NULL);
+      &num_eos);
   g_signal_connect (test_bus, "message::application",
-      (GCallback) message_func_app, NULL);
+      (GCallback) message_func_app, &num_app);
 
   g_idle_add ((GSourceFunc) send_messages, NULL);
   while (g_main_context_pending (NULL))
     g_main_context_iteration (NULL, FALSE);
 
+  fail_unless_equals_int (num_eos, 10);
+  fail_unless_equals_int (num_app, 10);
+
   g_source_remove (id);
   g_main_loop_unref (main_loop);
 
@@ -183,6 +194,53 @@ GST_START_TEST (test_watch)
 
 GST_END_TEST;
 
+/* test if adding a signal watch for different message types calls the
+ * respective callbacks. */
+GST_START_TEST (test_watch_with_custom_context)
+{
+  GMainContext *ctx;
+  GSource *source;
+  guint num_eos = 0;
+  guint num_app = 0;
+  guint id;
+
+  test_bus = gst_bus_new ();
+
+  ctx = g_main_context_new ();
+  main_loop = g_main_loop_new (ctx, FALSE);
+
+  source = gst_bus_create_watch (test_bus);
+  g_source_set_callback (source, (GSourceFunc) gst_bus_async_signal_func, NULL,
+      NULL);
+  id = g_source_attach (source, ctx);
+  g_source_unref (source);
+  fail_if (id == 0);
+
+  g_signal_connect (test_bus, "message::eos", (GCallback) message_func_eos,
+      &num_eos);
+  g_signal_connect (test_bus, "message::application",
+      (GCallback) message_func_app, &num_app);
+
+  source = g_idle_source_new ();
+  g_source_set_callback (source, (GSourceFunc) send_messages, NULL, NULL);
+  g_source_attach (source, ctx);
+  g_source_unref (source);
+
+  while (g_main_context_pending (ctx))
+    g_main_context_iteration (ctx, FALSE);
+
+  fail_unless_equals_int (num_eos, 10);
+  fail_unless_equals_int (num_app, 10);
+
+  g_source_remove (id);
+  g_main_loop_unref (main_loop);
+  g_main_context_unref (ctx);
+
+  gst_object_unref (test_bus);
+}
+
+GST_END_TEST;
+
 static gint messages_seen;
 
 static void
@@ -441,6 +499,7 @@ gst_bus_suite (void)
   tcase_add_test (tc_chain, test_hammer_bus);
   tcase_add_test (tc_chain, test_watch);
   tcase_add_test (tc_chain, test_watch_with_poll);
+  tcase_add_test (tc_chain, test_watch_with_custom_context);
   tcase_add_test (tc_chain, test_timed_pop);
   tcase_add_test (tc_chain, test_timed_pop_thread);
   tcase_add_test (tc_chain, test_timed_pop_filtered);

tests/check/libs/.gitignore

@@ -7,5 +7,6 @@ controller
 gstnetclientclock
 gstnettimeprovider
 libsabi
+transform1
 typefindhelper
 *.check.xml