/* GStreamer

 * Copyright (C) 2004 Wim Taymans <wim@fluendo.com>

 *

 * gstbus.c: GstBus subsystem

 *

 * This library is free software; you can redistribute it and/or

 * modify it under the terms of the GNU Library General Public

 * License as published by the Free Software Foundation; either

 * version 2 of the License, or (at your option) any later version.

 *

 * This library is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

 * Library General Public License for more details.

 *

 * You should have received a copy of the GNU Library General Public

 * License along with this library; if not, write to the

 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,

 * Boston, MA 02110-1301, USA.

 */

/**

 * SECTION:gstbus

 * @title: GstBus

 * @short_description: Asynchronous message bus subsystem

 * @see_also: #GstMessage, #GstElement

 *

 * The #GstBus is an object responsible for delivering #GstMessage packets in

 * a first-in first-out way from the streaming threads (see #GstTask) to the

 * application.

 *

 * Since the application typically only wants to deal with delivery of these

 * messages from one thread, the GstBus will marshall the messages between

 * different threads. This is important since the actual streaming of media

 * is done in another thread than the application.

 *

 * The GstBus provides support for #GSource based notifications. This makes it

 * possible to handle the delivery in the glib mainloop.

 *

 * The #GSource callback function gst_bus_async_signal_func() can be used to

 * convert all bus messages into signal emissions.

 *

 * A message is posted on the bus with the gst_bus_post() method. With the

 * gst_bus_peek() and gst_bus_pop() methods one can look at or retrieve a

 * previously posted message.

 *

 * The bus can be polled with the gst_bus_poll() method. This methods blocks

 * up to the specified timeout value until one of the specified messages types

 * is posted on the bus. The application can then gst_bus_pop() the messages

 * from the bus to handle them.

 * Alternatively the application can register an asynchronous bus function

 * using gst_bus_add_watch_full() or gst_bus_add_watch(). This function will

 * install a #GSource in the default glib main loop and will deliver messages

 * a short while after they have been posted. Note that the main loop should

 * be running for the asynchronous callbacks.

 *

 * It is also possible to get messages from the bus without any thread

 * marshalling with the gst_bus_set_sync_handler() method. This makes it

 * possible to react to a message in the same thread that posted the

 * message on the bus. This should only be used if the application is able

 * to deal with messages from different threads.

 *

 * Every #GstPipeline has one bus.

 *

 * Note that a #GstPipeline will set its bus into flushing state when changing

 * from READY to NULL state.

 */

#include "gst_private.h"

#include <errno.h>

#ifdef HAVE_UNISTD_H

#  include <unistd.h>

#endif

#include <sys/types.h>

#include "gstatomicqueue.h"

#include "gstinfo.h"

#include "gstpoll.h"

#include "gstbus.h"

#include "glib-compat-private.h"

#ifdef G_OS_WIN32

#  ifndef EWOULDBLOCK

#  define EWOULDBLOCK EAGAIN    /* This is just to placate gcc */

#  endif

#endif /* G_OS_WIN32 */

#define GST_CAT_DEFAULT GST_CAT_BUS

/* bus signals */

enum

{

  SYNC_MESSAGE,

  ASYNC_MESSAGE,

  /* add more above */

  LAST_SIGNAL

};

#define DEFAULT_ENABLE_ASYNC (TRUE)

enum

{

  PROP_0,

  PROP_ENABLE_ASYNC

};

static void gst_bus_dispose (GObject * object);

static void gst_bus_finalize (GObject * object);

static guint gst_bus_signals[LAST_SIGNAL] = { 0 };

struct _GstBusPrivate

{

  GstAtomicQueue *queue;

  GMutex queue_lock;

  GstBusSyncHandler sync_handler;

  gpointer sync_handler_data;

  GDestroyNotify sync_handler_notify;

  guint num_signal_watchers;

  guint num_sync_message_emitters;

  GSource *signal_watch;

  gboolean enable_async;

  GstPoll *poll;

  GPollFD pollfd;

};

#define gst_bus_parent_class parent_class

