+2005-11-10 Andy Wingo <wingo@pobox.com>
+
+ * check/gst/gstbin.c (test_message_state_changed_children): Style
+ fix..
+
+ * gst/gstbus.c (poll_destroy, poll_func, gst_bus_poll): Implement
+ gst_bus_poll with the signal watch. Ensures that poll and a signal
+ watch see the same messages.
+
+ * check/gst/gstbus.c (test_watch_with_poll): New test, checks that
+ a poll and a watch at the same time get the same messages.
+
2005-11-10 Thomas Vander Stichele <thomas at apestaart dot org>
* gst/base/gstbasetransform.c: (gst_base_transform_transform_caps):
ASSERT_OBJECT_REFCOUNT (pipeline, "pipeline", 2);
pop_messages (bus, 3);
- fail_if ((gst_bus_pop (bus)) != NULL);
+ fail_if (gst_bus_have_pending (bus), "unexpected pending messages");
ASSERT_OBJECT_REFCOUNT (bus, "bus", 2);
ASSERT_OBJECT_REFCOUNT (src, "src", 1);
gst_object_unref ((GstObject *) test_bus);
}
+GST_END_TEST static gint messages_seen = 0;
+
+static void
+message_func (GstBus * bus, GstMessage * message, gpointer data)
+{
+ g_return_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_APPLICATION);
+
+ messages_seen++;
+}
+
+static void
+send_10_app_messages (void)
+{
+ GstMessage *m;
+ GstStructure *s;
+ gint i;
+
+ for (i = 0; i < 10; i++) {
+ s = gst_structure_new ("test_message", "msg_id", G_TYPE_INT, i, NULL);
+ m = gst_message_new_application (NULL, s);
+ gst_bus_post (test_bus, m);
+ }
+}
+
+/* test that you get the same messages from a poll as from signal watches. */
+GST_START_TEST (test_watch_with_poll)
+{
+ guint i;
+
+ test_bus = gst_bus_new ();
+
+ gst_bus_add_signal_watch (test_bus);
+ g_signal_connect (test_bus, "message", (GCallback) message_func, NULL);
+
+ send_10_app_messages ();
+
+ for (i = 0; i < 10; i++)
+ gst_message_unref (gst_bus_poll (test_bus, GST_MESSAGE_APPLICATION,
+ GST_CLOCK_TIME_NONE));
+
+ fail_if (gst_bus_have_pending (test_bus), "unexpected messages on bus");
+ fail_unless (messages_seen == 10, "signal handler didn't get 10 messages");
+
+ gst_bus_remove_signal_watch (test_bus);
+
+ gst_object_unref (test_bus);
+}
GST_END_TEST Suite * gstbus_suite (void)
{
Suite *s = suite_create ("GstBus");
suite_add_tcase (s, tc_chain);
tcase_add_test (tc_chain, test_hammer_bus);
tcase_add_test (tc_chain, test_watch);
+ tcase_add_test (tc_chain, test_watch_with_poll);
return s;
}
GstMessage *message;
} GstBusPollData;
-static gboolean
+static void
poll_func (GstBus * bus, GstMessage * message, GstBusPollData * poll_data)
{
if (!g_main_loop_is_running (poll_data->loop)) {
GST_DEBUG ("mainloop %p not running", poll_data->loop);
- return TRUE;
+ return;
}
if (GST_MESSAGE_TYPE (message) & poll_data->events) {
- g_return_val_if_fail (poll_data->message == NULL, FALSE);
+ g_return_if_fail (poll_data->message == NULL);
/* keep ref to message */
poll_data->message = gst_message_ref (message);
GST_DEBUG ("mainloop %p quit", poll_data->loop);
g_main_loop_quit (poll_data->loop);
}
-
- /* we always keep the source alive so that we don't accidentialy
- * free the poll_data */
- return TRUE;
}
static gboolean
}
static void
-poll_destroy (GstBusPollData * poll_data)
+poll_destroy (GstBusPollData * poll_data, gpointer unused)
{
poll_data->source_running = FALSE;
if (!poll_data->timeout_id) {
*
* All messages not in @events will be popped off the bus and will be ignored.
*
- * This function will enter the default mainloop while polling.
+ * Because poll is implemented using the "message" signal enabled by
+ * gst_bus_add_signal_watch(), calling gst_bus_poll() will cause the "message"
+ * signal to be emitted for every message that poll sees. Thus a "message"
+ * signal handler will see the same messages that this function sees -- neither
+ * will steal messages from the other.
+ *
+ * This function will run a main loop from the default main context when
+ * polling.
*
* Returns: The message that was received, or NULL if the poll timed out.
* The message is taken from the bus and needs to be unreffed after usage.
{
GstBusPollData *poll_data;
GstMessage *ret;
- guint id;
+ gulong id;
poll_data = g_new0 (GstBusPollData, 1);
poll_data->source_running = TRUE;
else
poll_data->timeout_id = 0;
- id = gst_bus_add_watch_full (bus, G_PRIORITY_DEFAULT,
- (GstBusFunc) poll_func, poll_data, (GDestroyNotify) poll_destroy);
+ id = g_signal_connect_data (bus, "message", G_CALLBACK (poll_func), poll_data,
+ (GClosureNotify) poll_destroy, 0);
+
+ /* these can be nested, so it's ok */
+ gst_bus_add_signal_watch (bus);
GST_DEBUG ("running mainloop %p", poll_data->loop);
g_main_loop_run (poll_data->loop);
GST_DEBUG ("mainloop stopped %p", poll_data->loop);
+
+ gst_bus_remove_signal_watch (bus);
+
/* holds a ref */
ret = poll_data->message;
if (poll_data->timeout_id)
g_source_remove (poll_data->timeout_id);
- /* poll_data may get destroyed at any time now */
- g_source_remove (id);
+ /* poll_data will be freed now */
+ g_signal_handler_disconnect (bus, id);
GST_DEBUG_OBJECT (bus, "finished poll with message %p", ret);
ASSERT_OBJECT_REFCOUNT (pipeline, "pipeline", 2);
pop_messages (bus, 3);
- fail_if ((gst_bus_pop (bus)) != NULL);
+ fail_if (gst_bus_have_pending (bus), "unexpected pending messages");
ASSERT_OBJECT_REFCOUNT (bus, "bus", 2);
ASSERT_OBJECT_REFCOUNT (src, "src", 1);
gst_object_unref ((GstObject *) test_bus);
}
+GST_END_TEST static gint messages_seen = 0;
+
+static void
+message_func (GstBus * bus, GstMessage * message, gpointer data)
+{
+ g_return_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_APPLICATION);
+
+ messages_seen++;
+}
+
+static void
+send_10_app_messages (void)
+{
+ GstMessage *m;
+ GstStructure *s;
+ gint i;
+
+ for (i = 0; i < 10; i++) {
+ s = gst_structure_new ("test_message", "msg_id", G_TYPE_INT, i, NULL);
+ m = gst_message_new_application (NULL, s);
+ gst_bus_post (test_bus, m);
+ }
+}
+
+/* test that you get the same messages from a poll as from signal watches. */
+GST_START_TEST (test_watch_with_poll)
+{
+ guint i;
+
+ test_bus = gst_bus_new ();
+
+ gst_bus_add_signal_watch (test_bus);
+ g_signal_connect (test_bus, "message", (GCallback) message_func, NULL);
+
+ send_10_app_messages ();
+
+ for (i = 0; i < 10; i++)
+ gst_message_unref (gst_bus_poll (test_bus, GST_MESSAGE_APPLICATION,
+ GST_CLOCK_TIME_NONE));
+
+ fail_if (gst_bus_have_pending (test_bus), "unexpected messages on bus");
+ fail_unless (messages_seen == 10, "signal handler didn't get 10 messages");
+
+ gst_bus_remove_signal_watch (test_bus);
+
+ gst_object_unref (test_bus);
+}
GST_END_TEST Suite * gstbus_suite (void)
{
Suite *s = suite_create ("GstBus");
suite_add_tcase (s, tc_chain);
tcase_add_test (tc_chain, test_hammer_bus);
tcase_add_test (tc_chain, test_watch);
+ tcase_add_test (tc_chain, test_watch_with_poll);
return s;
}
projects / platform / upstream / gstreamer.git / commitdiff