/* 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., 59 Temple Place - Suite 330,

 * Boston, MA 02111-1307, USA.

 */

/**

 * SECTION:gstbus

 * @short_description: Asynchronous message bus subsystem

 * @see_also: #GstMessage, #GstElement

 *

 * The #GstBus is an object responsible for delivering #GstMessages in

 * a first-in first-out way from the streaming threads 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 _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.

 *

 * Last reviewed on 2006-03-12 (0.10.5)

 */

#include "gst_private.h"

#include <errno.h>

#ifdef HAVE_UNISTD_H

#  include <unistd.h>

#endif

#include <sys/types.h>

#include "gstinfo.h"

#include "gstbus.h"

#ifdef __SYMBIAN32__

#include <glib_global.h>

#endif

#define GST_CAT_DEFAULT GST_CAT_BUS

/* bus signals */

enum

{

  SYNC_MESSAGE,

  ASYNC_MESSAGE,

  /* add more above */

  LAST_SIGNAL

};

static void gst_bus_class_init (GstBusClass * klass);

static void gst_bus_init (GstBus * bus);

static void gst_bus_dispose (GObject * object);

static GstObjectClass *parent_class = NULL;

static guint gst_bus_signals[LAST_SIGNAL] = { 0 };

struct _GstBusPrivate

{

  guint num_sync_message_emitters;

  GCond *queue_cond;

};

/* fixme: do something about this */

static void

marshal_VOID__MINIOBJECT (GClosure * closure, GValue * return_value,

    guint n_param_values, const GValue * param_values, gpointer invocation_hint,

    gpointer marshal_data)

{

  typedef void (*marshalfunc_VOID__MINIOBJECT) (gpointer obj, gpointer arg1,

      gpointer data2);

  register marshalfunc_VOID__MINIOBJECT callback;

  register GCClosure *cc = (GCClosure *) closure;

  register gpointer data1, data2;

  g_return_if_fail (n_param_values == 2);

  if (G_CCLOSURE_SWAP_DATA (closure)) {

    data1 = closure->data;

    data2 = g_value_peek_pointer (param_values + 0);

  } else {

    data1 = g_value_peek_pointer (param_values + 0);

    data2 = closure->data;

  }

  callback =

  callback (data1, gst_value_get_mini_object (param_values + 1), data2);

}

static void

gst_bus_class_init (GstBusClass * klass)

{

  parent_class = g_type_class_peek_parent (klass);

  gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_bus_dispose);

  /**

   * 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 set up

   * gst_bus_sync_signal_handler() as a sync handler if you want this

   * signal to be emitted when a message is posted on the bus, like this:

   * <programlisting>

   * gst_bus_set_sync_handler (bus, gst_bus_sync_signal_handler, yourdata);

   * </programlisting>

   */

  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,

      marshal_VOID__MINIOBJECT, 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,

      marshal_VOID__MINIOBJECT, G_TYPE_NONE, 1, GST_TYPE_MESSAGE);

  g_type_class_add_private (klass, sizeof (GstBusPrivate));

}

static void

gst_bus_init (GstBus * bus)

{

  bus->queue = g_queue_new ();

  bus->queue_lock = g_mutex_new ();

  bus->priv = G_TYPE_INSTANCE_GET_PRIVATE (bus, GST_TYPE_BUS, GstBusPrivate);

  bus->priv->queue_cond = g_cond_new ();

  GST_DEBUG_OBJECT (bus, "created");

}

static void

gst_bus_dispose (GObject * object)

{

  if (bus->queue) {

    GstMessage *message;

    g_mutex_lock (bus->queue_lock);

    do {

      message = g_queue_pop_head (bus->queue);

      if (message)

        gst_message_unref (message);

    } while (message != NULL);

    g_queue_free (bus->queue);

    bus->queue = NULL;

    g_mutex_unlock (bus->queue_lock);

    g_mutex_free (bus->queue_lock);

    bus->queue_lock = NULL;

    g_cond_free (bus->priv->queue_cond);

    bus->priv->queue_cond = NULL;

  }

  G_OBJECT_CLASS (parent_class)->dispose (object);

}

static void

{

}

static void

{

}

/**

 * gst_bus_new:

 *

 * Creates a new #GstBus instance.

 *

 * Returns: a new #GstBus instance

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

GstBus *

gst_bus_new (void)

{

  GstBus *result;

  result = g_object_new (gst_bus_get_type (), NULL);

  GST_DEBUG_OBJECT (result, "created new bus");

  return result;

}

/**

 * gst_bus_post:

 * @bus: a #GstBus to post on

 * @message: 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.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

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, type %s, %" GST_PTR_FORMAT

      " from source %" GST_PTR_FORMAT,

      message, GST_MESSAGE_TYPE_NAME (message), message->structure,

      message->src);

  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->sync_handler;

  handler_data = bus->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);

  /* 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);

      g_mutex_lock (bus->queue_lock);

      g_queue_push_tail (bus->queue, message);

      g_cond_broadcast (bus->priv->queue_cond);

      g_mutex_unlock (bus->queue_lock);

      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 */

      GMutex *lock = g_mutex_new ();

      GCond *cond = g_cond_new ();

      GST_MESSAGE_COND (message) = cond;

      GST_MESSAGE_GET_LOCK (message) = lock;

      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);

      g_mutex_lock (bus->queue_lock);

      g_queue_push_tail (bus->queue, message);

      g_cond_broadcast (bus->priv->queue_cond);

      g_mutex_unlock (bus->queue_lock);

      /* now block till the message is freed */

      g_cond_wait (cond, lock);

      g_mutex_unlock (lock);

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

      g_mutex_free (lock);

      g_cond_free (cond);

      break;

    }

    default:

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

      break;

  }

  return TRUE;

  /* ERRORS */

is_flushing:

  {

    GST_DEBUG_OBJECT (bus, "bus is flushing");

    gst_message_unref (message);

    GST_OBJECT_UNLOCK (bus);

    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.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

gboolean

gst_bus_have_pending (GstBus * bus)

{

  gboolean result;

  g_return_val_if_fail (GST_IS_BUS (bus), FALSE);

  g_mutex_lock (bus->queue_lock);

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

  result = !g_queue_is_empty (bus->queue);

  g_mutex_unlock (bus->queue_lock);

  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.

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

void

gst_bus_set_flushing (GstBus * bus, gboolean flushing)

{

  GstMessage *message;

  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)))

      gst_message_unref (message);

  } else {

    GST_DEBUG_OBJECT (bus, "unset bus flushing");

    GST_OBJECT_FLAG_UNSET (bus, GST_BUS_FLUSHING);

  }

  GST_OBJECT_UNLOCK (bus);

}

/**

 * 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: 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.

 *

 * Since: 0.10.15

 */

#ifdef __SYMBIAN32__

EXPORT_C

#endif

GstMessage *

gst_bus_timed_pop_filtered (GstBus * bus, GstClockTime timeout,

    GstMessageType types)

{

  GstMessage *message;

  GTimeVal *timeval, abstimeout;

  gboolean first_round = TRUE;

  g_return_val_if_fail (GST_IS_BUS (bus), NULL);

  g_return_val_if_fail (types != 0, NULL);

  g_mutex_lock (bus->queue_lock);

  while (TRUE) {

    GST_LOG_OBJECT (bus, "have %d messages", g_queue_get_length (bus->queue));

    while ((message = g_queue_pop_head (bus->queue))) {

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

          message, GST_MESSAGE_TYPE_NAME (message), (guint) types);