G_DEFINE_TYPE_WITH_PRIVATE (GstBus, gst_bus, GST_TYPE_OBJECT);

static void

gst_bus_set_property (GObject * object,

    guint prop_id, const GValue * value, GParamSpec * pspec)

{

  GstBus *bus = GST_BUS_CAST (object);

  switch (prop_id) {

    case PROP_ENABLE_ASYNC:

      bus->priv->enable_async = g_value_get_boolean (value);

      break;

    default:

      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);

      break;

  }

}

static void

gst_bus_constructed (GObject * object)

{

  GstBus *bus = GST_BUS_CAST (object);

  if (bus->priv->enable_async) {

    bus->priv->poll = gst_poll_new_timer ();

    gst_poll_get_read_gpollfd (bus->priv->poll, &bus->priv->pollfd);

  }

  G_OBJECT_CLASS (gst_bus_parent_class)->constructed (object);

}

static void

gst_bus_class_init (GstBusClass * klass)

{

  GObjectClass *gobject_class = (GObjectClass *) klass;

  gobject_class->dispose = gst_bus_dispose;

  gobject_class->finalize = gst_bus_finalize;

  gobject_class->set_property = gst_bus_set_property;

  gobject_class->constructed = gst_bus_constructed;

  /**

   * GstBus::enable-async:

   *

   * Enable async message delivery support for bus watches,

   * gst_bus_pop() and similar API. Without this only the

   * synchronous message handlers are called.

   *

   * This property is used to create the child element buses

   * in #GstBin.

   */

  g_object_class_install_property (gobject_class, PROP_ENABLE_ASYNC,

      g_param_spec_boolean ("enable-async", "Enable Async",

          "Enable async message delivery for bus watches and gst_bus_pop()",

          DEFAULT_ENABLE_ASYNC,

          G_PARAM_CONSTRUCT_ONLY | G_PARAM_WRITABLE | G_PARAM_STATIC_STRINGS));

  /**

   * GstBus::sync-message:

   * @bus: the object which received the signal

   * @message: the message that has been posted synchronously

   *

   * A message has been posted on the bus. This signal is emitted from the

   * thread that posted the message so one has to be careful with locking.

   *

   * This signal will not be emitted by default, you have to call

   * gst_bus_enable_sync_message_emission() before.

   */

  gst_bus_signals[SYNC_MESSAGE] =

      g_signal_new ("sync-message", G_TYPE_FROM_CLASS (klass),

      G_SIGNAL_RUN_LAST | G_SIGNAL_DETAILED,

      G_STRUCT_OFFSET (GstBusClass, sync_message), NULL, NULL,

      g_cclosure_marshal_generic, G_TYPE_NONE, 1, GST_TYPE_MESSAGE);

  /**

   * GstBus::message:

   * @bus: the object which received the signal

   * @message: the message that has been posted asynchronously

   *

   * A message has been posted on the bus. This signal is emitted from a

   * GSource added to the mainloop. this signal will only be emitted when

   * there is a mainloop running.

   */

  gst_bus_signals[ASYNC_MESSAGE] =

      g_signal_new ("message", G_TYPE_FROM_CLASS (klass),

      G_SIGNAL_RUN_LAST | G_SIGNAL_DETAILED,

      G_STRUCT_OFFSET (GstBusClass, message), NULL, NULL,

      g_cclosure_marshal_generic, G_TYPE_NONE, 1, GST_TYPE_MESSAGE);

}

static void

gst_bus_init (GstBus * bus)

{

  bus->priv = gst_bus_get_instance_private (bus);

  bus->priv->enable_async = DEFAULT_ENABLE_ASYNC;

  g_mutex_init (&bus->priv->queue_lock);

  bus->priv->queue = gst_atomic_queue_new (32);

  GST_DEBUG_OBJECT (bus, "created");

}

static void

gst_bus_dispose (GObject * object)

