tests/check/gst/gstbus.c

   1 /* GStreamer
   2  * Copyright (C) 2005 Andy Wingo <wingo@pobox.com>
   3  *
   4  * gstbus.c: Unit test for the message bus
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 #include <gst/check/gstcheck.h>
  23
  24 static GstBus *test_bus = NULL;
  25 static GMainLoop *main_loop;
  26
  27 #define NUM_MESSAGES 1000
  28 #define NUM_THREADS 10
  29
  30 static gpointer
  31 pound_bus_with_messages (gpointer data)
  32 {
  33   gint thread_id = GPOINTER_TO_INT (data);
  34   gint i;
  35
  36   for (i = 0; i < NUM_MESSAGES; i++) {
  37     GstMessage *m;
  38     GstStructure *s;
  39
  40     s = gst_structure_new ("test_message",
  41         "thread_id", G_TYPE_INT, thread_id, "msg_id", G_TYPE_INT, i, NULL);
  42     m = gst_message_new_application (NULL, s);
  43     gst_bus_post (test_bus, m);
  44   }
  45   return NULL;
  46 }
  47
  48 static void
  49 pull_messages ()
  50 {
  51   GstMessage *m;
  52   const GstStructure *s;
  53   guint message_ids[NUM_THREADS];
  54   gint i;
  55
  56   for (i = 0; i < NUM_THREADS; i++)
  57     message_ids[i] = 0;
  58
  59   while (1) {
  60     gint _t, _i;
  61
  62     m = gst_bus_pop (test_bus);
  63     if (!m)
  64       break;
  65     g_return_if_fail (GST_MESSAGE_TYPE (m) == GST_MESSAGE_APPLICATION);
  66
  67     s = gst_message_get_structure (m);
  68     if (!gst_structure_get_int (s, "thread_id", &_t))
  69       g_critical ("Invalid message");
  70     if (!gst_structure_get_int (s, "msg_id", &_i))
  71       g_critical ("Invalid message");
  72
  73     g_return_if_fail (_t < NUM_THREADS);
  74     g_return_if_fail (_i == message_ids[_t]++);
  75
  76     gst_message_unref (m);
  77   }
  78
  79   for (i = 0; i < NUM_THREADS; i++)
  80     g_return_if_fail (message_ids[i] == NUM_MESSAGES);
  81 }
  82
  83 GST_START_TEST (test_hammer_bus)
  84 {
  85   GThread *threads[NUM_THREADS];
  86   gint i;
  87
  88   test_bus = gst_bus_new ();
  89
  90   for (i = 0; i < NUM_THREADS; i++)
  91     threads[i] = g_thread_create (pound_bus_with_messages, GINT_TO_POINTER (i),
  92         TRUE, NULL);
  93
  94   for (i = 0; i < NUM_THREADS; i++)
  95     g_thread_join (threads[i]);
  96
  97   pull_messages ();
  98
  99   gst_object_unref ((GstObject *) test_bus);
 100 }
 101
 102 GST_END_TEST;
 103
 104 static gboolean
 105 message_func_eos (GstBus * bus, GstMessage * message, gpointer data)
 106 {
 107   const GstStructure *s;
 108   gint i;
 109
 110   g_return_val_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_EOS, FALSE);
 111
 112   GST_DEBUG ("got EOS message");
 113
 114   s = gst_message_get_structure (message);
 115   if (!gst_structure_get_int (s, "msg_id", &i))
 116     g_critical ("Invalid message");
 117
 118   return i != 9;
 119 }
 120
 121 static gboolean
 122 message_func_app (GstBus * bus, GstMessage * message, gpointer data)
 123 {
 124   const GstStructure *s;
 125   gint i;
 126
 127   g_return_val_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_APPLICATION,
 128       FALSE);
 129
 130   GST_DEBUG ("got APP message");
 131
 132   s = gst_message_get_structure (message);
 133   if (!gst_structure_get_int (s, "msg_id", &i))
 134     g_critical ("Invalid message");
 135
 136   return i != 9;
 137 }
 138
 139 static gboolean
 140 send_messages (gpointer data)
 141 {
 142   GstMessage *m;
 143   GstStructure *s;
 144   gint i;
 145
 146   for (i = 0; i < 10; i++) {
 147     s = gst_structure_new ("test_message", "msg_id", G_TYPE_INT, i, NULL);
 148     m = gst_message_new_application (NULL, s);
 149     gst_bus_post (test_bus, m);
 150     s = gst_structure_new ("test_message", "msg_id", G_TYPE_INT, i, NULL);
 151     m = gst_message_new_custom (GST_MESSAGE_EOS, NULL, s);
 152     gst_bus_post (test_bus, m);
 153   }
 154
 155   return FALSE;
 156 }
 157
 158 /* test if adding a signal watch for different message types calls the
 159  * respective callbacks. */
 160 GST_START_TEST (test_watch)
 161 {
 162   guint id;
 163
 164   test_bus = gst_bus_new ();
 165
 166   main_loop = g_main_loop_new (NULL, FALSE);
 167
 168   id = gst_bus_add_watch (test_bus, gst_bus_async_signal_func, NULL);
 169   g_signal_connect (test_bus, "message::eos", (GCallback) message_func_eos,
 170       NULL);
 171   g_signal_connect (test_bus, "message::application",
 172       (GCallback) message_func_app, NULL);
 173
 174   g_idle_add ((GSourceFunc) send_messages, NULL);
 175   while (g_main_context_pending (NULL))
 176     g_main_context_iteration (NULL, FALSE);
 177
 178   g_source_remove (id);
 179   g_main_loop_unref (main_loop);
 180
 181   gst_object_unref ((GstObject *) test_bus);
 182 }
 183
 184 GST_END_TEST;
 185
 186 static gint messages_seen = 0;
 187
 188 static void
 189 message_func (GstBus * bus, GstMessage * message, gpointer data)
 190 {
 191   g_return_if_fail (GST_MESSAGE_TYPE (message) == GST_MESSAGE_APPLICATION);
 192
 193   messages_seen++;
 194 }
 195
 196 static void
 197 send_10_app_messages (void)
 198 {
 199   GstMessage *m;
 200   GstStructure *s;
 201   gint i;
 202
 203   for (i = 0; i < 10; i++) {
 204     s = gst_structure_new ("test_message", "msg_id", G_TYPE_INT, i, NULL);
 205     m = gst_message_new_application (NULL, s);
 206     gst_bus_post (test_bus, m);
 207   }
 208 }
 209
 210 /* test that you get the same messages from a poll as from signal watches. */
 211 GST_START_TEST (test_watch_with_poll)
 212 {
 213   guint i;
 214
 215   test_bus = gst_bus_new ();
 216
 217   gst_bus_add_signal_watch (test_bus);
 218   g_signal_connect (test_bus, "message", (GCallback) message_func, NULL);
 219
 220   send_10_app_messages ();
 221
 222   for (i = 0; i < 10; i++)
 223     gst_message_unref (gst_bus_poll (test_bus, GST_MESSAGE_APPLICATION,
 224             GST_CLOCK_TIME_NONE));
 225
 226   fail_if (gst_bus_have_pending (test_bus), "unexpected messages on bus");
 227   fail_unless (messages_seen == 10, "signal handler didn't get 10 messages");
 228
 229   gst_bus_remove_signal_watch (test_bus);
 230
 231   gst_object_unref (test_bus);
 232 }
 233
 234 GST_END_TEST;
 235
 236 Suite *
 237 gst_bus_suite (void)
 238 {
 239   Suite *s = suite_create ("GstBus");
 240   TCase *tc_chain = tcase_create ("stresstest");
 241
 242   tcase_set_timeout (tc_chain, 20);
 243
 244   suite_add_tcase (s, tc_chain);
 245   tcase_add_test (tc_chain, test_hammer_bus);
 246   tcase_add_test (tc_chain, test_watch);
 247   tcase_add_test (tc_chain, test_watch_with_poll);
 248   return s;
 249 }
 250
 251 GST_CHECK_MAIN (gst_bus);