renatofilho@787: /* GStreamer
renatofilho@787:  * Copyright (C) 2006 Edward Hervey <edward@fluendo.com>
renatofilho@787:  *
renatofilho@787:  * gstdataqueue.c:
renatofilho@787:  *
renatofilho@787:  * This library is free software; you can redistribute it and/or
renatofilho@787:  * modify it under the terms of the GNU Library General Public
renatofilho@787:  * License as published by the Free Software Foundation; either
renatofilho@787:  * version 2 of the License, or (at your option) any later version.
renatofilho@787:  *
renatofilho@787:  * This library is distributed in the hope that it will be useful,
renatofilho@787:  * but WITHOUT ANY WARRANTY; without even the implied warranty of
renatofilho@787:  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
renatofilho@787:  * Library General Public License for more details.
renatofilho@787:  *
renatofilho@787:  * You should have received a copy of the GNU Library General Public
renatofilho@787:  * License along with this library; if not, write to the
renatofilho@787:  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
renatofilho@787:  * Boston, MA 02111-1307, USA.
renatofilho@787:  */
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * SECTION:gstdataqueue
renatofilho@787:  * @short_description: Threadsafe queueing object
renatofilho@787:  *
renatofilho@787:  * #GstDataQueue is an object that handles threadsafe queueing of objects. It
renatofilho@787:  * also provides size-related functionality. This object should be used for
renatofilho@787:  * any #GstElement that wishes to provide some sort of queueing functionality.
renatofilho@787:  */
renatofilho@787: 
renatofilho@787: #include <gst/gst.h>
renatofilho@787: #include "gstdataqueue.h"
renatofilho@787: 
renatofilho@787: GST_DEBUG_CATEGORY_STATIC (data_queue_debug);
renatofilho@787: #define GST_CAT_DEFAULT (data_queue_debug)
renatofilho@787: GST_DEBUG_CATEGORY_STATIC (data_queue_dataflow);
renatofilho@787: 
renatofilho@787: 
renatofilho@787: /* Queue signals and args */
renatofilho@787: enum
renatofilho@787: {
renatofilho@787:   SIGNAL_EMPTY,
renatofilho@787:   SIGNAL_FULL,
renatofilho@787:   LAST_SIGNAL
renatofilho@787: };
renatofilho@787: 
renatofilho@787: enum
renatofilho@787: {
renatofilho@787:   ARG_0,
renatofilho@787:   ARG_CUR_LEVEL_VISIBLE,
renatofilho@787:   ARG_CUR_LEVEL_BYTES,
renatofilho@787:   ARG_CUR_LEVEL_TIME
renatofilho@787:       /* FILL ME */
renatofilho@787: };
renatofilho@787: 
renatofilho@787: #define GST_DATA_QUEUE_MUTEX_LOCK(q) G_STMT_START {                     \
renatofilho@787:     GST_CAT_LOG (data_queue_dataflow,                                   \
renatofilho@787:       "locking qlock from thread %p",                                   \
renatofilho@787:       g_thread_self ());                                                \
renatofilho@787:   g_mutex_lock (q->qlock);                                              \
renatofilho@787:   GST_CAT_LOG (data_queue_dataflow,                                     \
renatofilho@787:       "locked qlock from thread %p",                                    \
renatofilho@787:       g_thread_self ());                                                \
renatofilho@787: } G_STMT_END
renatofilho@787: 
renatofilho@787: #define GST_DATA_QUEUE_MUTEX_LOCK_CHECK(q, label) G_STMT_START {        \
renatofilho@787:     GST_DATA_QUEUE_MUTEX_LOCK (q);                                      \
renatofilho@787:     if (q->flushing)                                                    \
renatofilho@787:       goto label;                                                       \
renatofilho@787:   } G_STMT_END
renatofilho@787: 
renatofilho@787: #define GST_DATA_QUEUE_MUTEX_UNLOCK(q) G_STMT_START {                   \
renatofilho@787:     GST_CAT_LOG (data_queue_dataflow,                                   \
renatofilho@787:       "unlocking qlock from thread %p",                                 \
renatofilho@787:       g_thread_self ());                                                \
renatofilho@787:   g_mutex_unlock (q->qlock);                                            \
renatofilho@787: } G_STMT_END
renatofilho@787: 
renatofilho@787: #define STATUS(q, msg)                                                  \
renatofilho@787:   GST_CAT_LOG (data_queue_dataflow,                                     \
renatofilho@787:                "queue:%p " msg ": %u visible items, %u "                \
renatofilho@787:                "bytes, %"G_GUINT64_FORMAT                               \
renatofilho@787:                " ns, %u elements",                                      \
renatofilho@787:                queue,                                                   \
renatofilho@787:                q->cur_level.visible,                                    \
renatofilho@787:                q->cur_level.bytes,                                      \
renatofilho@787:                q->cur_level.time,                                       \
renatofilho@787:                q->queue->length)
renatofilho@787: 
renatofilho@787: static void gst_data_queue_base_init (GstDataQueueClass * klass);
renatofilho@787: static void gst_data_queue_class_init (GstDataQueueClass * klass);
renatofilho@787: static void gst_data_queue_init (GstDataQueue * queue);
renatofilho@787: static void gst_data_queue_finalize (GObject * object);
renatofilho@787: 
renatofilho@787: static void gst_data_queue_set_property (GObject * object,
renatofilho@787:     guint prop_id, const GValue * value, GParamSpec * pspec);
renatofilho@787: static void gst_data_queue_get_property (GObject * object,
renatofilho@787:     guint prop_id, GValue * value, GParamSpec * pspec);
renatofilho@787: 
renatofilho@787: static GObjectClass *parent_class = NULL;
renatofilho@787: static guint gst_data_queue_signals[LAST_SIGNAL] = { 0 };
renatofilho@787: 
renatofilho@787: GType
renatofilho@787: gst_data_queue_get_type (void)
renatofilho@787: {
renatofilho@787:   static GType queue_type = 0;
renatofilho@787: 
renatofilho@787:   if (!queue_type) {
renatofilho@787:     static const GTypeInfo queue_info = {
renatofilho@787:       sizeof (GstDataQueueClass),
renatofilho@787:       (GBaseInitFunc) gst_data_queue_base_init,
renatofilho@787:       NULL,
renatofilho@787:       (GClassInitFunc) gst_data_queue_class_init,
renatofilho@787:       NULL,
renatofilho@787:       NULL,
renatofilho@787:       sizeof (GstDataQueue),
renatofilho@787:       0,
renatofilho@787:       (GInstanceInitFunc) gst_data_queue_init,
renatofilho@787:       NULL
renatofilho@787:     };
renatofilho@787: 
renatofilho@787:     queue_type = g_type_register_static (G_TYPE_OBJECT,
renatofilho@787:         "GstDataQueue", &queue_info, 0);
renatofilho@787:     GST_DEBUG_CATEGORY_INIT (data_queue_debug, "dataqueue", 0,
renatofilho@787:         "data queue object");
renatofilho@787:     GST_DEBUG_CATEGORY_INIT (data_queue_dataflow, "data_queue_dataflow", 0,
renatofilho@787:         "dataflow inside the data queue object");
renatofilho@787:   }
renatofilho@787: 
renatofilho@787:   return queue_type;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static void
renatofilho@787: gst_data_queue_base_init (GstDataQueueClass * klass)
renatofilho@787: {
renatofilho@787:   /* Do we need anything here ?? */
renatofilho@787:   return;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static void
renatofilho@787: gst_data_queue_class_init (GstDataQueueClass * klass)
renatofilho@787: {
renatofilho@787:   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
renatofilho@787: 
renatofilho@787:   parent_class = g_type_class_peek_parent (klass);
renatofilho@787: 
renatofilho@787:   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_data_queue_set_property);
renatofilho@787:   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_data_queue_get_property);
renatofilho@787: 
renatofilho@787:   /* signals */
renatofilho@787:   /**
renatofilho@787:    * GstDataQueue::empty:
renatofilho@787:    * @queue: the queue instance
renatofilho@787:    *
renatofilho@787:    * Reports that the queue became empty (empty).
renatofilho@787:    * A queue is empty if the total amount of visible items inside it (num-visible, time,
renatofilho@787:    * size) is lower than the boundary values which can be set through the GObject
renatofilho@787:    * properties.
renatofilho@787:    */
renatofilho@787:   gst_data_queue_signals[SIGNAL_EMPTY] =
renatofilho@787:       g_signal_new ("empty", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
renatofilho@787:       G_STRUCT_OFFSET (GstDataQueueClass, empty), NULL, NULL,
renatofilho@787:       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
renatofilho@787: 
renatofilho@787:   /**
renatofilho@787:    * GstDataQueue::full:
renatofilho@787:    * @queue: the queue instance
renatofilho@787:    *
renatofilho@787:    * Reports that the queue became full (full).
renatofilho@787:    * A queue is full if the total amount of data inside it (num-visible, time,
renatofilho@787:    * size) is higher than the boundary values which can be set through the GObject
renatofilho@787:    * properties.
renatofilho@787:    */
renatofilho@787:   gst_data_queue_signals[SIGNAL_FULL] =
renatofilho@787:       g_signal_new ("full", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
renatofilho@787:       G_STRUCT_OFFSET (GstDataQueueClass, full), NULL, NULL,
renatofilho@787:       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
renatofilho@787: 
renatofilho@787:   /* properties */
renatofilho@787:   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_BYTES,
renatofilho@787:       g_param_spec_uint ("current-level-bytes", "Current level (kB)",
renatofilho@787:           "Current amount of data in the queue (bytes)",
renatofilho@787:           0, G_MAXUINT, 0, G_PARAM_READABLE));
renatofilho@787:   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_VISIBLE,
renatofilho@787:       g_param_spec_uint ("current-level-visible",
renatofilho@787:           "Current level (visible items)",
renatofilho@787:           "Current number of visible items in the queue", 0, G_MAXUINT, 0,
renatofilho@787:           G_PARAM_READABLE));
renatofilho@787:   g_object_class_install_property (gobject_class, ARG_CUR_LEVEL_TIME,
renatofilho@787:       g_param_spec_uint64 ("current-level-time", "Current level (ns)",
renatofilho@787:           "Current amount of data in the queue (in ns)", 0, G_MAXUINT64, 0,
renatofilho@787:           G_PARAM_READABLE));
renatofilho@787: 
renatofilho@787:   /* set several parent class virtual functions */
renatofilho@787:   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_data_queue_finalize);
renatofilho@787: 
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static void
renatofilho@787: gst_data_queue_init (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   queue->cur_level.visible = 0; /* no content */
renatofilho@787:   queue->cur_level.bytes = 0;   /* no content */
renatofilho@787:   queue->cur_level.time = 0;    /* no content */
renatofilho@787: 
renatofilho@787:   queue->checkfull = NULL;
renatofilho@787: 
renatofilho@787:   queue->qlock = g_mutex_new ();
renatofilho@787:   queue->item_add = g_cond_new ();
renatofilho@787:   queue->item_del = g_cond_new ();
renatofilho@787:   queue->queue = g_queue_new ();
renatofilho@787: 
renatofilho@787:   GST_DEBUG ("initialized queue's not_empty & not_full conditions");
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_new:
renatofilho@787:  * @checkfull: the callback used to tell if the element considers the queue full
renatofilho@787:  * or not.
renatofilho@787:  * @checkdata: a #gpointer that will be given in the @checkfull callback.
renatofilho@787:  *
renatofilho@787:  * Returns: a new #GstDataQueue.
renatofilho@787:  */
renatofilho@787: 
renatofilho@787: GstDataQueue *
renatofilho@787: gst_data_queue_new (GstDataQueueCheckFullFunction checkfull, gpointer checkdata)
renatofilho@787: {
renatofilho@787:   GstDataQueue *ret;
renatofilho@787: 
renatofilho@787:   g_return_val_if_fail (checkfull != NULL, NULL);
renatofilho@787: 
renatofilho@787:   ret = g_object_new (GST_TYPE_DATA_QUEUE, NULL);
renatofilho@787:   ret->checkfull = checkfull;
renatofilho@787:   ret->checkdata = checkdata;
renatofilho@787: 
renatofilho@787:   return ret;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static void
renatofilho@787: gst_data_queue_cleanup (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   while (!g_queue_is_empty (queue->queue)) {
renatofilho@787:     GstDataQueueItem *item = g_queue_pop_head (queue->queue);
renatofilho@787: 
renatofilho@787:     /* Just call the destroy notify on the item */
renatofilho@787:     item->destroy (item);
renatofilho@787:   }
renatofilho@787:   queue->cur_level.visible = 0;
renatofilho@787:   queue->cur_level.bytes = 0;
renatofilho@787:   queue->cur_level.time = 0;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /* called only once, as opposed to dispose */
renatofilho@787: static void
renatofilho@787: gst_data_queue_finalize (GObject * object)
renatofilho@787: {
renatofilho@787:   GstDataQueue *queue = GST_DATA_QUEUE (object);
renatofilho@787: 
renatofilho@787:   GST_DEBUG ("finalizing queue");
renatofilho@787: 
renatofilho@787:   gst_data_queue_cleanup (queue);
renatofilho@787:   g_queue_free (queue->queue);
renatofilho@787: 
renatofilho@787:   GST_DEBUG ("free mutex");
renatofilho@787:   g_mutex_free (queue->qlock);
renatofilho@787:   GST_DEBUG ("done free mutex");
renatofilho@787: 
renatofilho@787:   g_cond_free (queue->item_add);
renatofilho@787:   g_cond_free (queue->item_del);
renatofilho@787: 
renatofilho@787:   G_OBJECT_CLASS (parent_class)->finalize (object);
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static void
renatofilho@787: gst_data_queue_locked_flush (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   STATUS (queue, "before flushing");
renatofilho@787:   gst_data_queue_cleanup (queue);
renatofilho@787:   STATUS (queue, "after flushing");
renatofilho@787:   /* we deleted something... */
renatofilho@787:   g_cond_signal (queue->item_del);
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static gboolean
renatofilho@787: gst_data_queue_locked_is_empty (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   return (queue->queue->length == 0);
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static gboolean
renatofilho@787: gst_data_queue_locked_is_full (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   return queue->checkfull (queue, queue->cur_level.visible,
renatofilho@787:       queue->cur_level.bytes, queue->cur_level.time, queue->checkdata);
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_flush:
renatofilho@787:  * @queue: a #GstDataQueue.
renatofilho@787:  *
renatofilho@787:  * Flushes all the contents of the @queue. Any call to #gst_data_queue_pull and
renatofilho@787:  * #gst_data_queue_pop will be released.
renatofilho@787:  * MT safe.
renatofilho@787:  */
renatofilho@787: void
renatofilho@787: gst_data_queue_flush (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   GST_DEBUG ("queue:%p", queue);
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK (queue);
renatofilho@787:   gst_data_queue_locked_flush (queue);
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_is_empty:
renatofilho@787:  * @queue: a #GstDataQueue.
renatofilho@787:  *
renatofilho@787:  * Queries if there are any items in the @queue.
renatofilho@787:  * MT safe.
renatofilho@787:  *
renatofilho@787:  * Returns: #TRUE if @queue is empty.
renatofilho@787:  */
renatofilho@787: gboolean
renatofilho@787: gst_data_queue_is_empty (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   gboolean res;
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK (queue);
renatofilho@787:   res = gst_data_queue_locked_is_empty (queue);
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: 
renatofilho@787:   return res;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_is_full:
renatofilho@787:  * @queue: a #GstDataQueue.
renatofilho@787:  *
renatofilho@787:  * Queries if @queue is full. This check will be done using the
renatofilho@787:  * #GstDataQueueCheckFullCallback registered with @queue.
renatofilho@787:  * MT safe.
renatofilho@787:  *
renatofilho@787:  * Returns: #TRUE if @queue is full.
renatofilho@787:  */
renatofilho@787: gboolean
renatofilho@787: gst_data_queue_is_full (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   gboolean res;
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK (queue);
renatofilho@787:   res = gst_data_queue_locked_is_full (queue);
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: 
renatofilho@787:   return res;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_set_flushing:
renatofilho@787:  * @queue: a #GstDataQueue.
renatofilho@787:  * @flushing: a #gboolean stating if the queue will be flushing or not.
renatofilho@787:  *
renatofilho@787:  * Sets the queue to flushing state if @flushing is #TRUE. If set to flushing
renatofilho@787:  * state, any incoming data on the @queue will be discarded. Any call currently
renatofilho@787:  * blocking on #gst_data_queue_push or #gst_data_queue_pop will return straight
renatofilho@787:  * away with a return value of #FALSE. While the @queue is in flushing state, 
renatofilho@787:  * all calls to those two functions will return #FALSE.
renatofilho@787:  *
renatofilho@787:  * MT Safe.
renatofilho@787:  */
renatofilho@787: void
renatofilho@787: gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
renatofilho@787: {
renatofilho@787:   GST_DEBUG ("queue:%p , flushing:%d", queue, flushing);
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK (queue);
renatofilho@787:   queue->flushing = flushing;
renatofilho@787:   if (flushing) {
renatofilho@787:     /* release push/pop functions */
renatofilho@787:     g_cond_signal (queue->item_add);
renatofilho@787:     g_cond_signal (queue->item_del);
renatofilho@787:   }
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_push:
renatofilho@787:  * @queue: a #GstDataQueue.
renatofilho@787:  * @item: a #GstDataQueueItem.
renatofilho@787:  *
renatofilho@787:  * Pushes a #GstDataQueueItem (or a structure that begins with the same fields)
renatofilho@787:  * on the @queue. If the @queue is full, the call will block until space is
renatofilho@787:  * available, OR the @queue is set to flushing state.
renatofilho@787:  * MT safe.
renatofilho@787:  *
renatofilho@787:  * Note that this function has slightly different semantics than gst_pad_push()
renatofilho@787:  * and gst_pad_push_event(): this function only takes ownership of @item and
renatofilho@787:  * the #GstMiniObject contained in @item if the push was successful. If FALSE
renatofilho@787:  * is returned, the caller is responsible for freeing @item and its contents.
renatofilho@787:  *
renatofilho@787:  * Returns: #TRUE if the @item was successfully pushed on the @queue.
renatofilho@787:  */
renatofilho@787: gboolean
renatofilho@787: gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item)
renatofilho@787: {
renatofilho@787:   g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
renatofilho@787:   g_return_val_if_fail (item != NULL, FALSE);
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
renatofilho@787: 
renatofilho@787:   STATUS (queue, "before pushing");
renatofilho@787: 
renatofilho@787:   /* We ALWAYS need to check for queue fillness */
renatofilho@787:   if (gst_data_queue_locked_is_full (queue)) {
renatofilho@787:     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787:     g_signal_emit (G_OBJECT (queue), gst_data_queue_signals[SIGNAL_FULL], 0);
renatofilho@787:     GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
renatofilho@787: 
renatofilho@787:     /* signal might have removed some items */
renatofilho@787:     while (gst_data_queue_locked_is_full (queue)) {
renatofilho@787:       g_cond_wait (queue->item_del, queue->qlock);
renatofilho@787:       if (queue->flushing)
renatofilho@787:         goto flushing;
renatofilho@787:     }
renatofilho@787:   }
renatofilho@787: 
renatofilho@787:   g_queue_push_tail (queue->queue, item);
renatofilho@787: 
renatofilho@787:   if (item->visible)
renatofilho@787:     queue->cur_level.visible++;
renatofilho@787:   queue->cur_level.bytes += item->size;
renatofilho@787:   queue->cur_level.time += item->duration;
renatofilho@787: 
renatofilho@787:   STATUS (queue, "after pushing");
renatofilho@787:   g_cond_signal (queue->item_add);
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: 
renatofilho@787:   return TRUE;
renatofilho@787: 
renatofilho@787:   /* ERRORS */
renatofilho@787: flushing:
renatofilho@787:   {
renatofilho@787:     GST_DEBUG ("queue:%p, we are flushing", queue);
renatofilho@787:     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787:     return FALSE;
renatofilho@787:   }
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_pop:
renatofilho@787:  * @queue: a #GstDataQueue.
renatofilho@787:  * @item: pointer to store the returned #GstDataQueueItem.
renatofilho@787:  *
renatofilho@787:  * Retrieves the first @item available on the @queue. If the queue is currently
renatofilho@787:  * empty, the call will block until at least one item is available, OR the
renatofilho@787:  * @queue is set to the flushing state.
renatofilho@787:  * MT safe.
renatofilho@787:  *
renatofilho@787:  * Returns: #TRUE if an @item was successfully retrieved from the @queue.
renatofilho@787:  */
renatofilho@787: gboolean
renatofilho@787: gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item)
renatofilho@787: {
renatofilho@787:   g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
renatofilho@787:   g_return_val_if_fail (item != NULL, FALSE);
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
renatofilho@787: 
renatofilho@787:   STATUS (queue, "before popping");
renatofilho@787: 
renatofilho@787:   if (gst_data_queue_locked_is_empty (queue)) {
renatofilho@787:     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787:     g_signal_emit (G_OBJECT (queue), gst_data_queue_signals[SIGNAL_EMPTY], 0);
renatofilho@787:     GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
renatofilho@787: 
renatofilho@787:     while (gst_data_queue_locked_is_empty (queue)) {
renatofilho@787:       g_cond_wait (queue->item_add, queue->qlock);
renatofilho@787:       if (queue->flushing)
renatofilho@787:         goto flushing;
renatofilho@787:     }
renatofilho@787:   }
renatofilho@787: 
renatofilho@787:   /* Get the item from the GQueue */
renatofilho@787:   *item = g_queue_pop_head (queue->queue);
renatofilho@787: 
renatofilho@787:   /* update current level counter */
renatofilho@787:   if ((*item)->visible)
renatofilho@787:     queue->cur_level.visible--;
renatofilho@787:   queue->cur_level.bytes -= (*item)->size;
renatofilho@787:   queue->cur_level.time -= (*item)->duration;
renatofilho@787: 
renatofilho@787:   STATUS (queue, "after popping");
renatofilho@787:   g_cond_signal (queue->item_del);
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: 
renatofilho@787:   return TRUE;
renatofilho@787: 
renatofilho@787:   /* ERRORS */
renatofilho@787: flushing:
renatofilho@787:   {
renatofilho@787:     GST_DEBUG ("queue:%p, we are flushing", queue);
renatofilho@787:     GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787:     return FALSE;
renatofilho@787:   }
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_drop_head:
renatofilho@787:  * @queue: The #GstDataQueue to drop an item from.
renatofilho@787:  * @type: The #GType of the item to drop.
renatofilho@787:  *
renatofilho@787:  * Pop and unref the head-most #GstMiniObject with the given #GType.
renatofilho@787:  *
renatofilho@787:  * Returns: TRUE if an element was removed.
renatofilho@787:  */
renatofilho@787: gboolean
renatofilho@787: gst_data_queue_drop_head (GstDataQueue * queue, GType type)
renatofilho@787: {
renatofilho@787:   gboolean res = FALSE;
renatofilho@787:   GList *item;
renatofilho@787:   GstDataQueueItem *leak = NULL;
renatofilho@787: 
renatofilho@787:   g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
renatofilho@787: 
renatofilho@787:   GST_DEBUG ("queue:%p", queue);
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK (queue);
renatofilho@787:   for (item = g_queue_peek_head_link (queue->queue); item; item = item->next) {
renatofilho@787:     GstDataQueueItem *tmp = (GstDataQueueItem *) item->data;
renatofilho@787: 
renatofilho@787:     if (G_TYPE_CHECK_INSTANCE_TYPE (tmp->object, type)) {
renatofilho@787:       leak = tmp;
renatofilho@787:       break;
renatofilho@787:     }
renatofilho@787:   }
renatofilho@787: 
renatofilho@787:   if (!leak)
renatofilho@787:     goto done;
renatofilho@787: 
renatofilho@787:   g_queue_delete_link (queue->queue, item);
renatofilho@787: 
renatofilho@787:   if (leak->visible)
renatofilho@787:     queue->cur_level.visible--;
renatofilho@787:   queue->cur_level.bytes -= leak->size;
renatofilho@787:   queue->cur_level.time -= leak->duration;
renatofilho@787: 
renatofilho@787:   leak->destroy (leak);
renatofilho@787: 
renatofilho@787:   res = TRUE;
renatofilho@787: 
renatofilho@787: done:
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: 
renatofilho@787:   GST_DEBUG ("queue:%p , res:%d", queue, res);
renatofilho@787: 
renatofilho@787:   return res;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_limits_changed:
renatofilho@787:  * @queue: The #GstDataQueue 
renatofilho@787:  *
renatofilho@787:  * Inform the queue that the limits for the fullness check have changed and that
renatofilho@787:  * any blocking gst_data_queue_push() should be unblocked to recheck the limts.
renatofilho@787:  */
renatofilho@787: void
renatofilho@787: gst_data_queue_limits_changed (GstDataQueue * queue)
renatofilho@787: {
renatofilho@787:   g_return_if_fail (GST_IS_DATA_QUEUE (queue));
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK (queue);
renatofilho@787:   GST_DEBUG ("signal del");
renatofilho@787:   g_cond_signal (queue->item_del);
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: }
renatofilho@787: 
renatofilho@787: /**
renatofilho@787:  * gst_data_queue_get_level:
renatofilho@787:  * @queue: The #GstDataQueue
renatofilho@787:  * @level: the location to store the result
renatofilho@787:  *
renatofilho@787:  * Get the current level of the queue.
renatofilho@787:  */
renatofilho@787: void
renatofilho@787: gst_data_queue_get_level (GstDataQueue * queue, GstDataQueueSize * level)
renatofilho@787: {
renatofilho@787:   level->visible = queue->cur_level.visible;
renatofilho@787:   level->bytes = queue->cur_level.bytes;
renatofilho@787:   level->time = queue->cur_level.time;
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static void
renatofilho@787: gst_data_queue_set_property (GObject * object,
renatofilho@787:     guint prop_id, const GValue * value, GParamSpec * pspec)
renatofilho@787: {
renatofilho@787:   switch (prop_id) {
renatofilho@787:     default:
renatofilho@787:       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
renatofilho@787:       break;
renatofilho@787:   }
renatofilho@787: }
renatofilho@787: 
renatofilho@787: static void
renatofilho@787: gst_data_queue_get_property (GObject * object,
renatofilho@787:     guint prop_id, GValue * value, GParamSpec * pspec)
renatofilho@787: {
renatofilho@787:   GstDataQueue *queue = GST_DATA_QUEUE (object);
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_LOCK (queue);
renatofilho@787: 
renatofilho@787:   switch (prop_id) {
renatofilho@787:     case ARG_CUR_LEVEL_BYTES:
renatofilho@787:       g_value_set_uint (value, queue->cur_level.bytes);
renatofilho@787:       break;
renatofilho@787:     case ARG_CUR_LEVEL_VISIBLE:
renatofilho@787:       g_value_set_uint (value, queue->cur_level.visible);
renatofilho@787:       break;
renatofilho@787:     case ARG_CUR_LEVEL_TIME:
renatofilho@787:       g_value_set_uint64 (value, queue->cur_level.time);
renatofilho@787:       break;
renatofilho@787:     default:
renatofilho@787:       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
renatofilho@787:       break;
renatofilho@787:   }
renatofilho@787: 
renatofilho@787:   GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
renatofilho@787: }