{

  GstBus *bus = GST_BUS (object);

  if (bus->priv->queue) {

    GstMessage *message;

    g_mutex_lock (&bus->priv->queue_lock);

    do {

      message = gst_atomic_queue_pop (bus->priv->queue);

      if (message)

        gst_message_unref (message);

    } while (message != NULL);

    gst_atomic_queue_unref (bus->priv->queue);

    bus->priv->queue = NULL;

    g_mutex_unlock (&bus->priv->queue_lock);

    g_mutex_clear (&bus->priv->queue_lock);

    if (bus->priv->poll)

      gst_poll_free (bus->priv->poll);

    bus->priv->poll = NULL;

  }

  G_OBJECT_CLASS (parent_class)->dispose (object);

}

static void

gst_bus_finalize (GObject * object)

{

  GstBus *bus = GST_BUS (object);

  if (bus->priv->sync_handler_notify)

    bus->priv->sync_handler_notify (bus->priv->sync_handler_data);

  G_OBJECT_CLASS (parent_class)->finalize (object);

}

/**

 * gst_bus_new:

 *

 * Creates a new #GstBus instance.

 *

 * Returns: (transfer full): a new #GstBus instance

 */

GstBus *

gst_bus_new (void)

{

  GstBus *result;

  result = g_object_new (gst_bus_get_type (), NULL);

  GST_DEBUG_OBJECT (result, "created new bus");

  /* clear floating flag */

  gst_object_ref_sink (result);

  return result;

}

/**

 * gst_bus_post:

 * @bus: a #GstBus to post on

 * @message: (transfer full): the #GstMessage to post

 *

 * Post a message on the given bus. Ownership of the message

 * is taken by the bus.

 *

 * Returns: %TRUE if the message could be posted, %FALSE if the bus is flushing.

 *

 * MT safe.

 */

gboolean

gst_bus_post (GstBus * bus, GstMessage * message)

{

  GstBusSyncReply reply = GST_BUS_PASS;

  GstBusSyncHandler handler;

  gboolean emit_sync_message;

  gpointer handler_data;

  g_return_val_if_fail (GST_IS_BUS (bus), FALSE);

  g_return_val_if_fail (GST_IS_MESSAGE (message), FALSE);

  GST_DEBUG_OBJECT (bus, "[msg %p] posting on bus %" GST_PTR_FORMAT, message,

      message);

  /* check we didn't accidentally add a public flag that maps to same value */

  g_assert (!GST_MINI_OBJECT_FLAG_IS_SET (message,

          GST_MESSAGE_FLAG_ASYNC_DELIVERY));

  GST_OBJECT_LOCK (bus);

  /* check if the bus is flushing */

  if (GST_OBJECT_FLAG_IS_SET (bus, GST_BUS_FLUSHING))

    goto is_flushing;

  handler = bus->priv->sync_handler;

  handler_data = bus->priv->sync_handler_data;

  emit_sync_message = bus->priv->num_sync_message_emitters > 0;

  GST_OBJECT_UNLOCK (bus);

  /* first call the sync handler if it is installed */

  if (handler)

    reply = handler (bus, message, handler_data);

  /* emit sync-message if requested to do so via

     gst_bus_enable_sync_message_emission. terrible but effective */

  if (emit_sync_message && reply != GST_BUS_DROP

      && handler != gst_bus_sync_signal_handler)

    gst_bus_sync_signal_handler (bus, message, NULL);

  /* If this is a bus without async message delivery

   * always drop the message */

  if (!bus->priv->poll)

    reply = GST_BUS_DROP;

  /* now see what we should do with the message */

  switch (reply) {

    case GST_BUS_DROP:

      /* drop the message */

      GST_DEBUG_OBJECT (bus, "[msg %p] dropped", message);

      break;

    case GST_BUS_PASS:

      /* pass the message to the async queue, refcount passed in the queue */

      GST_DEBUG_OBJECT (bus, "[msg %p] pushing on async queue", message);

      gst_atomic_queue_push (bus->priv->queue, message);

      gst_poll_write_control (bus->priv->poll);

      GST_DEBUG_OBJECT (bus, "[msg %p] pushed on async queue", message);

      break;

    case GST_BUS_ASYNC:

    {

      /* async delivery, we need a mutex and a cond to block

       * on */

      GCond *cond = GST_MESSAGE_GET_COND (message);

      GMutex *lock = GST_MESSAGE_GET_LOCK (message);

      g_cond_init (cond);

      g_mutex_init (lock);

      GST_MINI_OBJECT_FLAG_SET (message, GST_MESSAGE_FLAG_ASYNC_DELIVERY);

      GST_DEBUG_OBJECT (bus, "[msg %p] waiting for async delivery", message);

      /* now we lock the message mutex, send the message to the async

       * queue. When the message is handled by the app and destroyed,

       * the cond will be signalled and we can continue */

      g_mutex_lock (lock);

      gst_atomic_queue_push (bus->priv->queue, message);

      gst_poll_write_control (bus->priv->poll);

      /* now block till the message is freed */

      g_cond_wait (cond, lock);

      /* we acquired a new ref from gst_message_dispose() so we can clean up */

      g_mutex_unlock (lock);

      GST_DEBUG_OBJECT (bus, "[msg %p] delivered asynchronously", message);

      GST_MINI_OBJECT_FLAG_UNSET (message, GST_MESSAGE_FLAG_ASYNC_DELIVERY);

      g_mutex_clear (lock);

      g_cond_clear (cond);

      gst_message_unref (message);

      break;

    }

    default:

      g_warning ("invalid return from bus sync handler");

      break;

  }

  return TRUE;

  /* ERRORS */

is_flushing:

  {

    GST_DEBUG_OBJECT (bus, "bus is flushing");

    GST_OBJECT_UNLOCK (bus);

    gst_message_unref (message);

    return FALSE;

  }

}

