2 * Copyright (C) 2006 Edward Hervey <edward@fluendo.com>
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.
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.
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., 51 Franklin St, Fifth Floor,
19 * Boston, MA 02110-1301, USA.
23 * SECTION:gstdataqueue
24 * @title: GstDataQueue
25 * @short_description: Threadsafe queueing object
27 * #GstDataQueue is an object that handles threadsafe queueing of objects. It
28 * also provides size-related functionality. This object should be used for
29 * any #GstElement that wishes to provide some sort of queueing functionality.
34 #include "gstdataqueue.h"
35 #include "gstqueuearray.h"
36 #include "gst/glib-compat-private.h"
38 GST_DEBUG_CATEGORY_STATIC (data_queue_debug);
39 #define GST_CAT_DEFAULT (data_queue_debug)
40 GST_DEBUG_CATEGORY_STATIC (data_queue_dataflow);
43 /* Queue signals and args */
54 PROP_CUR_LEVEL_VISIBLE,
60 struct _GstDataQueuePrivate
62 /* the array of data we're keeping our grubby hands on */
65 GstDataQueueSize cur_level; /* size of the queue */
66 GstDataQueueCheckFullFunction checkfull; /* Callback to check if the queue is full */
69 GMutex qlock; /* lock for queue (vs object lock) */
71 GCond item_add; /* signals buffers now available for reading */
73 GCond item_del; /* signals space now available for writing */
74 gboolean flushing; /* indicates whether conditions where signalled because
75 * of external flushing */
76 GstDataQueueFullCallback fullcallback;
77 GstDataQueueEmptyCallback emptycallback;
80 #define GST_DATA_QUEUE_MUTEX_LOCK(q) G_STMT_START { \
81 GST_CAT_TRACE (data_queue_dataflow, \
82 "locking qlock from thread %p", \
84 g_mutex_lock (&q->priv->qlock); \
85 GST_CAT_TRACE (data_queue_dataflow, \
86 "locked qlock from thread %p", \
90 #define GST_DATA_QUEUE_MUTEX_LOCK_CHECK(q, label) G_STMT_START { \
91 GST_DATA_QUEUE_MUTEX_LOCK (q); \
92 if (q->priv->flushing) \
96 #define GST_DATA_QUEUE_MUTEX_UNLOCK(q) G_STMT_START { \
97 GST_CAT_TRACE (data_queue_dataflow, \
98 "unlocking qlock from thread %p", \
100 g_mutex_unlock (&q->priv->qlock); \
103 #define STATUS(q, msg) \
104 GST_CAT_LOG (data_queue_dataflow, \
105 "queue:%p " msg ": %u visible items, %u " \
106 "bytes, %"G_GUINT64_FORMAT \
107 " ns, %u elements", \
109 q->priv->cur_level.visible, \
110 q->priv->cur_level.bytes, \
111 q->priv->cur_level.time, \
112 gst_queue_array_get_length (q->priv->queue))
114 static void gst_data_queue_finalize (GObject * object);
116 static void gst_data_queue_set_property (GObject * object,
117 guint prop_id, const GValue * value, GParamSpec * pspec);
118 static void gst_data_queue_get_property (GObject * object,
119 guint prop_id, GValue * value, GParamSpec * pspec);
121 static guint gst_data_queue_signals[LAST_SIGNAL] = { 0 };
125 GST_DEBUG_CATEGORY_INIT (data_queue_debug, "dataqueue", 0, \
126 "data queue object"); \
127 GST_DEBUG_CATEGORY_INIT (data_queue_dataflow, "data_queue_dataflow", 0, \
128 "dataflow inside the data queue object"); \
131 #define parent_class gst_data_queue_parent_class
132 G_DEFINE_TYPE_WITH_CODE (GstDataQueue, gst_data_queue, G_TYPE_OBJECT, _do_init);
135 gst_data_queue_class_init (GstDataQueueClass * klass)
137 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
139 g_type_class_add_private (klass, sizeof (GstDataQueuePrivate));
141 gobject_class->set_property = gst_data_queue_set_property;
142 gobject_class->get_property = gst_data_queue_get_property;
146 * GstDataQueue::empty: (skip)
147 * @queue: the queue instance
149 * Reports that the queue became empty (empty).
150 * A queue is empty if the total amount of visible items inside it (num-visible, time,
151 * size) is lower than the boundary values which can be set through the GObject
154 gst_data_queue_signals[SIGNAL_EMPTY] =
155 g_signal_new ("empty", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
156 G_STRUCT_OFFSET (GstDataQueueClass, empty), NULL, NULL,
157 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
160 * GstDataQueue::full: (skip)
161 * @queue: the queue instance
163 * Reports that the queue became full (full).
164 * A queue is full if the total amount of data inside it (num-visible, time,
165 * size) is higher than the boundary values which can be set through the GObject
168 gst_data_queue_signals[SIGNAL_FULL] =
169 g_signal_new ("full", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,
170 G_STRUCT_OFFSET (GstDataQueueClass, full), NULL, NULL,
171 g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
174 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_BYTES,
175 g_param_spec_uint ("current-level-bytes", "Current level (kB)",
176 "Current amount of data in the queue (bytes)",
177 0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
178 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_VISIBLE,
179 g_param_spec_uint ("current-level-visible",
180 "Current level (visible items)",
181 "Current number of visible items in the queue", 0, G_MAXUINT, 0,
182 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
183 g_object_class_install_property (gobject_class, PROP_CUR_LEVEL_TIME,
184 g_param_spec_uint64 ("current-level-time", "Current level (ns)",
185 "Current amount of data in the queue (in ns)", 0, G_MAXUINT64, 0,
186 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
188 gobject_class->finalize = gst_data_queue_finalize;
192 gst_data_queue_init (GstDataQueue * queue)
195 G_TYPE_INSTANCE_GET_PRIVATE (queue, GST_TYPE_DATA_QUEUE,
196 GstDataQueuePrivate);
198 queue->priv->cur_level.visible = 0; /* no content */
199 queue->priv->cur_level.bytes = 0; /* no content */
200 queue->priv->cur_level.time = 0; /* no content */
202 queue->priv->checkfull = NULL;
204 g_mutex_init (&queue->priv->qlock);
205 g_cond_init (&queue->priv->item_add);
206 g_cond_init (&queue->priv->item_del);
207 queue->priv->queue = gst_queue_array_new (50);
209 GST_DEBUG ("initialized queue's not_empty & not_full conditions");
213 * gst_data_queue_new: (skip)
214 * @checkfull: the callback used to tell if the element considers the queue full
216 * @fullcallback: the callback which will be called when the queue is considered full.
217 * @emptycallback: the callback which will be called when the queue is considered empty.
218 * @checkdata: a #gpointer that will be passed to the @checkfull, @fullcallback,
219 * and @emptycallback callbacks.
221 * Creates a new #GstDataQueue. If @fullcallback or @emptycallback are supplied, then
222 * the #GstDataQueue will call the respective callback to signal full or empty condition.
223 * If the callbacks are NULL the #GstDataQueue will instead emit 'full' and 'empty'
226 * Returns: a new #GstDataQueue.
231 gst_data_queue_new (GstDataQueueCheckFullFunction checkfull,
232 GstDataQueueFullCallback fullcallback,
233 GstDataQueueEmptyCallback emptycallback, gpointer checkdata)
237 g_return_val_if_fail (checkfull != NULL, NULL);
239 ret = g_object_new (GST_TYPE_DATA_QUEUE, NULL);
240 ret->priv->checkfull = checkfull;
241 ret->priv->checkdata = checkdata;
242 ret->priv->fullcallback = fullcallback;
243 ret->priv->emptycallback = emptycallback;
249 gst_data_queue_cleanup (GstDataQueue * queue)
251 GstDataQueuePrivate *priv = queue->priv;
253 while (!gst_queue_array_is_empty (priv->queue)) {
254 GstDataQueueItem *item = gst_queue_array_pop_head (priv->queue);
256 /* Just call the destroy notify on the item */
257 item->destroy (item);
259 priv->cur_level.visible = 0;
260 priv->cur_level.bytes = 0;
261 priv->cur_level.time = 0;
264 /* called only once, as opposed to dispose */
266 gst_data_queue_finalize (GObject * object)
268 GstDataQueue *queue = GST_DATA_QUEUE (object);
269 GstDataQueuePrivate *priv = queue->priv;
271 GST_DEBUG ("finalizing queue");
273 gst_data_queue_cleanup (queue);
274 gst_queue_array_free (priv->queue);
276 GST_DEBUG ("free mutex");
277 g_mutex_clear (&priv->qlock);
278 GST_DEBUG ("done free mutex");
280 g_cond_clear (&priv->item_add);
281 g_cond_clear (&priv->item_del);
283 G_OBJECT_CLASS (parent_class)->finalize (object);
287 gst_data_queue_locked_flush (GstDataQueue * queue)
289 GstDataQueuePrivate *priv = queue->priv;
291 STATUS (queue, "before flushing");
292 gst_data_queue_cleanup (queue);
293 STATUS (queue, "after flushing");
294 /* we deleted something... */
295 if (priv->waiting_del)
296 g_cond_signal (&priv->item_del);
299 static inline gboolean
300 gst_data_queue_locked_is_empty (GstDataQueue * queue)
302 GstDataQueuePrivate *priv = queue->priv;
304 return (gst_queue_array_get_length (priv->queue) == 0);
307 static inline gboolean
308 gst_data_queue_locked_is_full (GstDataQueue * queue)
310 GstDataQueuePrivate *priv = queue->priv;
312 return priv->checkfull (queue, priv->cur_level.visible,
313 priv->cur_level.bytes, priv->cur_level.time, priv->checkdata);
317 * gst_data_queue_flush: (skip)
318 * @queue: a #GstDataQueue.
320 * Flushes all the contents of the @queue. Any call to #gst_data_queue_push and
321 * #gst_data_queue_pop will be released.
327 gst_data_queue_flush (GstDataQueue * queue)
329 GST_DEBUG ("queue:%p", queue);
330 GST_DATA_QUEUE_MUTEX_LOCK (queue);
331 gst_data_queue_locked_flush (queue);
332 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
336 * gst_data_queue_is_empty: (skip)
337 * @queue: a #GstDataQueue.
339 * Queries if there are any items in the @queue.
342 * Returns: %TRUE if @queue is empty.
347 gst_data_queue_is_empty (GstDataQueue * queue)
351 GST_DATA_QUEUE_MUTEX_LOCK (queue);
352 res = gst_data_queue_locked_is_empty (queue);
353 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
359 * gst_data_queue_is_full: (skip)
360 * @queue: a #GstDataQueue.
362 * Queries if @queue is full. This check will be done using the
363 * #GstDataQueueCheckFullFunction registered with @queue.
366 * Returns: %TRUE if @queue is full.
371 gst_data_queue_is_full (GstDataQueue * queue)
375 GST_DATA_QUEUE_MUTEX_LOCK (queue);
376 res = gst_data_queue_locked_is_full (queue);
377 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
383 * gst_data_queue_set_flushing: (skip)
384 * @queue: a #GstDataQueue.
385 * @flushing: a #gboolean stating if the queue will be flushing or not.
387 * Sets the queue to flushing state if @flushing is %TRUE. If set to flushing
388 * state, any incoming data on the @queue will be discarded. Any call currently
389 * blocking on #gst_data_queue_push or #gst_data_queue_pop will return straight
390 * away with a return value of %FALSE. While the @queue is in flushing state,
391 * all calls to those two functions will return %FALSE.
398 gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
400 GstDataQueuePrivate *priv = queue->priv;
402 GST_DEBUG ("queue:%p , flushing:%d", queue, flushing);
404 GST_DATA_QUEUE_MUTEX_LOCK (queue);
405 priv->flushing = flushing;
407 /* release push/pop functions */
408 if (priv->waiting_add)
409 g_cond_signal (&priv->item_add);
410 if (priv->waiting_del)
411 g_cond_signal (&priv->item_del);
413 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
417 gst_data_queue_push_force_unlocked (GstDataQueue * queue,
418 GstDataQueueItem * item)
420 GstDataQueuePrivate *priv = queue->priv;
422 gst_queue_array_push_tail (priv->queue, item);
425 priv->cur_level.visible++;
426 priv->cur_level.bytes += item->size;
427 priv->cur_level.time += item->duration;
431 * gst_data_queue_push_force: (skip)
432 * @queue: a #GstDataQueue.
433 * @item: a #GstDataQueueItem.
435 * Pushes a #GstDataQueueItem (or a structure that begins with the same fields)
436 * on the @queue. It ignores if the @queue is full or not and forces the @item
437 * to be pushed anyway.
440 * Note that this function has slightly different semantics than gst_pad_push()
441 * and gst_pad_push_event(): this function only takes ownership of @item and
442 * the #GstMiniObject contained in @item if the push was successful. If %FALSE
443 * is returned, the caller is responsible for freeing @item and its contents.
445 * Returns: %TRUE if the @item was successfully pushed on the @queue.
450 gst_data_queue_push_force (GstDataQueue * queue, GstDataQueueItem * item)
452 GstDataQueuePrivate *priv = queue->priv;
454 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
455 g_return_val_if_fail (item != NULL, FALSE);
457 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
459 STATUS (queue, "before pushing");
460 gst_data_queue_push_force_unlocked (queue, item);
461 STATUS (queue, "after pushing");
462 if (priv->waiting_add)
463 g_cond_signal (&priv->item_add);
465 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
472 GST_DEBUG ("queue:%p, we are flushing", queue);
473 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
479 * gst_data_queue_push: (skip)
480 * @queue: a #GstDataQueue.
481 * @item: a #GstDataQueueItem.
483 * Pushes a #GstDataQueueItem (or a structure that begins with the same fields)
484 * on the @queue. If the @queue is full, the call will block until space is
485 * available, OR the @queue is set to flushing state.
488 * Note that this function has slightly different semantics than gst_pad_push()
489 * and gst_pad_push_event(): this function only takes ownership of @item and
490 * the #GstMiniObject contained in @item if the push was successful. If %FALSE
491 * is returned, the caller is responsible for freeing @item and its contents.
493 * Returns: %TRUE if the @item was successfully pushed on the @queue.
498 gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item)
500 GstDataQueuePrivate *priv = queue->priv;
502 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
503 g_return_val_if_fail (item != NULL, FALSE);
505 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
507 STATUS (queue, "before pushing");
509 /* We ALWAYS need to check for queue fillness */
510 if (gst_data_queue_locked_is_full (queue)) {
511 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
512 if (G_LIKELY (priv->fullcallback))
513 priv->fullcallback (queue, priv->checkdata);
515 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_FULL], 0);
516 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
518 /* signal might have removed some items */
519 while (gst_data_queue_locked_is_full (queue)) {
520 priv->waiting_del = TRUE;
521 g_cond_wait (&priv->item_del, &priv->qlock);
522 priv->waiting_del = FALSE;
528 gst_data_queue_push_force_unlocked (queue, item);
530 STATUS (queue, "after pushing");
531 if (priv->waiting_add)
532 g_cond_signal (&priv->item_add);
534 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
541 GST_DEBUG ("queue:%p, we are flushing", queue);
542 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
548 _gst_data_queue_wait_non_empty (GstDataQueue * queue)
550 GstDataQueuePrivate *priv = queue->priv;
552 while (gst_data_queue_locked_is_empty (queue)) {
553 priv->waiting_add = TRUE;
554 g_cond_wait (&priv->item_add, &priv->qlock);
555 priv->waiting_add = FALSE;
563 * gst_data_queue_pop: (skip)
564 * @queue: a #GstDataQueue.
565 * @item: pointer to store the returned #GstDataQueueItem.
567 * Retrieves the first @item available on the @queue. If the queue is currently
568 * empty, the call will block until at least one item is available, OR the
569 * @queue is set to the flushing state.
572 * Returns: %TRUE if an @item was successfully retrieved from the @queue.
577 gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item)
579 GstDataQueuePrivate *priv = queue->priv;
581 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
582 g_return_val_if_fail (item != NULL, FALSE);
584 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
586 STATUS (queue, "before popping");
588 if (gst_data_queue_locked_is_empty (queue)) {
589 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
590 if (G_LIKELY (priv->emptycallback))
591 priv->emptycallback (queue, priv->checkdata);
593 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_EMPTY], 0);
594 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
596 if (!_gst_data_queue_wait_non_empty (queue))
600 /* Get the item from the GQueue */
601 *item = gst_queue_array_pop_head (priv->queue);
603 /* update current level counter */
604 if ((*item)->visible)
605 priv->cur_level.visible--;
606 priv->cur_level.bytes -= (*item)->size;
607 priv->cur_level.time -= (*item)->duration;
609 STATUS (queue, "after popping");
610 if (priv->waiting_del)
611 g_cond_signal (&priv->item_del);
613 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
620 GST_DEBUG ("queue:%p, we are flushing", queue);
621 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
627 is_of_type (gconstpointer a, gconstpointer b)
629 return !G_TYPE_CHECK_INSTANCE_TYPE (a, GPOINTER_TO_SIZE (b));
633 * gst_data_queue_peek: (skip)
634 * @queue: a #GstDataQueue.
635 * @item: pointer to store the returned #GstDataQueueItem.
637 * Retrieves the first @item available on the @queue without removing it.
638 * If the queue is currently empty, the call will block until at least
639 * one item is available, OR the @queue is set to the flushing state.
642 * Returns: %TRUE if an @item was successfully retrieved from the @queue.
647 gst_data_queue_peek (GstDataQueue * queue, GstDataQueueItem ** item)
649 GstDataQueuePrivate *priv = queue->priv;
651 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
652 g_return_val_if_fail (item != NULL, FALSE);
654 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
656 STATUS (queue, "before peeking");
658 if (gst_data_queue_locked_is_empty (queue)) {
659 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
660 if (G_LIKELY (priv->emptycallback))
661 priv->emptycallback (queue, priv->checkdata);
663 g_signal_emit (queue, gst_data_queue_signals[SIGNAL_EMPTY], 0);
664 GST_DATA_QUEUE_MUTEX_LOCK_CHECK (queue, flushing);
666 if (!_gst_data_queue_wait_non_empty (queue))
670 /* Get the item from the GQueue */
671 *item = gst_queue_array_peek_head (priv->queue);
673 STATUS (queue, "after peeking");
674 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
681 GST_DEBUG ("queue:%p, we are flushing", queue);
682 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
688 * gst_data_queue_drop_head: (skip)
689 * @queue: The #GstDataQueue to drop an item from.
690 * @type: The #GType of the item to drop.
692 * Pop and unref the head-most #GstMiniObject with the given #GType.
694 * Returns: %TRUE if an element was removed.
699 gst_data_queue_drop_head (GstDataQueue * queue, GType type)
701 gboolean res = FALSE;
702 GstDataQueueItem *leak = NULL;
704 GstDataQueuePrivate *priv = queue->priv;
706 g_return_val_if_fail (GST_IS_DATA_QUEUE (queue), FALSE);
708 GST_DEBUG ("queue:%p", queue);
710 GST_DATA_QUEUE_MUTEX_LOCK (queue);
711 idx = gst_queue_array_find (priv->queue, is_of_type, GSIZE_TO_POINTER (type));
716 leak = gst_queue_array_drop_element (priv->queue, idx);
719 priv->cur_level.visible--;
720 priv->cur_level.bytes -= leak->size;
721 priv->cur_level.time -= leak->duration;
723 leak->destroy (leak);
728 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
730 GST_DEBUG ("queue:%p , res:%d", queue, res);
736 * gst_data_queue_limits_changed: (skip)
737 * @queue: The #GstDataQueue
739 * Inform the queue that the limits for the fullness check have changed and that
740 * any blocking gst_data_queue_push() should be unblocked to recheck the limits.
745 gst_data_queue_limits_changed (GstDataQueue * queue)
747 GstDataQueuePrivate *priv = queue->priv;
749 g_return_if_fail (GST_IS_DATA_QUEUE (queue));
751 GST_DATA_QUEUE_MUTEX_LOCK (queue);
752 if (priv->waiting_del) {
753 GST_DEBUG ("signal del");
754 g_cond_signal (&priv->item_del);
756 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);
760 * gst_data_queue_get_level: (skip)
761 * @queue: The #GstDataQueue
762 * @level: the location to store the result
764 * Get the current level of the queue.
769 gst_data_queue_get_level (GstDataQueue * queue, GstDataQueueSize * level)
771 GstDataQueuePrivate *priv = queue->priv;
773 memcpy (level, (&priv->cur_level), sizeof (GstDataQueueSize));
777 gst_data_queue_set_property (GObject * object,
778 guint prop_id, const GValue * value, GParamSpec * pspec)
782 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
788 gst_data_queue_get_property (GObject * object,
789 guint prop_id, GValue * value, GParamSpec * pspec)
791 GstDataQueue *queue = GST_DATA_QUEUE (object);
792 GstDataQueuePrivate *priv = queue->priv;
794 GST_DATA_QUEUE_MUTEX_LOCK (queue);
797 case PROP_CUR_LEVEL_BYTES:
798 g_value_set_uint (value, priv->cur_level.bytes);
800 case PROP_CUR_LEVEL_VISIBLE:
801 g_value_set_uint (value, priv->cur_level.visible);
803 case PROP_CUR_LEVEL_TIME:
804 g_value_set_uint64 (value, priv->cur_level.time);
807 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
811 GST_DATA_QUEUE_MUTEX_UNLOCK (queue);