gst/gstbus.c

   1 /* GStreamer
   2  * Copyright (C) 2004 Wim Taymans <wim@fluendo.com>
   3  *
   4  * gstbus.c: GstBus subsystem
   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 /**
  23  * SECTION:gstbus
  24  * @short_description: Asynchronous message bus subsystem
  25  * @see_also: #GstMessage, #GstElement
  26  *
  27  * The #GstBus is an object responsible for delivering #GstMessages in
  28  * a first-in first-out way from the streaming threads to the application.
  29  *
  30  * Since the application typically only wants to deal with delivery of these
  31  * messages from one thread, the GstBus will marshall the messages between
  32  * different threads. This is important since the actual streaming of media
  33  * is done in another thread than the application.
  34  *
  35  * The GstBus provides support for #GSource based notifications. This makes it
  36  * possible to handle the delivery in the glib mainloop.
  37  *
  38  * The #GSource callback function gst_bus_async_signal_func() can be used to
  39  * convert all bus messages into signal emissions.
  40  *
  41  * A message is posted on the bus with the gst_bus_post() method. With the
  42  * gst_bus_peek() and gst_bus_pop() methods one can look at or retrieve a
  43  * previously posted message.
  44  *
  45  * The bus can be polled with the gst_bus_poll() method. This methods blocks
  46  * up to the specified timeout value until one of the specified messages types
  47  * is posted on the bus. The application can then _pop() the messages from the
  48  * bus to handle them.
  49  * Alternatively the application can register an asynchronous bus function
  50  * using gst_bus_add_watch_full() or gst_bus_add_watch(). This function will
  51  * install a #GSource in the default glib main loop and will deliver messages
  52  * a short while after they have been posted. Note that the main loop should
  53  * be running for the asynchronous callbacks.
  54  *
  55  * It is also possible to get messages from the bus without any thread
  56  * marshalling with the gst_bus_set_sync_handler() method. This makes it
  57  * possible to react to a message in the same thread that posted the
  58  * message on the bus. This should only be used if the application is able
  59  * to deal with messages from different threads.
  60  *
  61  * Every #GstPipeline has one bus.
  62  *
  63  * Note that a #GstPipeline will set its bus into flushing state when changing
  64  * from READY to NULL state.
  65  *
  66  * Last reviewed on 2006-03-12 (0.10.5)
  67  */
  68
  69 #include "gst_private.h"
  70 #include <errno.h>
  71 #ifdef HAVE_UNISTD_H
  72 #  include <unistd.h>
  73 #endif
  74 #include <sys/types.h>
  75
  76 #include "gstinfo.h"
  77
  78 #include "gstbus.h"
  79
  80 #define GST_CAT_DEFAULT GST_CAT_BUS
  81 /* bus signals */
  82 enum
  83 {
  84   SYNC_MESSAGE,
  85   ASYNC_MESSAGE,
  86   /* add more above */
  87   LAST_SIGNAL
  88 };
  89
  90 static void gst_bus_class_init (GstBusClass * klass);
  91 static void gst_bus_init (GstBus * bus);
  92 static void gst_bus_dispose (GObject * object);
  93
  94 static void gst_bus_set_main_context (GstBus * bus, GMainContext * ctx);
  95
  96 static GstObjectClass *parent_class = NULL;
  97 static guint gst_bus_signals[LAST_SIGNAL] = { 0 };
  98
  99 struct _GstBusPrivate
 100 {
 101   guint num_sync_message_emitters;
 102   GCond *queue_cond;
 103   GSource *watch_id;
 104   GMainContext *main_context;
 105 };
 106
 107 G_DEFINE_TYPE (GstBus, gst_bus, GST_TYPE_OBJECT);
 108
 109 /* fixme: do something about this */
 110 static void
 111 marshal_VOID__MINIOBJECT (GClosure * closure, GValue * return_value,
 112     guint n_param_values, const GValue * param_values, gpointer invocation_hint,
 113     gpointer marshal_data)
 114 {
 115   typedef void (*marshalfunc_VOID__MINIOBJECT) (gpointer obj, gpointer arg1,
 116       gpointer data2);
 117   register marshalfunc_VOID__MINIOBJECT callback;
 118   register GCClosure *cc = (GCClosure *) closure;
 119   register gpointer data1, data2;
 120
 121   g_return_if_fail (n_param_values == 2);
 122
 123   if (G_CCLOSURE_SWAP_DATA (closure)) {
 124     data1 = closure->data;
 125     data2 = g_value_peek_pointer (param_values + 0);
 126   } else {
 127     data1 = g_value_peek_pointer (param_values + 0);
 128     data2 = closure->data;
 129   }
 130   callback =
 131       (marshalfunc_VOID__MINIOBJECT) (marshal_data ? marshal_data :
 132       cc->callback);
 133
 134   callback (data1, gst_value_get_mini_object (param_values + 1), data2);
 135 }
 136
 137 static void
 138 gst_bus_class_init (GstBusClass * klass)
 139 {
 140   GObjectClass *gobject_class = (GObjectClass *) klass;
 141
 142   parent_class = g_type_class_peek_parent (klass);
 143
 144   gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_bus_dispose);
 145
 146   /**
 147    * GstBus::sync-message:
 148    * @bus: the object which received the signal
 149    * @message: the message that has been posted synchronously
 150    *
 151    * A message has been posted on the bus. This signal is emitted from the
 152    * thread that posted the message so one has to be careful with locking.
 153    *
 154    * This signal will not be emitted by default, you have to set up
 155    * gst_bus_sync_signal_handler() as a sync handler if you want this
 156    * signal to be emitted when a message is posted on the bus, like this:
 157    * <programlisting>
 158    * gst_bus_set_sync_handler (bus, gst_bus_sync_signal_handler, yourdata);
 159    * </programlisting>
 160    */
 161   gst_bus_signals[SYNC_MESSAGE] =
 162       g_signal_new ("sync-message", G_TYPE_FROM_CLASS (klass),
 163       G_SIGNAL_RUN_LAST | G_SIGNAL_DETAILED,
 164       G_STRUCT_OFFSET (GstBusClass, sync_message), NULL, NULL,
 165       marshal_VOID__MINIOBJECT, G_TYPE_NONE, 1, GST_TYPE_MESSAGE);
 166
 167   /**
 168    * GstBus::message:
 169    * @bus: the object which received the signal
 170    * @message: the message that has been posted asynchronously
 171    *
 172    * A message has been posted on the bus. This signal is emitted from a
 173    * GSource added to the mainloop. this signal will only be emitted when
 174    * there is a mainloop running.
 175    */
 176   gst_bus_signals[ASYNC_MESSAGE] =
 177       g_signal_new ("message", G_TYPE_FROM_CLASS (klass),
 178       G_SIGNAL_RUN_LAST | G_SIGNAL_DETAILED,
 179       G_STRUCT_OFFSET (GstBusClass, message), NULL, NULL,
 180       marshal_VOID__MINIOBJECT, G_TYPE_NONE, 1, GST_TYPE_MESSAGE);
 181
 182   g_type_class_add_private (klass, sizeof (GstBusPrivate));
 183 }
 184
 185 static void
 186 gst_bus_init (GstBus * bus)
 187 {
 188   bus->queue = g_queue_new ();
 189   bus->queue_lock = g_mutex_new ();
 190
 191   bus->priv = G_TYPE_INSTANCE_GET_PRIVATE (bus, GST_TYPE_BUS, GstBusPrivate);
 192   bus->priv->queue_cond = g_cond_new ();
 193
 194   GST_DEBUG_OBJECT (bus, "created");
 195 }
 196
 197 static void
 198 gst_bus_dispose (GObject * object)
 199 {
 200   GstBus *bus = GST_BUS (object);
 201
 202   if (bus->queue) {
 203     GstMessage *message;
 204
 205     g_mutex_lock (bus->queue_lock);
 206     do {
 207       message = g_queue_pop_head (bus->queue);
 208       if (message)
 209         gst_message_unref (message);
 210     } while (message != NULL);
 211     g_queue_free (bus->queue);
 212     bus->queue = NULL;
 213     g_mutex_unlock (bus->queue_lock);
 214     g_mutex_free (bus->queue_lock);
 215     bus->queue_lock = NULL;
 216     g_cond_free (bus->priv->queue_cond);
 217     bus->priv->queue_cond = NULL;
 218   }
 219
 220   if (bus->priv->main_context) {
 221     g_main_context_unref (bus->priv->main_context);
 222     bus->priv->main_context = NULL;
 223   }
 224
 225   G_OBJECT_CLASS (parent_class)->dispose (object);
 226 }
 227
 228 static void
 229 gst_bus_wakeup_main_context (GstBus * bus)
 230 {
 231   GMainContext *ctx;
 232
 233   GST_OBJECT_LOCK (bus);
 234   if ((ctx = bus->priv->main_context))
 235     g_main_context_ref (ctx);
 236   GST_OBJECT_UNLOCK (bus);
 237
 238   g_main_context_wakeup (ctx);
 239
 240   if (ctx)
 241     g_main_context_unref (ctx);
 242 }
 243
 244 static void
 245 gst_bus_set_main_context (GstBus * bus, GMainContext * ctx)
 246 {
 247   GST_OBJECT_LOCK (bus);
 248
 249   if (bus->priv->main_context != NULL) {
 250     g_main_context_unref (bus->priv->main_context);
 251     bus->priv->main_context = NULL;
 252   }
 253
 254   if (ctx != NULL) {
 255     bus->priv->main_context = g_main_context_ref (ctx);
 256   }
 257
 258   GST_DEBUG_OBJECT (bus, "setting main context to %p, GLib default context: %p",
 259       ctx, g_main_context_default ());
 260
 261   GST_OBJECT_UNLOCK (bus);
 262 }
 263
 264 /**
 265  * gst_bus_new:
 266  *
 267  * Creates a new #GstBus instance.
 268  *
 269  * Returns: a new #GstBus instance
 270  */
 271 GstBus *
 272 gst_bus_new (void)
 273 {
 274   GstBus *result;
 275
 276   result = g_object_new (gst_bus_get_type (), NULL);
 277   GST_DEBUG_OBJECT (result, "created new bus");
 278
 279   return result;
 280 }
 281
 282 /**
 283  * gst_bus_post:
 284  * @bus: a #GstBus to post on
 285  * @message: The #GstMessage to post
 286  *
 287  * Post a message on the given bus. Ownership of the message
 288  * is taken by the bus.
 289  *
 290  * Returns: TRUE if the message could be posted, FALSE if the bus is flushing.
 291  *
 292  * MT safe.
 293  */
 294 gboolean
 295 gst_bus_post (GstBus * bus, GstMessage * message)
 296 {
 297   GstBusSyncReply reply = GST_BUS_PASS;
 298   GstBusSyncHandler handler;
 299   gboolean emit_sync_message;
 300   gpointer handler_data;
 301
 302   g_return_val_if_fail (GST_IS_BUS (bus), FALSE);
 303   g_return_val_if_fail (GST_IS_MESSAGE (message), FALSE);
 304
 305   GST_DEBUG_OBJECT (bus, "[msg %p] posting on bus, type %s, %" GST_PTR_FORMAT
 306       " from source %" GST_PTR_FORMAT,
 307       message, GST_MESSAGE_TYPE_NAME (message), message->structure,
 308       message->src);
 309
 310   GST_OBJECT_LOCK (bus);
 311   /* check if the bus is flushing */
 312   if (GST_OBJECT_FLAG_IS_SET (bus, GST_BUS_FLUSHING))
 313     goto is_flushing;
 314
 315   handler = bus->sync_handler;
 316   handler_data = bus->sync_handler_data;
 317   emit_sync_message = bus->priv->num_sync_message_emitters > 0;
 318   GST_OBJECT_UNLOCK (bus);
 319
 320   /* first call the sync handler if it is installed */
 321   if (handler)
 322     reply = handler (bus, message, handler_data);
 323
 324   /* emit sync-message if requested to do so via
 325      gst_bus_enable_sync_message_emission. terrible but effective */
 326   if (emit_sync_message && reply != GST_BUS_DROP
 327       && handler != gst_bus_sync_signal_handler)
 328     gst_bus_sync_signal_handler (bus, message, NULL);
 329
 330   /* now see what we should do with the message */
 331   switch (reply) {
 332     case GST_BUS_DROP:
 333       /* drop the message */
 334       GST_DEBUG_OBJECT (bus, "[msg %p] dropped", message);
 335       break;
 336     case GST_BUS_PASS:
 337       /* pass the message to the async queue, refcount passed in the queue */
 338       GST_DEBUG_OBJECT (bus, "[msg %p] pushing on async queue", message);
 339       g_mutex_lock (bus->queue_lock);
 340       g_queue_push_tail (bus->queue, message);
 341       g_cond_broadcast (bus->priv->queue_cond);
 342       g_mutex_unlock (bus->queue_lock);
 343       GST_DEBUG_OBJECT (bus, "[msg %p] pushed on async queue", message);
 344
 345       gst_bus_wakeup_main_context (bus);
 346
 347       break;
 348     case GST_BUS_ASYNC:
 349     {
 350       /* async delivery, we need a mutex and a cond to block
 351        * on */
 352       GMutex *lock = g_mutex_new ();
 353       GCond *cond = g_cond_new ();
 354
 355       GST_MESSAGE_COND (message) = cond;
 356       GST_MESSAGE_GET_LOCK (message) = lock;
 357
 358       GST_DEBUG_OBJECT (bus, "[msg %p] waiting for async delivery", message);
 359
 360       /* now we lock the message mutex, send the message to the async
 361        * queue. When the message is handled by the app and destroyed,
 362        * the cond will be signalled and we can continue */
 363       g_mutex_lock (lock);
 364       g_mutex_lock (bus->queue_lock);
 365       g_queue_push_tail (bus->queue, message);
 366       g_cond_broadcast (bus->priv->queue_cond);
 367       g_mutex_unlock (bus->queue_lock);
 368
 369       gst_bus_wakeup_main_context (bus);
 370
 371       /* now block till the message is freed */
 372       g_cond_wait (cond, lock);
 373       g_mutex_unlock (lock);
 374
 375       GST_DEBUG_OBJECT (bus, "[msg %p] delivered asynchronously", message);
 376
 377       g_mutex_free (lock);
 378       g_cond_free (cond);
 379       break;
 380     }
 381     default:
 382       g_warning ("invalid return from bus sync handler");
 383       break;
 384   }
 385   return TRUE;
 386
 387   /* ERRORS */
 388 is_flushing:
 389   {
 390     GST_DEBUG_OBJECT (bus, "bus is flushing");
 391     gst_message_unref (message);
 392     GST_OBJECT_UNLOCK (bus);
 393
 394     return FALSE;
 395   }
 396 }
 397
 398 /**
 399  * gst_bus_have_pending:
 400  * @bus: a #GstBus to check
 401  *
 402  * Check if there are pending messages on the bus that
 403  * should be handled.
 404  *
 405  * Returns: TRUE if there are messages on the bus to be handled, FALSE
 406  * otherwise.
 407  *
 408  * MT safe.
 409  */
 410 gboolean
 411 gst_bus_have_pending (GstBus * bus)
 412 {
 413   gboolean result;
 414
 415   g_return_val_if_fail (GST_IS_BUS (bus), FALSE);
 416
 417   g_mutex_lock (bus->queue_lock);
 418   /* see if there is a message on the bus */
 419   result = !g_queue_is_empty (bus->queue);
 420   g_mutex_unlock (bus->queue_lock);
 421
 422   return result;
 423 }
 424
 425 /**
 426  * gst_bus_set_flushing:
 427  * @bus: a #GstBus
 428  * @flushing: whether or not to flush the bus
 429  *
 430  * If @flushing, flush out and unref any messages queued in the bus. Releases
 431  * references to the message origin objects. Will flush future messages until
 432  * gst_bus_set_flushing() sets @flushing to #FALSE.
 433  *
 434  * MT safe.
 435  */
 436 void
 437 gst_bus_set_flushing (GstBus * bus, gboolean flushing)
 438 {
 439   GstMessage *message;
 440
 441   GST_OBJECT_LOCK (bus);
 442
 443   if (flushing) {
 444     GST_OBJECT_FLAG_SET (bus, GST_BUS_FLUSHING);
 445
 446     GST_DEBUG_OBJECT (bus, "set bus flushing");
 447
 448     while ((message = gst_bus_pop (bus)))
 449       gst_message_unref (message);
 450   } else {
 451     GST_DEBUG_OBJECT (bus, "unset bus flushing");
 452     GST_OBJECT_FLAG_UNSET (bus, GST_BUS_FLUSHING);
 453   }
 454
 455   GST_OBJECT_UNLOCK (bus);
 456 }
 457
 458 /**
 459  * gst_bus_timed_pop_filtered:
 460  * @bus: a #GstBus to pop from
 461  * @timeout: a timeout in nanoseconds, or GST_CLOCK_TIME_NONE to wait forever
 462  * @types: message types to take into account, GST_MESSAGE_ANY for any type
 463  *
 464  * Get a message from the bus whose type matches the message type mask @types,
 465  * waiting up to the specified timeout (and discarding any messages that do not
 466  * match the mask provided).
 467  *
 468  * If @timeout is 0, this function behaves like gst_bus_pop_filtered(). If
 469  * @timeout is #GST_CLOCK_TIME_NONE, this function will block forever until a
 470  * matching message was posted on the bus.
 471  *
 472  * Returns: a #GstMessage matching the filter in @types, or NULL if no matching
 473  * message was found on the bus until the timeout expired.
 474  * The message is taken from the bus and needs to be unreffed with
 475  * gst_message_unref() after usage.
 476  *
 477  * MT safe.
 478  *
 479  * Since: 0.10.15
 480  */
 481 GstMessage *
 482 gst_bus_timed_pop_filtered (GstBus * bus, GstClockTime timeout,
 483     GstMessageType types)
 484 {
 485   GstMessage *message;
 486   GTimeVal *timeval, abstimeout;
 487   gboolean first_round = TRUE;
 488
 489   g_return_val_if_fail (GST_IS_BUS (bus), NULL);
 490   g_return_val_if_fail (types != 0, NULL);
 491
 492   g_mutex_lock (bus->queue_lock);
 493
 494   while (TRUE) {
 495     GST_LOG_OBJECT (bus, "have %d messages", g_queue_get_length (bus->queue));
 496
 497     while ((message = g_queue_pop_head (bus->queue))) {
 498       GST_DEBUG_OBJECT (bus, "got message %p, %s, type mask is %u",
 499           message, GST_MESSAGE_TYPE_NAME (message), (guint) types);
 500       if ((GST_MESSAGE_TYPE (message) & types) != 0) {
 501         /* exit the loop, we have a message */
 502         goto beach;
 503       } else {
 504         GST_DEBUG_OBJECT (bus, "discarding message, does not match mask");
 505         gst_message_unref (message);
 506         message = NULL;
 507       }
 508     }
 509
 510     /* no need to wait, exit loop */
 511     if (timeout == 0)
 512       break;
 513
 514     if (timeout == GST_CLOCK_TIME_NONE) {
 515       /* wait forever */
 516       timeval = NULL;
 517     } else if (first_round) {
 518       glong add = timeout / 1000;
 519
 520       if (add == 0)
 521         /* no need to wait */
 522         break;
 523
 524       /* make timeout absolute */
 525       g_get_current_time (&abstimeout);
 526       g_time_val_add (&abstimeout, add);
 527       timeval = &abstimeout;
 528       first_round = FALSE;
 529       GST_DEBUG_OBJECT (bus, "blocking for message, timeout %ld", add);
 530     } else {
 531       /* calculated the absolute end time already, no need to do it again */
 532       GST_DEBUG_OBJECT (bus, "blocking for message, again");
 533       timeval = &abstimeout;    /* fool compiler */
 534     }
 535     if (!g_cond_timed_wait (bus->priv->queue_cond, bus->queue_lock, timeval)) {
 536       GST_INFO_OBJECT (bus, "timed out, breaking loop");
 537       break;
 538     } else {
 539       GST_INFO_OBJECT (bus, "we got woken up, recheck for message");
 540     }
 541   }
 542
 543 beach:
 544
 545   g_mutex_unlock (bus->queue_lock);
 546
 547   return message;
 548 }
 549
 550
 551 /**
 552  * gst_bus_timed_pop:
 553  * @bus: a #GstBus to pop
 554  * @timeout: a timeout
 555  *
 556  * Get a message from the bus, waiting up to the specified timeout.
 557  *
 558  * If @timeout is 0, this function behaves like gst_bus_pop(). If @timeout is
 559  * #GST_CLOCK_TIME_NONE, this function will block forever until a message was
 560  * posted on the bus.
 561  *
 562  * Returns: The #GstMessage that is on the bus after the specified timeout
 563  * or NULL if the bus is empty after the timeout expired.
 564  * The message is taken from the bus and needs to be unreffed with
 565  * gst_message_unref() after usage.
 566  *
 567  * MT safe.
 568  *
 569  * Since: 0.10.12
 570  */
 571 GstMessage *
 572 gst_bus_timed_pop (GstBus * bus, GstClockTime timeout)
 573 {
 574   g_return_val_if_fail (GST_IS_BUS (bus), NULL);
 575
 576   return gst_bus_timed_pop_filtered (bus, timeout, GST_MESSAGE_ANY);
 577 }
 578
 579 /**
 580  * gst_bus_pop_filtered:
 581  * @bus: a #GstBus to pop
 582  * @types: message types to take into account
 583  *
 584  * Get a message matching @type from the bus.  Will discard all messages on
 585  * the bus that do not match @type and that have been posted before the first
 586  * message that does match @type.  If there is no message matching @type on
 587  * the bus, all messages will be discarded.
 588  *
 589  * Returns: The next #GstMessage matching @type that is on the bus, or NULL if
 590  *     the bus is empty or there is no message matching @type.
 591  * The message is taken from the bus and needs to be unreffed with
 592  * gst_message_unref() after usage.
 593  *
 594  * MT safe.
 595  *
 596  * Since: 0.10.15
 597  */
 598 GstMessage *
 599 gst_bus_pop_filtered (GstBus * bus, GstMessageType types)
 600 {
 601   g_return_val_if_fail (GST_IS_BUS (bus), NULL);
 602   g_return_val_if_fail (types != 0, NULL);
 603
 604   return gst_bus_timed_pop_filtered (bus, 0, types);
 605 }
 606
 607 /**
 608  * gst_bus_pop:
 609  * @bus: a #GstBus to pop
 610  *
 611  * Get a message from the bus.
 612  *
 613  * Returns: The #GstMessage that is on the bus, or NULL if the bus is empty.
 614  * The message is taken from the bus and needs to be unreffed with
 615  * gst_message_unref() after usage.
 616  *
 617  * MT safe.
 618  */
 619 GstMessage *
 620 gst_bus_pop (GstBus * bus)
 621 {
 622   g_return_val_if_fail (GST_IS_BUS (bus), NULL);
 623
 624   return gst_bus_timed_pop_filtered (bus, 0, GST_MESSAGE_ANY);
 625 }
 626
 627 /**
 628  * gst_bus_peek:
 629  * @bus: a #GstBus
 630  *
 631  * Peek the message on the top of the bus' queue. The message will remain
 632  * on the bus' message queue. A reference is returned, and needs to be unreffed
 633  * by the caller.
 634  *
 635  * Returns: The #GstMessage that is on the bus, or NULL if the bus is empty.
 636  *
 637  * MT safe.
 638  */
 639 GstMessage *
 640 gst_bus_peek (GstBus * bus)
 641 {
 642   GstMessage *message;
 643
 644   g_return_val_if_fail (GST_IS_BUS (bus), NULL);
 645
 646   g_mutex_lock (bus->queue_lock);
 647   message = g_queue_peek_head (bus->queue);
 648   if (message)
 649     gst_message_ref (message);
 650   g_mutex_unlock (bus->queue_lock);
 651
 652   GST_DEBUG_OBJECT (bus, "peek on bus, got message %p", message);
 653
 654   return message;
 655 }
 656
 657 /**
 658  * gst_bus_set_sync_handler:
 659  * @bus: a #GstBus to install the handler on
 660  * @func: The handler function to install
 661  * @data: User data that will be sent to the handler function.
 662  *
 663  * Sets the synchronous handler on the bus. The function will be called
 664  * every time a new message is posted on the bus. Note that the function
 665  * will be called in the same thread context as the posting object. This
 666  * function is usually only called by the creator of the bus. Applications
 667  * should handle messages asynchronously using the gst_bus watch and poll
 668  * functions.
 669  *
 670  * You cannot replace an existing sync_handler. You can pass NULL to this
 671  * function, which will clear the existing handler.
 672  */
 673 void
 674 gst_bus_set_sync_handler (GstBus * bus, GstBusSyncHandler func, gpointer data)
 675 {
 676   g_return_if_fail (GST_IS_BUS (bus));
 677
 678   GST_OBJECT_LOCK (bus);
 679
 680   /* Assert if the user attempts to replace an existing sync_handler,
 681    * other than to clear it */
 682   if (func != NULL && bus->sync_handler != NULL)
 683     goto no_replace;
 684
 685   bus->sync_handler = func;
 686   bus->sync_handler_data = data;
 687   GST_OBJECT_UNLOCK (bus);
 688
 689   return;
 690
 691 no_replace:
 692   {
 693     GST_OBJECT_UNLOCK (bus);
 694     g_warning ("cannot replace existing sync handler");
 695     return;
 696   }
 697 }
 698
 699 /* GSource for the bus
 700  */
 701 typedef struct
 702 {
 703   GSource source;
 704   GstBus *bus;
 705   gboolean inited;
 706 } GstBusSource;
 707
 708 static gboolean
 709 gst_bus_source_prepare (GSource * source, gint * timeout)
 710 {
 711   GstBusSource *bsrc = (GstBusSource *) source;
 712
 713   /* we do this here now that we know that we're attached to a main context
 714    * (we don't support detaching a source from a main context and then
 715    * re-attaching it to a different main context) */
 716   if (G_UNLIKELY (!bsrc->inited)) {
 717     gst_bus_set_main_context (bsrc->bus, g_source_get_context (source));
 718     bsrc->inited = TRUE;
 719   }
 720
 721   *timeout = -1;
 722   return gst_bus_have_pending (bsrc->bus);
 723 }
 724
 725 static gboolean
 726 gst_bus_source_check (GSource * source)
 727 {
 728   GstBusSource *bsrc = (GstBusSource *) source;
 729
 730   return gst_bus_have_pending (bsrc->bus);
 731 }
 732
 733 static gboolean
 734 gst_bus_source_dispatch (GSource * source, GSourceFunc callback,
 735     gpointer user_data)
 736 {
 737   GstBusFunc handler = (GstBusFunc) callback;
 738   GstBusSource *bsource = (GstBusSource *) source;
 739   GstMessage *message;
 740   gboolean keep;
 741   GstBus *bus;
 742
 743   g_return_val_if_fail (bsource != NULL, FALSE);
 744
 745   bus = bsource->bus;
 746
 747   g_return_val_if_fail (GST_IS_BUS (bus), FALSE);
 748
 749   message = gst_bus_pop (bus);
 750
 751   /* The message queue might be empty if some other thread or callback set
 752    * the bus to flushing between check/prepare and dispatch */
 753   if (G_UNLIKELY (message == NULL))
 754     return TRUE;
 755
 756   if (!handler)
 757     goto no_handler;
 758
 759   GST_DEBUG_OBJECT (bus, "source %p calling dispatch with %p", source, message);
 760
 761   keep = handler (bus, message, user_data);
 762   gst_message_unref (message);
 763
 764   GST_DEBUG_OBJECT (bus, "source %p handler returns %d", source, keep);
 765
 766   return keep;
 767
 768 no_handler:
 769   {
 770     g_warning ("GstBus watch dispatched without callback\n"
 771         "You must call g_source_set_callback().");
 772     gst_message_unref (message);
 773     return FALSE;
 774   }
 775 }
 776
 777 static void
 778 gst_bus_source_finalize (GSource * source)
 779 {
 780   GstBusSource *bsource = (GstBusSource *) source;
 781   GstBus *bus;
 782
 783   bus = bsource->bus;
 784
 785   GST_DEBUG_OBJECT (bus, "finalize source %p", source);
 786
 787   GST_OBJECT_LOCK (bus);
 788   if (bus->priv->watch_id == source)
 789     bus->priv->watch_id = NULL;
 790   GST_OBJECT_UNLOCK (bus);
 791
 792   gst_bus_set_main_context (bsource->bus, NULL);
 793   gst_object_unref (bsource->bus);
 794   bsource->bus = NULL;
 795 }
 796
 797 static GSourceFuncs gst_bus_source_funcs = {
 798   gst_bus_source_prepare,
 799   gst_bus_source_check,
 800   gst_bus_source_dispatch,
 801   gst_bus_source_finalize
 802 };
 803
 804 /**
 805  * gst_bus_create_watch:
 806  * @bus: a #GstBus to create the watch for
 807  *
 808  * Create watch for this bus. The GSource will be dispatched whenever
 809  * a message is on the bus. After the GSource is dispatched, the
 810  * message is popped off the bus and unreffed.
 811  *
 812  * Returns: A #GSource that can be added to a mainloop.
 813  */
 814 GSource *
 815 gst_bus_create_watch (GstBus * bus)
 816 {
 817   GstBusSource *source;
 818
 819   g_return_val_if_fail (GST_IS_BUS (bus), NULL);
 820
 821   source = (GstBusSource *) g_source_new (&gst_bus_source_funcs,
 822       sizeof (GstBusSource));
 823   source->bus = gst_object_ref (bus);
 824   source->inited = FALSE;
 825
 826   return (GSource *) source;
 827 }
 828
 829 /* must be called with the bus OBJECT LOCK */
 830 static guint
 831 gst_bus_add_watch_full_unlocked (GstBus * bus, gint priority,
 832     GstBusFunc func, gpointer user_data, GDestroyNotify notify)
 833 {
 834   guint id;
 835   GSource *source;
 836
 837   if (bus->priv->watch_id) {
 838     GST_ERROR_OBJECT (bus,
 839         "Tried to add new watch while one was already there");
 840     return 0;
 841   }
 842
 843   source = gst_bus_create_watch (bus);
 844
 845   if (priority != G_PRIORITY_DEFAULT)
 846     g_source_set_priority (source, priority);
 847
 848   g_source_set_callback (source, (GSourceFunc) func, user_data, notify);
 849
 850   id = g_source_attach (source, NULL);
 851   g_source_unref (source);
 852
 853   if (id) {
 854     bus->priv->watch_id = source;
 855   }
 856
 857   GST_DEBUG_OBJECT (bus, "New source %p with id %u", source, id);
 858   return id;
 859 }
 860
 861 /**
 862  * gst_bus_add_watch_full:
 863  * @bus: a #GstBus to create the watch for.
 864  * @priority: The priority of the watch.
 865  * @func: A function to call when a message is received.
 866  * @user_data: user data passed to @func.
 867  * @notify: the function to call when the source is removed.
 868  *
 869  * Adds a bus watch to the default main context with the given @priority.
 870  * This function is used to receive asynchronous messages in the main loop.
 871  * There can only be a single bus watch per bus, you must remove it before you
 872  * can set a new one.
 873  *
 874  * When @func is called, the message belongs to the caller; if you want to
 875  * keep a copy of it, call gst_message_ref() before leaving @func.
 876  *
 877  * The watch can be removed using g_source_remove() or by returning FALSE
 878  * from @func.
 879  *
 880  * Returns: The event source id.
 881  *
 882  * MT safe.
 883  */
 884 guint
 885 gst_bus_add_watch_full (GstBus * bus, gint priority,
 886     GstBusFunc func, gpointer user_data, GDestroyNotify notify)
 887 {
 888   guint id;
 889
 890   g_return_val_if_fail (GST_IS_BUS (bus), 0);
 891
 892   GST_OBJECT_LOCK (bus);
 893   id = gst_bus_add_watch_full_unlocked (bus, priority, func, user_data, notify);
 894   GST_OBJECT_UNLOCK (bus);
 895
 896   return id;
 897 }
 898
 899 /**
 900  * gst_bus_add_watch:
 901  * @bus: a #GstBus to create the watch for
 902  * @func: A function to call when a message is received.
 903  * @user_data: user data passed to @func.
 904  *
 905  * Adds a bus watch to the default main context with the default priority.
 906  * This function is used to receive asynchronous messages in the main loop.
 907  * There can only be a single bus watch per bus, you must remove it before you
 908  * can set a new one.
 909  *
 910  * The watch can be removed using g_source_remove() or by returning FALSE
 911  * from @func.
 912  *
 913  * Returns: The event source id.
 914  *
 915  * MT safe.
 916  */
 917 guint
 918 gst_bus_add_watch (GstBus * bus, GstBusFunc func, gpointer user_data)
 919 {
 920   return gst_bus_add_watch_full (bus, G_PRIORITY_DEFAULT, func,
 921       user_data, NULL);
 922 }
 923
 924 typedef struct
 925 {
 926   GMainLoop *loop;
 927   guint timeout_id;
 928   gboolean source_running;
 929   GstMessageType events;
 930   GstMessage *message;
 931 } GstBusPollData;
 932
 933 static void
 934 poll_func (GstBus * bus, GstMessage * message, GstBusPollData * poll_data)
 935 {
 936   GstMessageType type;
 937
 938   if (!g_main_loop_is_running (poll_data->loop)) {
 939     GST_DEBUG ("mainloop %p not running", poll_data->loop);
 940     return;
 941   }
 942
 943   type = GST_MESSAGE_TYPE (message);
 944
 945   if (type & poll_data->events) {
 946     g_return_if_fail (poll_data->message == NULL);
 947     /* keep ref to message */
 948     poll_data->message = gst_message_ref (message);
 949     GST_DEBUG ("mainloop %p quit", poll_data->loop);
 950     g_main_loop_quit (poll_data->loop);
 951   } else {
 952     GST_DEBUG ("type %08x does not match %08x", type, poll_data->events);
 953   }
 954 }
 955
 956 static gboolean
 957 poll_timeout (GstBusPollData * poll_data)
 958 {
 959   GST_DEBUG ("mainloop %p quit", poll_data->loop);
 960   g_main_loop_quit (poll_data->loop);
 961
 962   /* we don't remove the GSource as this would free our poll_data,
 963    * which we still need */
 964   return TRUE;
 965 }
 966
 967 static void
 968 poll_destroy (GstBusPollData * poll_data, gpointer unused)
 969 {
 970   poll_data->source_running = FALSE;
 971   if (!poll_data->timeout_id) {
 972     g_main_loop_unref (poll_data->loop);
 973     g_free (poll_data);
 974   }
 975 }
 976
 977 static void
 978 poll_destroy_timeout (GstBusPollData * poll_data)
 979 {
 980   poll_data->timeout_id = 0;
 981   if (!poll_data->source_running) {
 982     g_main_loop_unref (poll_data->loop);
 983     g_free (poll_data);
 984   }
 985 }
 986
 987 /**
 988  * gst_bus_poll:
 989  * @bus: a #GstBus
 990  * @events: a mask of #GstMessageType, representing the set of message types to
 991  * poll for.
 992  * @timeout: the poll timeout, as a #GstClockTimeDiff, or -1 to poll
 993  * indefinitely.
 994  *
 995  * Poll the bus for messages. Will block while waiting for messages to come.
 996  * You can specify a maximum time to poll with the @timeout parameter. If
 997  * @timeout is negative, this function will block indefinitely.
 998  *
 999  * All messages not in @events will be popped off the bus and will be ignored.
1000  *
1001  * Because poll is implemented using the "message" signal enabled by
1002  * gst_bus_add_signal_watch(), calling gst_bus_poll() will cause the "message"
1003  * signal to be emitted for every message that poll sees. Thus a "message"
1004  * signal handler will see the same messages that this function sees -- neither
1005  * will steal messages from the other.
1006  *
1007  * This function will run a main loop from the default main context when
1008  * polling.
1009  *
1010  * You should never use this function, since it is pure evil. This is
1011  * especially true for GUI applications based on Gtk+ or Qt, but also for any
1012  * other non-trivial application that uses the GLib main loop. As this function
1013  * runs a GLib main loop, any callback attached to the default GLib main
1014  * context may be invoked. This could be timeouts, GUI events, I/O events etc.;
1015  * even if gst_bus_poll() is called with a 0 timeout. Any of these callbacks
1016  * may do things you do not expect, e.g. destroy the main application window or
1017  * some other resource; change other application state; display a dialog and
1018  * run another main loop until the user clicks it away. In short, using this
1019  * function may add a lot of complexity to your code through unexpected
1020  * re-entrancy and unexpected changes to your application's state.
1021  *
1022  * For 0 timeouts use gst_bus_pop_filtered() instead of this function; for
1023  * other short timeouts use gst_bus_timed_pop_filtered(); everything else is
1024  * better handled by setting up an asynchronous bus watch and doing things
1025  * from there.
1026  *
1027  * Returns: The message that was received, or NULL if the poll timed out.
1028  * The message is taken from the bus and needs to be unreffed with
1029  * gst_message_unref() after usage.
1030  */
1031 GstMessage *
1032 gst_bus_poll (GstBus * bus, GstMessageType events, GstClockTimeDiff timeout)
1033 {
1034   GstBusPollData *poll_data;
1035   GstMessage *ret;
1036   gulong id;
1037
1038   poll_data = g_new0 (GstBusPollData, 1);
1039   poll_data->source_running = TRUE;
1040   poll_data->loop = g_main_loop_new (NULL, FALSE);
1041   poll_data->events = events;
1042   poll_data->message = NULL;
1043
1044   if (timeout >= 0)
1045     poll_data->timeout_id = g_timeout_add_full (G_PRIORITY_DEFAULT_IDLE,
1046         timeout / GST_MSECOND, (GSourceFunc) poll_timeout, poll_data,
1047         (GDestroyNotify) poll_destroy_timeout);
1048   else
1049     poll_data->timeout_id = 0;
1050
1051   id = g_signal_connect_data (bus, "message", G_CALLBACK (poll_func), poll_data,
1052       (GClosureNotify) poll_destroy, 0);
1053
1054   /* these can be nested, so it's ok */
1055   gst_bus_add_signal_watch (bus);
1056
1057   GST_DEBUG ("running mainloop %p", poll_data->loop);
1058   g_main_loop_run (poll_data->loop);
1059   GST_DEBUG ("mainloop stopped %p", poll_data->loop);
1060
1061   gst_bus_remove_signal_watch (bus);
1062
1063   /* holds a ref */
1064   ret = poll_data->message;
1065
1066   if (poll_data->timeout_id)
1067     g_source_remove (poll_data->timeout_id);
1068
1069   /* poll_data will be freed now */
1070   g_signal_handler_disconnect (bus, id);
1071
1072   GST_DEBUG_OBJECT (bus, "finished poll with message %p", ret);
1073
1074   return ret;
1075 }
1076
1077 /**
1078  * gst_bus_async_signal_func:
1079  * @bus: a #GstBus
1080  * @message: the #GstMessage received
1081  * @data: user data
1082  *
1083  * A helper #GstBusFunc that can be used to convert all asynchronous messages
1084  * into signals.
1085  *
1086  * Returns: TRUE
1087  */
1088 gboolean
1089 gst_bus_async_signal_func (GstBus * bus, GstMessage * message, gpointer data)
1090 {
1091   GQuark detail = 0;
1092
1093   g_return_val_if_fail (GST_IS_BUS (bus), TRUE);
1094   g_return_val_if_fail (message != NULL, TRUE);
1095
1096   detail = gst_message_type_to_quark (GST_MESSAGE_TYPE (message));
1097
1098   g_signal_emit (bus, gst_bus_signals[ASYNC_MESSAGE], detail, message);
1099
1100   /* we never remove this source based on signal emission return values */
1101   return TRUE;
1102 }
1103
1104 /**
1105  * gst_bus_sync_signal_handler:
1106  * @bus: a #GstBus
1107  * @message: the #GstMessage received
1108  * @data: user data
1109  *
1110  * A helper GstBusSyncHandler that can be used to convert all synchronous
1111  * messages into signals.
1112  *
1113  * Returns: GST_BUS_PASS
1114  */
1115 GstBusSyncReply
1116 gst_bus_sync_signal_handler (GstBus * bus, GstMessage * message, gpointer data)
1117 {
1118   GQuark detail = 0;
1119
1120   g_return_val_if_fail (GST_IS_BUS (bus), GST_BUS_DROP);
1121   g_return_val_if_fail (message != NULL, GST_BUS_DROP);
1122
1123   detail = gst_message_type_to_quark (GST_MESSAGE_TYPE (message));
1124
1125   g_signal_emit (bus, gst_bus_signals[SYNC_MESSAGE], detail, message);
1126
1127   return GST_BUS_PASS;
1128 }
1129
1130 /**
1131  * gst_bus_enable_sync_message_emission:
1132  * @bus: a #GstBus on which you want to receive the "sync-message" signal
1133  *
1134  * Instructs GStreamer to emit the "sync-message" signal after running the bus's
1135  * sync handler. This function is here so that code can ensure that they can
1136  * synchronously receive messages without having to affect what the bin's sync
1137  * handler is.
1138  *
1139  * This function may be called multiple times. To clean up, the caller is
1140  * responsible for calling gst_bus_disable_sync_message_emission() as many times
1141  * as this function is called.
1142  *
1143  * While this function looks similar to gst_bus_add_signal_watch(), it is not
1144  * exactly the same -- this function enables <emphasis>synchronous</emphasis> emission of
1145  * signals when messages arrive; gst_bus_add_signal_watch() adds an idle callback
1146  * to pop messages off the bus <emphasis>asynchronously</emphasis>. The sync-message signal
1147  * comes from the thread of whatever object posted the message; the "message"
1148  * signal is marshalled to the main thread via the main loop.
1149  *
1150  * MT safe.
1151  */
1152 void
1153 gst_bus_enable_sync_message_emission (GstBus * bus)
1154 {
1155   g_return_if_fail (GST_IS_BUS (bus));
1156
1157   GST_OBJECT_LOCK (bus);
1158   bus->priv->num_sync_message_emitters++;
1159   GST_OBJECT_UNLOCK (bus);
1160 }
1161
1162 /**
1163  * gst_bus_disable_sync_message_emission:
1164  * @bus: a #GstBus on which you previously called
1165  * gst_bus_enable_sync_message_emission()
1166  *
1167  * Instructs GStreamer to stop emitting the "sync-message" signal for this bus.
1168  * See gst_bus_enable_sync_message_emission() for more information.
1169  *
1170  * In the event that multiple pieces of code have called
1171  * gst_bus_enable_sync_message_emission(), the sync-message emissions will only
1172  * be stopped after all calls to gst_bus_enable_sync_message_emission() were
1173  * "cancelled" by calling this function. In this way the semantics are exactly
1174  * the same as gst_object_ref() that which calls enable should also call
1175  * disable.
1176  *
1177  * MT safe.
1178  */
1179 void
1180 gst_bus_disable_sync_message_emission (GstBus * bus)
1181 {
1182   g_return_if_fail (GST_IS_BUS (bus));
1183   g_return_if_fail (bus->num_signal_watchers == 0);
1184
1185   GST_OBJECT_LOCK (bus);
1186   bus->priv->num_sync_message_emitters--;
1187   GST_OBJECT_UNLOCK (bus);
1188 }
1189
1190 /**
1191  * gst_bus_add_signal_watch_full:
1192  * @bus: a #GstBus on which you want to receive the "message" signal
1193  * @priority: The priority of the watch.
1194  *
1195  * Adds a bus signal watch to the default main context with the given priority.
1196  * After calling this statement, the bus will emit the "message" signal for each
1197  * message posted on the bus when the main loop is running.
1198  *
1199  * This function may be called multiple times. To clean up, the caller is
1200  * responsible for calling gst_bus_remove_signal_watch() as many times as this
1201  * function is called.
1202  *
1203  * There can only be a single bus watch per bus, you most remove all signal watch
1204  * before you can set another type of watch.
1205  *
1206  * MT safe.
1207  */
1208 void
1209 gst_bus_add_signal_watch_full (GstBus * bus, gint priority)
1210 {
1211   g_return_if_fail (GST_IS_BUS (bus));
1212
1213   /* I know the callees don't take this lock, so go ahead and abuse it */
1214   GST_OBJECT_LOCK (bus);
1215
1216   if (bus->num_signal_watchers > 0)
1217     goto done;
1218
1219   /* this should not fail because the counter above takes care of it */
1220   g_assert (bus->signal_watch_id == 0);
1221
1222   bus->signal_watch_id =
1223       gst_bus_add_watch_full_unlocked (bus, priority, gst_bus_async_signal_func,
1224       NULL, NULL);
1225
1226   if (G_UNLIKELY (bus->signal_watch_id == 0))
1227     goto add_failed;
1228
1229 done:
1230
1231   bus->num_signal_watchers++;
1232
1233   GST_OBJECT_UNLOCK (bus);
1234   return;
1235
1236   /* ERRORS */
1237 add_failed:
1238   {
1239     g_critical ("Could not add signal watch to bus %s", GST_OBJECT_NAME (bus));
1240     GST_OBJECT_UNLOCK (bus);
1241     return;
1242   }
1243 }
1244
1245 /**
1246  * gst_bus_add_signal_watch:
1247  * @bus: a #GstBus on which you want to receive the "message" signal
1248  *
1249  * Adds a bus signal watch to the default main context with the default
1250  * priority.
1251  * After calling this statement, the bus will emit the "message" signal for each
1252  * message posted on the bus.
1253  *
1254  * This function may be called multiple times. To clean up, the caller is
1255  * responsible for calling gst_bus_remove_signal_watch() as many times as this
1256  * function is called.
1257  *
1258  * MT safe.
1259  */
1260 void
1261 gst_bus_add_signal_watch (GstBus * bus)
1262 {
1263   gst_bus_add_signal_watch_full (bus, G_PRIORITY_DEFAULT);
1264 }
1265
1266 /**
1267  * gst_bus_remove_signal_watch:
1268  * @bus: a #GstBus you previously added a signal watch to
1269  *
1270  * Removes a signal watch previously added with gst_bus_add_signal_watch().
1271  *
1272  * MT safe.
1273  */
1274 void
1275 gst_bus_remove_signal_watch (GstBus * bus)
1276 {
1277   guint id = 0;
1278
1279   g_return_if_fail (GST_IS_BUS (bus));
1280
1281   /* I know the callees don't take this lock, so go ahead and abuse it */
1282   GST_OBJECT_LOCK (bus);
1283
1284   if (bus->num_signal_watchers == 0)
1285     goto error;
1286
1287   bus->num_signal_watchers--;
1288
1289   if (bus->num_signal_watchers > 0)
1290     goto done;
1291
1292   id = bus->signal_watch_id;
1293   bus->signal_watch_id = 0;
1294
1295   GST_DEBUG_OBJECT (bus, "removing signal watch %u", id);
1296
1297 done:
1298   GST_OBJECT_UNLOCK (bus);
1299
1300   if (id)
1301     g_source_remove (id);
1302
1303   return;
1304
1305   /* ERRORS */
1306 error:
1307   {
1308     g_critical ("Bus %s has no signal watches attached", GST_OBJECT_NAME (bus));
1309     GST_OBJECT_UNLOCK (bus);
1310     return;
1311   }
1312 }