/**

 * gst_bus_have_pending:

 * @bus: a #GstBus to check

 *

 * Check if there are pending messages on the bus that

 * should be handled.

 *

 * Returns: %TRUE if there are messages on the bus to be handled, %FALSE

 * otherwise.

 *

 * MT safe.

 */

gboolean

gst_bus_have_pending (GstBus * bus)

{

  gboolean result;

  g_return_val_if_fail (GST_IS_BUS (bus), FALSE);

  /* see if there is a message on the bus */

  result = gst_atomic_queue_length (bus->priv->queue) != 0;

  return result;

}

/**

 * gst_bus_set_flushing:

 * @bus: a #GstBus

 * @flushing: whether or not to flush the bus

 *

 * If @flushing, flush out and unref any messages queued in the bus. Releases

 * references to the message origin objects. Will flush future messages until

 * gst_bus_set_flushing() sets @flushing to %FALSE.

 *

 * MT safe.

 */

void

gst_bus_set_flushing (GstBus * bus, gboolean flushing)

{

  GstMessage *message;

  GList *message_list = NULL;

  g_return_if_fail (GST_IS_BUS (bus));

  GST_OBJECT_LOCK (bus);

  if (flushing) {

    GST_OBJECT_FLAG_SET (bus, GST_BUS_FLUSHING);

    GST_DEBUG_OBJECT (bus, "set bus flushing");

    while ((message = gst_bus_pop (bus)))

      message_list = g_list_prepend (message_list, message);

  } else {

    GST_DEBUG_OBJECT (bus, "unset bus flushing");

    GST_OBJECT_FLAG_UNSET (bus, GST_BUS_FLUSHING);

  }

  GST_OBJECT_UNLOCK (bus);

  g_list_free_full (message_list, (GDestroyNotify) gst_message_unref);

}

/**

 * gst_bus_timed_pop_filtered:

 * @bus: a #GstBus to pop from

 * @timeout: a timeout in nanoseconds, or GST_CLOCK_TIME_NONE to wait forever

 * @types: message types to take into account, GST_MESSAGE_ANY for any type

 *

 * Get a message from the bus whose type matches the message type mask @types,

 * waiting up to the specified timeout (and discarding any messages that do not

 * match the mask provided).

 *

 * If @timeout is 0, this function behaves like gst_bus_pop_filtered(). If

 * @timeout is #GST_CLOCK_TIME_NONE, this function will block forever until a

 * matching message was posted on the bus.

 *

 * Returns: (transfer full) (nullable): a #GstMessage matching the

 *     filter in @types, or %NULL if no matching message was found on

 *     the bus until the timeout expired. The message is taken from

 *     the bus and needs to be unreffed with gst_message_unref() after

 *     usage.

 *

 * MT safe.

 */

GstMessage *

gst_bus_timed_pop_filtered (GstBus * bus, GstClockTime timeout,

    GstMessageType types)

{

  GstMessage *message;

  GTimeVal now, then;

  gboolean first_round = TRUE;

  GstClockTime elapsed = 0;

  g_return_val_if_fail (GST_IS_BUS (bus), NULL);

  g_return_val_if_fail (types != 0, NULL);

  g_return_val_if_fail (timeout == 0 || bus->priv->poll != NULL, NULL);

  g_mutex_lock (&bus->priv->queue_lock);

  while (TRUE) {

    gint ret;

    GST_LOG_OBJECT (bus, "have %d messages",

        gst_atomic_queue_length (bus->priv->queue));

    while ((message = gst_atomic_queue_pop (bus->priv->queue))) {

      if (bus->priv->poll) {

        while (!gst_poll_read_control (bus->priv->poll)) {

          if (errno == EWOULDBLOCK) {

            /* Retry, this can happen if pushing to the queue has finished,

             * popping here succeeded but writing control did not finish

             * before we got to this line. */

            /* Give other threads the chance to do something */

            g_thread_yield ();

            continue;

          } else {

            /* This is a real error and means that either the bus is in an

             * inconsistent state, or the GstPoll is invalid. GstPoll already

             * prints a critical warning about this, no need to do that again

             * ourselves */

            break;

          }

        }

      }

      GST_DEBUG_OBJECT (bus, "got message %p, %s from %s, type mask is %u",

          message, GST_MESSAGE_TYPE_NAME (message),

          GST_MESSAGE_SRC_NAME (message), (guint) types);

      if ((GST_MESSAGE_TYPE (message) & types) != 0) {

        /* Extra check to ensure extended types don't get matched unless

         * asked for */

        if ((!GST_MESSAGE_TYPE_IS_EXTENDED (message))

            || (types & GST_MESSAGE_EXTENDED)) {

          /* exit the loop, we have a message */

          goto beach;

        }

      }

      GST_DEBUG_OBJECT (bus, "discarding message, does not match mask");

      gst_message_unref (message);

      message = NULL;

    }

    /* no need to wait, exit loop */

    if (timeout == 0)

      break;

    else if (timeout != GST_CLOCK_TIME_NONE) {

      if (first_round) {

        g_get_current_time (&then;;

        first_round = FALSE;

      } else {

        g_get_current_time (&now;;

        elapsed = GST_TIMEVAL_TO_TIME (now) - GST_TIMEVAL_TO_TIME (then);

        if (elapsed > timeout)

          break;

      }

    }

    /* only here in timeout case */

    g_assert (bus->priv->poll);

    g_mutex_unlock (&bus->priv->queue_lock);

    ret = gst_poll_wait (bus->priv->poll, timeout - elapsed);

    g_mutex_lock (&bus->priv->queue_lock);

    if (ret == 0) {

      GST_INFO_OBJECT (bus, "timed out, breaking loop");

      break;

    } else {

      GST_INFO_OBJECT (bus, "we got woken up, recheck for message");

    }

  }

beach:

  g_mutex_unlock (&bus->priv->queue_lock);

  return message;

}

/**

 * gst_bus_timed_pop:

 * @bus: a #GstBus to pop

 * @timeout: a timeout

 *

 * Get a message from the bus, waiting up to the specified timeout.

 *

 * If @timeout is 0, this function behaves like gst_bus_pop(). If @timeout is

 * #GST_CLOCK_TIME_NONE, this function will block forever until a message was

 * posted on the bus.

 *

 * Returns: (transfer full) (nullable): the #GstMessage that is on the

 *     bus after the specified timeout or %NULL if the bus is empty

 *     after the timeout expired.  The message is taken from the bus

 *     and needs to be unreffed with gst_message_unref() after usage.

 *

 * MT safe.

 */

GstMessage *

gst_bus_timed_pop (GstBus * bus, GstClockTime timeout)

{

  g_return_val_if_fail (GST_IS_BUS (bus), NULL);

  return gst_bus_timed_pop_filtered (bus, timeout, GST_MESSAGE_ANY);

}

/**

 * gst_bus_pop_filtered:

 * @bus: a #GstBus to pop

 * @types: message types to take into account

 *

 * Get a message matching @type from the bus.  Will discard all messages on

 * the bus that do not match @type and that have been posted before the first

 * message that does match @type.  If there is no message matching @type on

 * the bus, all messages will be discarded. It is not possible to use message

 * enums beyond #GST_MESSAGE_EXTENDED in the @events mask.

 *

 * Returns: (transfer full) (nullable): the next #GstMessage matching

 *     @type that is on the bus, or %NULL if the bus is empty or there

 *     is no message matching @type. The message is taken from the bus

 *     and needs to be unreffed with gst_message_unref() after usage.

 *

 * MT safe.

 */

GstMessage *

gst_bus_pop_filtered (GstBus * bus, GstMessageType types)

{

  g_return_val_if_fail (GST_IS_BUS (bus), NULL);

  g_return_val_if_fail (types != 0, NULL);

  return gst_bus_timed_pop_filtered (bus, 0, types);

}

/**

 * gst_bus_pop:

 * @bus: a #GstBus to pop

 *

 * Get a message from the bus.

 *

 * Returns: (transfer full) (nullable): the #GstMessage that is on the

 *     bus, or %NULL if the bus is empty. The message is taken from

 *     the bus and needs to be unreffed with gst_message_unref() after

 *     usage.

 *

 * MT safe.

 */

GstMessage *

gst_bus_pop (GstBus * bus)

{

  g_return_val_if_fail (GST_IS_BUS (bus), NULL);

  return gst_bus_timed_pop_filtered (bus, 0, GST_MESSAGE_ANY);

}

/**

 * gst_bus_peek:

 * @bus: a #GstBus

 *

 * Peek the message on the top of the bus' queue. The message will remain

 * on the bus' message queue. A reference is returned, and needs to be unreffed

 * by the caller.

 *

 * Returns: (transfer full) (nullable): the #GstMessage that is on the

 *     bus, or %NULL if the bus is empty.

 *

 * MT safe.

 */

GstMessage *

gst_bus_peek (GstBus * bus)

{

  GstMessage *message;

  g_return_val_if_fail (GST_IS_BUS (bus), NULL);

  g_mutex_lock (&bus->priv->queue_lock);

  message = gst_atomic_queue_peek (bus->priv->queue);

  if (message)

    gst_message_ref (message);

  g_mutex_unlock (&bus->priv->queue_lock);

  GST_DEBUG_OBJECT (bus, "peek on bus, got message %p", message);

  return message;

}

/**

 * gst_bus_set_sync_handler:

 * @bus: a #GstBus to install the handler on

 * @func: (allow-none): The handler function to install

 * @user_data: User data that will be sent to the handler function.

 * @notify: called when @user_data becomes unused

 *

 * Sets the synchronous handler on the bus. The function will be called

 * every time a new message is posted on the bus. Note that the function

 * will be called in the same thread context as the posting object. This

 * function is usually only called by the creator of the bus. Applications

 * should handle messages asynchronously using the gst_bus watch and poll

 * functions.

 *

 * You cannot replace an existing sync_handler. You can pass %NULL to this

 * function, which will clear the existing handler.

 */

void

gst_bus_set_sync_handler (GstBus * bus, GstBusSyncHandler func,

    gpointer user_data, GDestroyNotify notify)

{

  GDestroyNotify old_notify;

  g_return_if_fail (GST_IS_BUS (bus));

  GST_OBJECT_LOCK (bus);

  /* Assert if the user attempts to replace an existing sync_handler,

   * other than to clear it */

  if (func != NULL && bus->priv->sync_handler != NULL)

    goto no_replace;

  if ((old_notify = bus->priv->sync_handler_notify)) {

    gpointer old_data = bus->priv->sync_handler_data;

    bus->priv->sync_handler_data = NULL;

    bus->priv->sync_handler_notify = NULL;

    GST_OBJECT_UNLOCK (bus);

    old_notify (old_data);

    GST_OBJECT_LOCK (bus);

  }

  bus->priv->sync_handler = func;

  bus->priv->sync_handler_data = user_data;

  bus->priv->sync_handler_notify = notify;

  GST_OBJECT_UNLOCK (bus);

  return;

no_replace:

  {

    GST_OBJECT_UNLOCK (bus);

    g_warning ("cannot replace existing sync handler");

    return;

  }

}

/**

 * gst_bus_get_pollfd:

 * @bus: A #GstBus

 * @fd: (out): A GPollFD to fill

 *

 * Gets the file descriptor from the bus which can be used to get notified about

 * messages being available with functions like g_poll(), and allows integration

 * into other event loops based on file descriptors.

 * Whenever a message is available, the POLLIN / %G_IO_IN event is set.

 *

 * Warning: NEVER read or write anything to the returned fd but only use it

 * for getting notifications via g_poll() or similar and then use the normal

 * GstBus API, e.g. gst_bus_pop().

 *

 * Since: 1.14

 */

void

gst_bus_get_pollfd (GstBus * bus, GPollFD * fd)

{

  g_return_if_fail (GST_IS_BUS (bus));

  g_return_if_fail (bus->priv->poll != NULL);

  *fd = bus->priv->pollfd;

}

/* GSource for the bus

 */

typedef struct

{

  GSource source;

  GstBus *bus;

} GstBusSource;

static gboolean

gst_bus_source_prepare (GSource * source, gint * timeout)

{

  *timeout = -1;

  return FALSE;

}

static gboolean

gst_bus_source_check (GSource * source)

{

  GstBusSource *bsrc = (GstBusSource *) source;

  return bsrc->bus->priv->pollfd.revents & (G_IO_IN | G_IO_HUP | G_IO_ERR);

}

static gboolean

gst_bus_source_dispatch (GSource * source, GSourceFunc callback,

    gpointer user_data)

{

  GstBusFunc handler = (GstBusFunc) callback;

  GstBusSource *bsource = (GstBusSource *) source;

  GstMessage *message;

  gboolean keep;

  GstBus *bus;

  g_return_val_if_fail (bsource != NULL, FALSE);

  bus = bsource->bus;

  g_return_val_if_fail (GST_IS_BUS (bus), FALSE);

  message = gst_bus_pop (bus);

  /* The message queue might be empty if some other thread or callback set

   * the bus to flushing between check/prepare and dispatch */

  if (G_UNLIKELY (message == NULL))

    return TRUE;

  if (!handler)

    goto no_handler;

  GST_DEBUG_OBJECT (bus, "source %p calling dispatch with %" GST_PTR_FORMAT,

      source, message);

  keep = handler (bus, message, user_data);

  gst_message_unref (message);

  GST_DEBUG_OBJECT (bus, "source %p handler returns %d", source, keep);

  return keep;

no_handler:

  {

    g_warning ("GstBus watch dispatched without callback\n"

        "You must call g_source_set_callback().");

    gst_message_unref (message);

    return FALSE;

  }

}

static void

gst_bus_source_finalize (GSource * source)

{

  GstBusSource *bsource = (GstBusSource *) source;

  GstBus *bus;

  bus = bsource->bus;

  GST_DEBUG_OBJECT (bus, "finalize source %p", source);

  GST_OBJECT_LOCK (bus);

  if (bus->priv->signal_watch == source)

    bus->priv->signal_watch = NULL;

  GST_OBJECT_UNLOCK (bus);

  gst_object_unref (bsource->bus);

  bsource->bus = NULL;

}

static GSourceFuncs gst_bus_source_funcs = {

  gst_bus_source_prepare,

  gst_bus_source_check,

  gst_bus_source_dispatch,

  gst_bus_source_finalize

};