libgstrtpmanager_la_SOURCES = gstrtpmanager.c \
gstrtpbin.c \
gstrtpclient.c \
- async_jitter_queue.c \
gstrtpjitterbuffer.c \
gstrtpptdemux.c \
gstrtpssrcdemux.c \
+ rtpjitterbuffer.c \
rtpsession.c \
rtpsource.c \
rtpstats.c \
noinst_HEADERS = gstrtpbin.h \
gstrtpclient.h \
- async_jitter_queue.h \
gstrtpjitterbuffer.h \
gstrtpptdemux.h \
gstrtpssrcdemux.h \
+ rtpjitterbuffer.h \
rtpsession.h \
rtpsource.h \
rtpstats.h \
+++ /dev/null
-/*
- * Async Jitter Queue based on g_async_queue
- * This code is GST RTP smart and deals with timestamps
- *
- * Farsight Voice+Video library
- * Copyright 2007 Collabora Ltd,
- * Copyright 2007 Nokia Corporation
- * @author: Philippe Khalaf <philippe.khalaf@collabora.co.uk>.
- *
- * This is an async queue that has a buffering mecanism based on the set low
- * and high threshold. When the lower threshold is reached, the queue will
- * fill itself up until the higher threshold is reached before allowing any
- * pops to occur. This allows a jitterbuffer of at least min threshold items
- * to be available.
- */
-
-/* GLIB - Library of useful routines for C programming
- * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald
- *
- * GAsyncQueue: asynchronous queue implementation, based on Gqueue.
- * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser 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
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser 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.
- */
-
-/*
- * MT safe
- */
-
-#include "config.h"
-
-#include "async_jitter_queue.h"
-
-#include <gst/gst.h>
-#include <gst/rtp/gstrtpbuffer.h>
-
-#define DEFAULT_LOW_THRESHOLD 0.1
-#define DEFAULT_HIGH_THRESHOLD 0.9
-
-struct _AsyncJitterQueue
-{
- GMutex *mutex;
- GCond *cond;
- GQueue *queue;
- guint waiting_threads;
- gint32 ref_count;
- gfloat low_threshold;
- gfloat high_threshold;
- guint32 max_queue_length;
- gboolean buffering;
- gboolean pop_flushing;
- gboolean pop_blocking;
- guint pops_remaining;
- guint32 tail_buffer_duration;
-};
-
-/**
- * async_jitter_queue_new:
- *
- * Creates a new asynchronous queue with the initial reference count of 1.
- *
- * Return value: the new #AsyncJitterQueue.
- **/
-AsyncJitterQueue *
-async_jitter_queue_new (void)
-{
- AsyncJitterQueue *retval = g_new (AsyncJitterQueue, 1);
-
- retval->mutex = g_mutex_new ();
- retval->cond = g_cond_new ();
- retval->queue = g_queue_new ();
- retval->waiting_threads = 0;
- retval->ref_count = 1;
- retval->low_threshold = DEFAULT_LOW_THRESHOLD;
- retval->high_threshold = DEFAULT_HIGH_THRESHOLD;
- retval->buffering = TRUE; /* we need to buffer initially */
- retval->pop_flushing = TRUE;
- retval->pop_blocking = TRUE;
- retval->pops_remaining = 0;
- retval->tail_buffer_duration = 0;
- return retval;
-}
-
-/* checks buffering state and wakes up waiting pops */
-void
-signal_waiting_threads (AsyncJitterQueue * queue)
-{
- if (async_jitter_queue_length_ts_units_unlocked (queue) >=
- queue->high_threshold * queue->max_queue_length) {
- GST_DEBUG ("stop buffering");
- queue->buffering = FALSE;
- }
-
- if (queue->waiting_threads > 0) {
- if (!queue->buffering) {
- g_cond_signal (queue->cond);
- }
- }
-}
-
-/**
- * async_jitter_queue_ref:
- * @queue: a #AsyncJitterQueue.
- *
- * Increases the reference count of the asynchronous @queue by 1. You
- * do not need to hold the lock to call this function.
- *
- * Returns: the @queue that was passed in (since 2.6)
- **/
-AsyncJitterQueue *
-async_jitter_queue_ref (AsyncJitterQueue * queue)
-{
- g_return_val_if_fail (queue, NULL);
- g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
-
- g_atomic_int_inc (&queue->ref_count);
-
- return queue;
-}
-
-/**
- * async_jitter_queue_ref_unlocked:
- * @queue: a #AsyncJitterQueue.
- *
- * Increases the reference count of the asynchronous @queue by 1.
- **/
-void
-async_jitter_queue_ref_unlocked (AsyncJitterQueue * queue)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- g_atomic_int_inc (&queue->ref_count);
-}
-
-/**
- * async_jitter_queue_set_low_threshold:
- * @queue: a #AsyncJitterQueue.
- * @threshold: the lower threshold (fraction of max size)
- *
- * Sets the low threshold on the queue. This threshold indicates the minimum
- * number of items allowed in the queue before we refill it up to the set
- * maximum threshold.
- **/
-void
-async_jitter_queue_set_low_threshold (AsyncJitterQueue * queue,
- gfloat threshold)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- queue->low_threshold = threshold;
-}
-
-/**
- * async_jitter_queue_set_max_threshold:
- * @queue: a #AsyncJitterQueue.
- * @threshold: the higher threshold (fraction of max size)
- *
- * Sets the high threshold on the queue. This threshold indicates the amount of
- * items to fill in the queue before releasing any blocking pop calls. This
- * blocking mecanism is only triggered when we reach the low threshold and must
- * refill the queue.
- **/
-void
-async_jitter_queue_set_high_threshold (AsyncJitterQueue * queue,
- gfloat threshold)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- queue->high_threshold = threshold;
-}
-
-/* set the maximum queue length in RTP timestamp units */
-void
-async_jitter_queue_set_max_queue_length (AsyncJitterQueue * queue,
- guint32 max_length)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- queue->max_queue_length = max_length;
-}
-
-GQueue *
-async_jitter_queue_get_g_queue (AsyncJitterQueue * queue)
-{
- g_return_val_if_fail (queue, NULL);
-
- return queue->queue;
-}
-
-static guint32
-calculate_ts_diff (guint32 high_ts, guint32 low_ts)
-{
- /* it needs to work if ts wraps */
- if (high_ts >= low_ts) {
- return high_ts - low_ts;
- } else {
- return high_ts + G_MAXUINT32 + 1 - low_ts;
- }
-}
-
-/* this function returns the length of the queue in timestamp units. It will
- * also add the duration of the last buffer in the queue */
-/* FIXME This function wrongly assumes that there are no missing packets inside
- * the buffer, in reality it needs to check for gaps and subsctract those from
- * the total */
-guint32
-async_jitter_queue_length_ts_units_unlocked (AsyncJitterQueue * queue)
-{
- guint32 tail_ts;
- guint32 head_ts;
- guint32 ret;
- GstBuffer *head;
- GstBuffer *tail;
-
- g_return_val_if_fail (queue, 0);
-
- if (queue->queue->length < 2) {
- return 0;
- }
-
- tail = g_queue_peek_tail (queue->queue);
- head = g_queue_peek_head (queue->queue);
-
- if (!GST_IS_BUFFER (tail) || !GST_IS_BUFFER (head))
- return 0;
-
- tail_ts = gst_rtp_buffer_get_timestamp (tail);
- head_ts = gst_rtp_buffer_get_timestamp (head);
-
- ret = calculate_ts_diff (head_ts, tail_ts);
-
- /* let's add the duration of the tail buffer */
- ret += queue->tail_buffer_duration;
-
- return ret;
-}
-
-/**
- * async_jitter_queue_unref_and_unlock:
- * @queue: a #AsyncJitterQueue.
- *
- * Decreases the reference count of the asynchronous @queue by 1 and
- * releases the lock. This function must be called while holding the
- * @queue's lock. If the reference count went to 0, the @queue will be
- * destroyed and the memory allocated will be freed.
- **/
-void
-async_jitter_queue_unref_and_unlock (AsyncJitterQueue * queue)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- g_mutex_unlock (queue->mutex);
- async_jitter_queue_unref (queue);
-}
-
-/**
- * async_jitter_queue_unref:
- * @queue: a #AsyncJitterQueue.
- *
- * Decreases the reference count of the asynchronous @queue by 1. If
- * the reference count went to 0, the @queue will be destroyed and the
- * memory allocated will be freed. So you are not allowed to use the
- * @queue afterwards, as it might have disappeared. You do not need to
- * hold the lock to call this function.
- **/
-void
-async_jitter_queue_unref (AsyncJitterQueue * queue)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- if (g_atomic_int_dec_and_test (&queue->ref_count)) {
- g_return_if_fail (queue->waiting_threads == 0);
- g_mutex_free (queue->mutex);
- if (queue->cond)
- g_cond_free (queue->cond);
- g_queue_free (queue->queue);
- g_free (queue);
- }
-}
-
-/**
- * async_jitter_queue_lock:
- * @queue: a #AsyncJitterQueue.
- *
- * Acquires the @queue's lock. After that you can only call the
- * <function>async_jitter_queue_*_unlocked()</function> function variants on that
- * @queue. Otherwise it will deadlock.
- **/
-void
-async_jitter_queue_lock (AsyncJitterQueue * queue)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- g_mutex_lock (queue->mutex);
-}
-
-/**
- * async_jitter_queue_unlock:
- * @queue: a #AsyncJitterQueue.
- *
- * Releases the queue's lock.
- **/
-void
-async_jitter_queue_unlock (AsyncJitterQueue * queue)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- g_mutex_unlock (queue->mutex);
-}
-
-/**
- * async_jitter_queue_push:
- * @queue: a #AsyncJitterQueue.
- * @data: @data to push into the @queue.
- *
- * Pushes the @data into the @queue. @data must not be %NULL.
- **/
-void
-async_jitter_queue_push (AsyncJitterQueue * queue, gpointer data)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
- g_return_if_fail (data);
-
- g_mutex_lock (queue->mutex);
- async_jitter_queue_push_unlocked (queue, data);
- g_mutex_unlock (queue->mutex);
-}
-
-/**
- * async_jitter_queue_push_unlocked:
- * @queue: a #AsyncJitterQueue.
- * @data: @data to push into the @queue.
- *
- * Pushes the @data into the @queue. @data must not be %NULL. This
- * function must be called while holding the @queue's lock.
- **/
-void
-async_jitter_queue_push_unlocked (AsyncJitterQueue * queue, gpointer data)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
- g_return_if_fail (data);
-
- g_queue_push_head (queue->queue, data);
-
- signal_waiting_threads (queue);
-}
-
-/**
- * async_jitter_queue_push_sorted:
- * @queue: a #AsyncJitterQueue
- * @data: the @data to push into the @queue
- * @func: the #GCompareDataFunc is used to sort @queue. This function
- * is passed two elements of the @queue. The function should return
- * 0 if they are equal, a negative value if the first element
- * should be higher in the @queue or a positive value if the first
- * element should be lower in the @queue than the second element.
- * @user_data: user data passed to @func.
- *
- * Inserts @data into @queue using @func to determine the new
- * position.
- *
- * This function requires that the @queue is sorted before pushing on
- * new elements.
- *
- * This function will lock @queue before it sorts the queue and unlock
- * it when it is finished.
- *
- * For an example of @func see async_jitter_queue_sort().
- *
- * Since: 2.10
- **/
-gboolean
-async_jitter_queue_push_sorted (AsyncJitterQueue * queue,
- gpointer data, GCompareDataFunc func, gpointer user_data)
-{
- gboolean ret;
-
- g_return_val_if_fail (queue != NULL, FALSE);
-
- g_mutex_lock (queue->mutex);
- ret = async_jitter_queue_push_sorted_unlocked (queue, data, func, user_data);
- g_mutex_unlock (queue->mutex);
-
- return ret;
-}
-
-/**
- * async_jitter_queue_push_sorted_unlocked:
- * @queue: a #AsyncJitterQueue
- * @data: the @data to push into the @queue
- * @func: the #GCompareDataFunc is used to sort @queue. This function
- * is passed two elements of the @queue. The function should return
- * 0 if they are equal, a negative value if the first element
- * should be higher in the @queue or a positive value if the first
- * element should be lower in the @queue than the second element.
- * @user_data: user data passed to @func.
- *
- * Inserts @data into @queue using @func to determine the new
- * position.
- *
- * This function requires that the @queue is sorted before pushing on
- * new elements.
- *
- * If @GCompareDataFunc returns 0, this function does not insert @data and
- * return FALSE.
- *
- * This function is called while holding the @queue's lock.
- *
- * For an example of @func see async_jitter_queue_sort().
- *
- * Since: 2.10
- **/
-gboolean
-async_jitter_queue_push_sorted_unlocked (AsyncJitterQueue * queue,
- gpointer data, GCompareDataFunc func, gpointer user_data)
-{
- GList *list;
- gint func_ret = TRUE;
-
- g_return_val_if_fail (queue != NULL, FALSE);
-
- list = queue->queue->head;
- while (list && (func_ret = func (list->data, data, user_data)) < 0)
- list = list->next;
-
- if (func_ret == 0) {
- return FALSE;
- }
- if (list) {
- g_queue_insert_before (queue->queue, list, data);
- } else {
- g_queue_push_tail (queue->queue, data);
- }
-
- signal_waiting_threads (queue);
- return TRUE;
-}
-
-void
-async_jitter_queue_insert_after_unlocked (AsyncJitterQueue * queue,
- GList * sibling, gpointer data)
-{
- g_return_if_fail (queue != NULL);
-
- g_queue_insert_before (queue->queue, sibling, data);
-
- signal_waiting_threads (queue);
-}
-
-static gpointer
-async_jitter_queue_pop_intern_unlocked (AsyncJitterQueue * queue)
-{
- gpointer retval;
- GstBuffer *tail_buffer = NULL;
- guint tsunits;
-
- if (queue->pop_flushing)
- return NULL;
-
- while (queue->pop_blocking) {
- queue->waiting_threads++;
- g_cond_wait (queue->cond, queue->mutex);
- queue->waiting_threads--;
- if (queue->pop_flushing)
- return NULL;
- }
-
-
- tsunits = async_jitter_queue_length_ts_units_unlocked (queue);
-
- GST_DEBUG ("tsunits %u, pops: %u, limit %d", tsunits, queue->pops_remaining,
- (int) (queue->low_threshold * queue->max_queue_length));
-
- if (tsunits <= queue->low_threshold * queue->max_queue_length
- && queue->pops_remaining == 0) {
- if (!queue->buffering) {
- GST_DEBUG ("start buffering");
- queue->buffering = TRUE;
- queue->pops_remaining = queue->queue->length;
- }
-
- GST_DEBUG ("wait for data");
- while (!g_queue_peek_tail (queue->queue) || queue->pop_blocking) {
- queue->waiting_threads++;
- g_cond_wait (queue->cond, queue->mutex);
- queue->waiting_threads--;
- if (queue->pop_flushing)
- return NULL;
- }
- }
-
- retval = g_queue_pop_tail (queue->queue);
- if (queue->pops_remaining)
- queue->pops_remaining--;
-
- tail_buffer = g_queue_peek_tail (queue->queue);
- if (tail_buffer) {
- if (!GST_IS_BUFFER (tail_buffer) || !GST_IS_BUFFER (retval)) {
- queue->tail_buffer_duration = 0;
- } else if (gst_rtp_buffer_get_seq (tail_buffer)
- - gst_rtp_buffer_get_seq (retval) == 1) {
- queue->tail_buffer_duration =
- calculate_ts_diff (gst_rtp_buffer_get_timestamp (tail_buffer),
- gst_rtp_buffer_get_timestamp (retval));
- } else {
- /* There is a sequence number gap -> we can't calculate the duration
- * let's just set it to 0 */
- queue->tail_buffer_duration = 0;
- }
- }
-
- g_assert (retval);
-
- return retval;
-}
-
-/**
- * async_jitter_queue_pop:
- * @queue: a #AsyncJitterQueue.
- *
- * Pops data from the @queue. This function blocks until data become
- * available. If pop is disabled, tis function return NULL.
- *
- * Return value: data from the queue.
- **/
-gpointer
-async_jitter_queue_pop (AsyncJitterQueue * queue)
-{
- gpointer retval;
-
- g_return_val_if_fail (queue, NULL);
- g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
-
- g_mutex_lock (queue->mutex);
- retval = async_jitter_queue_pop_intern_unlocked (queue);
- g_mutex_unlock (queue->mutex);
-
- return retval;
-}
-
-/**
- * async_jitter_queue_pop_unlocked:
- * @queue: a #AsyncJitterQueue.
- *
- * Pops data from the @queue. This function blocks until data become
- * available. This function must be called while holding the @queue's
- * lock.
- *
- * Return value: data from the queue.
- **/
-gpointer
-async_jitter_queue_pop_unlocked (AsyncJitterQueue * queue)
-{
- g_return_val_if_fail (queue, NULL);
- g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
-
- return async_jitter_queue_pop_intern_unlocked (queue);
-}
-
-/**
- * async_jitter_queue_length:
- * @queue: a #AsyncJitterQueue.
- *
- * Returns the length of the queue
- * Return value: the length of the @queue.
- **/
-gint
-async_jitter_queue_length (AsyncJitterQueue * queue)
-{
- gint retval;
-
- g_return_val_if_fail (queue, 0);
- g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
-
- g_mutex_lock (queue->mutex);
- retval = queue->queue->length;
- g_mutex_unlock (queue->mutex);
-
- return retval;
-}
-
-/**
- * async_jitter_queue_length_unlocked:
- * @queue: a #AsyncJitterQueue.
- *
- * Returns the length of the queue.
- *
- * Return value: the length of the @queue.
- **/
-gint
-async_jitter_queue_length_unlocked (AsyncJitterQueue * queue)
-{
- g_return_val_if_fail (queue, 0);
- g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
-
- return queue->queue->length;
-}
-
-/**
- * async_jitter_queue_set_flushing_unlocked:
- * @queue: a #AsyncJitterQueue.
- * @free_func: a function to call to free the elements
- * @user_data: user data passed to @free_func
- *
- * This function is used to set/unset flushing. If flushing is set any
- * waiting/blocked pops will be unblocked. Any subsequent calls to pop will
- * return NULL. Flushing is set by default.
- */
-void
-async_jitter_queue_set_flushing_unlocked (AsyncJitterQueue * queue,
- GFunc free_func, gpointer user_data)
-{
- gpointer elem;
-
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- queue->pop_flushing = TRUE;
- /* let's unblock any remaining pops */
- if (queue->waiting_threads > 0)
- g_cond_broadcast (queue->cond);
- /* free data from queue */
- while ((elem = g_queue_pop_head (queue->queue)))
- free_func (elem, user_data);
-}
-
-/**
- * async_jitter_queue_unset_flushing_unlocked:
- * @queue: a #AsyncJitterQueue.
- * @free_func: a function to call to free the elements
- * @user_data: user data passed to @free_func
- *
- * This function is used to set/unset flushing. If flushing is set any
- * waiting/blocked pops will be unblocked. Any subsequent calls to pop will
- * return NULL. Flushing is set by default.
- */
-void
-async_jitter_queue_unset_flushing_unlocked (AsyncJitterQueue * queue)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- queue->pop_flushing = FALSE;
- /* let's unblock any remaining pops */
- if (queue->waiting_threads > 0)
- g_cond_broadcast (queue->cond);
-}
-
-/**
- * async_jitter_queue_set_blocking_unlocked:
- * @queue: a #AsyncJitterQueue.
- * @enabled: a boolean to enable/disable blocking
- *
- * This function is used to enable/disable blocking. If blocking is enabled any
- * pops will be blocked until the queue is unblocked. The queue is blocked by
- * default.
- */
-void
-async_jitter_queue_set_blocking_unlocked (AsyncJitterQueue * queue,
- gboolean blocking)
-{
- g_return_if_fail (queue);
- g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
-
- queue->pop_blocking = blocking;
- /* let's unblock any remaining pops */
- if (queue->waiting_threads > 0)
- g_cond_broadcast (queue->cond);
-}
+++ /dev/null
-/* Async Jitter Queue based on g_async_queue
- *
- * Farsight Voice+Video library
- * Copyright 2007 Collabora Ltd,
- * Copyright 2007 Nokia Corporation
- * @author: Philippe Khalaf <philippe.khalaf@collabora.co.uk>.
- */
-
-/* GLIB - Library of useful routines for C programming
- * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser 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
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser 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.
- */
-
-/*
- * Modified by the GLib Team and others 1997-2000. See the AUTHORS
- * file for a list of people on the GLib Team. See the ChangeLog
- * files for a list of changes. These files are distributed with
- * GLib at ftp://ftp.gtk.org/pub/gtk/.
- */
-
-#ifndef __ASYNCJITTERQUEUE_H__
-#define __ASYNCJITTERQUEUE_H__
-
-#include <glib.h>
-#include <glib/gthread.h>
-
-G_BEGIN_DECLS
-
-typedef struct _AsyncJitterQueue AsyncJitterQueue;
-
-/* Asyncronous Queues, can be used to communicate between threads
- */
-
-/* Get a new AsyncJitterQueue with the ref_count 1 */
-AsyncJitterQueue* async_jitter_queue_new (void);
-
-/* Lock and unlock a AsyncJitterQueue. All functions lock the queue for
- * themselves, but in certain cirumstances you want to hold the lock longer,
- * thus you lock the queue, call the *_unlocked functions and unlock it again.
- */
-void async_jitter_queue_lock (AsyncJitterQueue *queue);
-void async_jitter_queue_unlock (AsyncJitterQueue *queue);
-
-/* Ref and unref the AsyncJitterQueue. */
-AsyncJitterQueue* async_jitter_queue_ref (AsyncJitterQueue *queue);
-void async_jitter_queue_unref (AsyncJitterQueue *queue);
-#ifndef G_DISABLE_DEPRECATED
-/* You don't have to hold the lock for calling *_ref and *_unref anymore. */
-void async_jitter_queue_ref_unlocked (AsyncJitterQueue *queue);
-void async_jitter_queue_unref_and_unlock (AsyncJitterQueue *queue);
-#endif /* !G_DISABLE_DEPRECATED */
-
-void async_jitter_queue_set_low_threshold (AsyncJitterQueue *queue,
- gfloat threshold);
-void async_jitter_queue_set_high_threshold (AsyncJitterQueue *queue,
- gfloat threshold);
-
-void async_jitter_queue_set_max_queue_length (AsyncJitterQueue *queue,
- guint32 max_length);
-
-/* Push data into the async queue. Must not be NULL. */
-void async_jitter_queue_push (AsyncJitterQueue *queue,
- gpointer data);
-void async_jitter_queue_push_unlocked (AsyncJitterQueue *queue,
- gpointer data);
-gboolean async_jitter_queue_push_sorted (AsyncJitterQueue *queue,
- gpointer data,
- GCompareDataFunc func,
- gpointer user_data);
-
-void async_jitter_queue_insert_after_unlocked(AsyncJitterQueue *queue,
- GList *sibling,
- gpointer data);
-
-gboolean async_jitter_queue_push_sorted_unlocked(AsyncJitterQueue *queue,
- gpointer data,
- GCompareDataFunc func,
- gpointer user_data);
-
-/* Pop data from the async queue. When no data is there, the thread is blocked
- * until data arrives. */
-gpointer async_jitter_queue_pop (AsyncJitterQueue *queue);
-gpointer async_jitter_queue_pop_unlocked (AsyncJitterQueue *queue);
-
-/* Try to pop data. NULL is returned in case of empty queue. */
-gpointer async_jitter_queue_try_pop (AsyncJitterQueue *queue);
-gpointer async_jitter_queue_try_pop_unlocked (AsyncJitterQueue *queue);
-
-/* Wait for data until at maximum until end_time is reached. NULL is returned
- * in case of empty queue. */
-gpointer async_jitter_queue_timed_pop (AsyncJitterQueue *queue,
- GTimeVal *end_time);
-gpointer async_jitter_queue_timed_pop_unlocked (AsyncJitterQueue *queue,
- GTimeVal *end_time);
-
-/* Return the length of the queue. Negative values mean that threads
- * are waiting, positve values mean that there are entries in the
- * queue. Actually this function returns the length of the queue minus
- * the number of waiting threads, async_jitter_queue_length == 0 could also
- * mean 'n' entries in the queue and 'n' thread waiting. Such can
- * happen due to locking of the queue or due to scheduling. */
-gint async_jitter_queue_length (AsyncJitterQueue *queue);
-gint async_jitter_queue_length_unlocked (AsyncJitterQueue *queue);
-
-void async_jitter_queue_set_flushing_unlocked (AsyncJitterQueue* queue,
- GFunc free_func, gpointer user_data);
-void async_jitter_queue_unset_flushing_unlocked (AsyncJitterQueue* queue);
-void async_jitter_queue_set_blocking_unlocked (AsyncJitterQueue* queue,
- gboolean blocking);
-guint32
-async_jitter_queue_length_ts_units_unlocked (AsyncJitterQueue *queue);
-
-G_END_DECLS
-
-#endif /* __ASYNCJITTERQUEUE_H__ */
-
BOXED:UINT
BOXED:UINT,UINT
VOID:UINT,OBJECT
+VOID:UINT,UINT
{
SIGNAL_REQUEST_PT_MAP,
SIGNAL_CLEAR_PT_MAP,
+
+ SIGNAL_ON_NEW_SSRC,
+ SIGNAL_ON_SSRC_COLLISION,
+ SIGNAL_ON_SSRC_VALIDATED,
+ SIGNAL_ON_BYE_SSRC,
+ SIGNAL_ON_BYE_TIMEOUT,
+ SIGNAL_ON_TIMEOUT,
LAST_SIGNAL
};
return NULL;
}
+static void
+on_new_ssrc (GstElement * session, guint32 ssrc, GstRTPBinSession * sess)
+{
+ g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_NEW_SSRC], 0,
+ sess->id, ssrc);
+}
+
+static void
+on_ssrc_collision (GstElement * session, guint32 ssrc, GstRTPBinSession * sess)
+{
+ g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_COLLISION], 0,
+ sess->id, ssrc);
+}
+
+static void
+on_ssrc_validated (GstElement * session, guint32 ssrc, GstRTPBinSession * sess)
+{
+ g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_VALIDATED], 0,
+ sess->id, ssrc);
+}
+
+static void
+on_bye_ssrc (GstElement * session, guint32 ssrc, GstRTPBinSession * sess)
+{
+ g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_BYE_SSRC], 0,
+ sess->id, ssrc);
+}
+
+static void
+on_bye_timeout (GstElement * session, guint32 ssrc, GstRTPBinSession * sess)
+{
+ g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_BYE_TIMEOUT], 0,
+ sess->id, ssrc);
+}
+
+static void
+on_timeout (GstElement * session, guint32 ssrc, GstRTPBinSession * sess)
+{
+ g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_TIMEOUT], 0,
+ sess->id, ssrc);
+}
+
/* create a session with the given id. Must be called with RTP_BIN_LOCK */
static GstRTPBinSession *
create_session (GstRTPBin * rtpbin, gint id)
g_signal_connect (session, "request-pt-map",
(GCallback) pt_map_requested, sess);
+ g_signal_connect (sess->session, "on-new-ssrc",
+ (GCallback) on_new_ssrc, sess);
+ g_signal_connect (sess->session, "on-ssrc-collision",
+ (GCallback) on_ssrc_collision, sess);
+ g_signal_connect (sess->session, "on-ssrc-validated",
+ (GCallback) on_ssrc_validated, sess);
+ g_signal_connect (sess->session, "on-bye-ssrc",
+ (GCallback) on_bye_ssrc, sess);
+ g_signal_connect (sess->session, "on-bye-timeout",
+ (GCallback) on_bye_timeout, sess);
+ g_signal_connect (sess->session, "on-timeout", (GCallback) on_timeout, sess);
+
gst_bin_add (GST_BIN_CAST (rtpbin), session);
gst_element_set_state (session, GST_STATE_PLAYING);
gst_bin_add (GST_BIN_CAST (rtpbin), demux);
G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRTPBinClass, clear_pt_map),
NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
+ /**
+ * GstRTPBin::on-new-ssrc:
+ * @rtpbin: the object which received the signal
+ * @session: the session
+ * @ssrc: the SSRC
+ *
+ * Notify of a new SSRC that entered @session.
+ */
+ gst_rtp_bin_signals[SIGNAL_ON_NEW_SSRC] =
+ g_signal_new ("on-new-ssrc", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPBinClass, on_new_ssrc),
+ NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_UINT, G_TYPE_NONE, 2,
+ G_TYPE_UINT, G_TYPE_UINT);
+ /**
+ * GstRTPBin::on-ssrc_collision:
+ * @rtpbin: the object which received the signal
+ * @session: the session
+ * @ssrc: the SSRC
+ *
+ * Notify when we have an SSRC collision
+ */
+ gst_rtp_bin_signals[SIGNAL_ON_SSRC_COLLISION] =
+ g_signal_new ("on-ssrc-collision", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPBinClass, on_ssrc_collision),
+ NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_UINT, G_TYPE_NONE, 2,
+ G_TYPE_UINT, G_TYPE_UINT);
+ /**
+ * GstRTPBin::on-ssrc_validated:
+ * @rtpbin: the object which received the signal
+ * @session: the session
+ * @ssrc: the SSRC
+ *
+ * Notify of a new SSRC that became validated.
+ */
+ gst_rtp_bin_signals[SIGNAL_ON_SSRC_VALIDATED] =
+ g_signal_new ("on-ssrc-validated", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPBinClass, on_ssrc_validated),
+ NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_UINT, G_TYPE_NONE, 2,
+ G_TYPE_UINT, G_TYPE_UINT);
+
+ /**
+ * GstRTPBin::on-bye-ssrc:
+ * @rtpbin: the object which received the signal
+ * @session: the session
+ * @ssrc: the SSRC
+ *
+ * Notify of an SSRC that became inactive because of a BYE packet.
+ */
+ gst_rtp_bin_signals[SIGNAL_ON_BYE_SSRC] =
+ g_signal_new ("on-bye-ssrc", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPBinClass, on_bye_ssrc),
+ NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_UINT, G_TYPE_NONE, 2,
+ G_TYPE_UINT, G_TYPE_UINT);
+ /**
+ * GstRTPBin::on-bye-timeout:
+ * @rtpbin: the object which received the signal
+ * @session: the session
+ * @ssrc: the SSRC
+ *
+ * Notify of an SSRC that has timed out because of BYE
+ */
+ gst_rtp_bin_signals[SIGNAL_ON_BYE_TIMEOUT] =
+ g_signal_new ("on-bye-timeout", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPBinClass, on_bye_timeout),
+ NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_UINT, G_TYPE_NONE, 2,
+ G_TYPE_UINT, G_TYPE_UINT);
+ /**
+ * GstRTPBin::on-timeout:
+ * @rtpbin: the object which received the signal
+ * @session: the session
+ * @ssrc: the SSRC
+ *
+ * Notify of an SSRC that has timed out
+ */
+ gst_rtp_bin_signals[SIGNAL_ON_TIMEOUT] =
+ g_signal_new ("on-timeout", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPBinClass, on_timeout),
+ NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_UINT, G_TYPE_NONE, 2,
+ G_TYPE_UINT, G_TYPE_UINT);
+
gstelement_class->provide_clock =
GST_DEBUG_FUNCPTR (gst_rtp_bin_provide_clock);
gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_rtp_bin_change_state);
/* get the caps for pt */
GstCaps* (*request_pt_map) (GstRTPBin *rtpbin, guint session, guint pt);
-
void (*clear_pt_map) (GstRTPBin *rtpbin);
+
+ void (*on_new_ssrc) (GstRTPBin *rtpbin, guint session, guint32 ssrc);
+ void (*on_ssrc_collision) (GstRTPBin *rtpbin, guint session, guint32 ssrc);
+ void (*on_ssrc_validated) (GstRTPBin *rtpbin, guint session, guint32 ssrc);
+ void (*on_bye_ssrc) (GstRTPBin *rtpbin, guint session, guint32 ssrc);
+ void (*on_bye_timeout) (GstRTPBin *rtpbin, guint session, guint32 ssrc);
+ void (*on_timeout) (GstRTPBin *rtpbin, guint session, guint32 ssrc);
};
GType gst_rtp_bin_get_type (void);
* Copyright 2007 Collabora Ltd,
* Copyright 2007 Nokia Corporation
* @author: Philippe Kalaf <philippe.kalaf@collabora.co.uk>.
- * Copyright 2007 Wim Taymans <wim@fluendo.com>
+ * Copyright 2007 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
#include "gstrtpbin-marshal.h"
#include "gstrtpjitterbuffer.h"
-#include "async_jitter_queue.h"
+#include "rtpjitterbuffer.h"
GST_DEBUG_CATEGORY (rtpjitterbuffer_debug);
#define GST_CAT_DEFAULT (rtpjitterbuffer_debug)
"Filter/Network/RTP",
"A buffer that deals with network jitter and other transmission faults",
"Philippe Kalaf <philippe.kalaf@collabora.co.uk>, "
- "Wim Taymans <wim@fluendo.com>");
+ "Wim Taymans <wim.taymans@gmail.com>");
/* RTPJitterBuffer signals and args */
enum
PROP_DROP_ON_LATENCY
};
+#define JBUF_LOCK(priv) (g_mutex_lock ((priv)->jbuf_lock))
+
+#define JBUF_LOCK_CHECK(priv,label) G_STMT_START { \
+ JBUF_LOCK (priv); \
+ if (priv->srcresult != GST_FLOW_OK) \
+ goto label; \
+} G_STMT_END
+
+#define JBUF_UNLOCK(priv) (g_mutex_unlock ((priv)->jbuf_lock))
+#define JBUF_WAIT(priv) (g_cond_wait ((priv)->jbuf_cond, (priv)->jbuf_lock))
+
+#define JBUF_WAIT_CHECK(priv,label) G_STMT_START { \
+ JBUF_WAIT(priv); \
+ if (priv->srcresult != GST_FLOW_OK) \
+ goto label; \
+} G_STMT_END
+
+#define JBUF_SIGNAL(priv) (g_cond_signal ((priv)->jbuf_cond))
+
struct _GstRTPJitterBufferPrivate
{
GstPad *sinkpad, *srcpad;
- AsyncJitterQueue *queue;
+ RTPJitterBuffer *jbuf;
+ GMutex *jbuf_lock;
+ GCond *jbuf_cond;
/* properties */
guint latency_ms;
/* the next expected seqnum */
guint32 next_seqnum;
+ /* state */
+ gboolean eos;
+
/* clock rate and rtp timestamp offset */
gint32 clock_rate;
gint64 clock_base;
/* when we are shutting down */
GstFlowReturn srcresult;
+ gboolean blocked;
/* for sync */
GstSegment segment;
priv->latency_ms = DEFAULT_LATENCY_MS;
priv->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
- priv->queue = async_jitter_queue_new ();
- async_jitter_queue_set_low_threshold (priv->queue, LOW_THRESHOLD);
- async_jitter_queue_set_high_threshold (priv->queue, HIGH_THRESHOLD);
+ priv->jbuf = rtp_jitter_buffer_new ();
+ priv->jbuf_lock = g_mutex_new ();
+ priv->jbuf_cond = g_cond_new ();
priv->waiting_seqnum = -1;
GstRTPJitterBuffer *jitterbuffer;
jitterbuffer = GST_RTP_JITTER_BUFFER (object);
- if (jitterbuffer->priv->queue) {
- async_jitter_queue_unref (jitterbuffer->priv->queue);
- jitterbuffer->priv->queue = NULL;
+ if (jitterbuffer->priv->jbuf) {
+ g_object_unref (jitterbuffer->priv->jbuf);
+ jitterbuffer->priv->jbuf = NULL;
}
G_OBJECT_CLASS (parent_class)->dispose (object);
} else
priv->next_seqnum = -1;
- async_jitter_queue_set_max_queue_length (priv->queue,
- priv->latency_ms * priv->clock_rate / 1000);
-
return TRUE;
/* ERRORS */
}
static void
-free_func (gpointer data, GstRTPJitterBuffer * user_data)
-{
- if (GST_IS_BUFFER (data))
- gst_buffer_unref (GST_BUFFER_CAST (data));
- else
- gst_event_unref (GST_EVENT_CAST (data));
-}
-
-static void
gst_rtp_jitter_buffer_flush_start (GstRTPJitterBuffer * jitterbuffer)
{
GstRTPJitterBufferPrivate *priv;
priv = jitterbuffer->priv;
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK (priv);
/* mark ourselves as flushing */
priv->srcresult = GST_FLOW_WRONG_STATE;
GST_DEBUG_OBJECT (jitterbuffer, "Disabling pop on queue");
/* this unblocks any waiting pops on the src pad task */
- async_jitter_queue_set_flushing_unlocked (jitterbuffer->priv->queue,
- (GFunc) free_func, jitterbuffer);
+ JBUF_SIGNAL (priv);
+ rtp_jitter_buffer_flush (priv->jbuf);
/* unlock clock, we just unschedule, the entry will be released by the
* locking streaming thread. */
if (priv->clock_id)
gst_clock_id_unschedule (priv->clock_id);
-
- async_jitter_queue_unlock (priv->queue);
+ JBUF_UNLOCK (priv);
}
static void
priv = jitterbuffer->priv;
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK (priv);
GST_DEBUG_OBJECT (jitterbuffer, "Enabling pop on queue");
/* Mark as non flushing */
priv->srcresult = GST_FLOW_OK;
priv->last_popped_seqnum = -1;
priv->next_seqnum = -1;
priv->clock_rate = -1;
- /* allow pops from the src pad task */
- async_jitter_queue_unset_flushing_unlocked (jitterbuffer->priv->queue);
- async_jitter_queue_unlock (priv->queue);
+ priv->eos = FALSE;
+ JBUF_UNLOCK (priv);
}
static gboolean
case GST_STATE_CHANGE_NULL_TO_READY:
break;
case GST_STATE_CHANGE_READY_TO_PAUSED:
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK (priv);
/* reset negotiated values */
priv->clock_rate = -1;
priv->clock_base = -1;
/* block until we go to PLAYING */
- async_jitter_queue_set_blocking_unlocked (jitterbuffer->priv->queue,
- TRUE);
- async_jitter_queue_unlock (priv->queue);
+ priv->blocked = TRUE;
+ JBUF_UNLOCK (priv);
break;
case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK (priv);
/* unblock to allow streaming in PLAYING */
- async_jitter_queue_set_blocking_unlocked (jitterbuffer->priv->queue,
- FALSE);
- async_jitter_queue_unlock (priv->queue);
+ priv->blocked = FALSE;
+ JBUF_SIGNAL (priv);
+ JBUF_UNLOCK (priv);
break;
default:
break;
ret = GST_STATE_CHANGE_NO_PREROLL;
break;
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK (priv);
/* block to stop streaming when PAUSED */
- async_jitter_queue_set_blocking_unlocked (jitterbuffer->priv->queue,
- TRUE);
- async_jitter_queue_unlock (priv->queue);
+ priv->blocked = TRUE;
+ JBUF_UNLOCK (priv);
if (ret != GST_STATE_CHANGE_FAILURE)
ret = GST_STATE_CHANGE_NO_PREROLL;
break;
}
}
-/**
- * gets the seqnum from the buffers and compare them
- */
-static gint
-compare_rtp_buffers_seq_num (GstBuffer * a, GstBuffer * b)
-{
- gint ret;
-
- if (GST_IS_BUFFER (a) && GST_IS_BUFFER (b)) {
- /* two buffers */
- ret = priv_compare_rtp_seq_lt
- (gst_rtp_buffer_get_seq (GST_BUFFER_CAST (a)),
- gst_rtp_buffer_get_seq (GST_BUFFER_CAST (b)));
- } else {
- /* one of them is an event, the event always goes before the other element
- * so we return -1. */
- if (GST_IS_EVENT (a))
- ret = -1;
- else
- ret = 1;
- }
- return ret;
-}
-
static gboolean
gst_rtp_jitter_buffer_sink_event (GstPad * pad, GstEvent * event)
{
case GST_EVENT_EOS:
{
/* push EOS in queue. We always push it at the head */
- async_jitter_queue_lock (priv->queue);
- GST_DEBUG_OBJECT (jitterbuffer, "queuing EOS");
+ JBUF_LOCK (priv);
/* check for flushing, we need to discard the event and return FALSE when
* we are flushing */
ret = priv->srcresult == GST_FLOW_OK;
- if (ret)
- async_jitter_queue_push_unlocked (priv->queue, event);
- else
+ if (ret) {
+ GST_DEBUG_OBJECT (jitterbuffer, "queuing EOS");
+ priv->eos = TRUE;
+ JBUF_SIGNAL (priv);
+ } else {
+ GST_DEBUG_OBJECT (jitterbuffer, "dropping EOS, reason %s",
+ gst_flow_get_name (priv->srcresult));
gst_event_unref (event);
- async_jitter_queue_unlock (priv->queue);
+ }
+ JBUF_UNLOCK (priv);
break;
}
default:
GstRTPJitterBuffer *jitterbuffer;
GstRTPJitterBufferPrivate *priv;
guint16 seqnum;
- GstFlowReturn ret;
+ GstFlowReturn ret = GST_FLOW_OK;
jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
seqnum = gst_rtp_buffer_get_seq (buffer);
GST_DEBUG_OBJECT (jitterbuffer, "Received packet #%d", seqnum);
- async_jitter_queue_lock (priv->queue);
- ret = priv->srcresult;
- if (ret != GST_FLOW_OK)
- goto out_flushing;
+ JBUF_LOCK_CHECK (priv, out_flushing);
+ /* don't accept more data on EOS */
+ if (priv->eos)
+ goto have_eos;
/* let's check if this buffer is too late, we cannot accept packets with
* bigger seqnum than the one we already pushed. */
/* let's drop oldest packet if the queue is already full and drop-on-latency
* is set. */
if (priv->drop_on_latency) {
- if (async_jitter_queue_length_ts_units_unlocked (priv->queue) >=
- priv->latency_ms * priv->clock_rate / 1000) {
+ guint64 latency_ts;
+
+ latency_ts =
+ gst_util_uint64_scale_int (priv->latency_ms, priv->clock_rate, 1000);
+
+ if (rtp_jitter_buffer_get_ts_diff (priv->jbuf) >= latency_ts) {
GstBuffer *old_buf;
GST_DEBUG_OBJECT (jitterbuffer, "Queue full, dropping old packet #%d",
seqnum);
- old_buf = async_jitter_queue_pop_unlocked (priv->queue);
+ old_buf = rtp_jitter_buffer_pop (priv->jbuf);
gst_buffer_unref (old_buf);
}
}
/* now insert the packet into the queue in sorted order. This function returns
* FALSE if a packet with the same seqnum was already in the queue, meaning we
* have a duplicate. */
- if (!async_jitter_queue_push_sorted_unlocked (priv->queue, buffer,
- (GCompareDataFunc) compare_rtp_buffers_seq_num, NULL))
+ if (!rtp_jitter_buffer_insert (priv->jbuf, buffer))
goto duplicate;
+ /* signal addition of new buffer */
+ JBUF_SIGNAL (priv);
+
/* let's unschedule and unblock any waiting buffers. We only want to do this
* if there is a currently waiting newer (> seqnum) buffer */
if (priv->clock_id) {
}
}
- GST_DEBUG_OBJECT (jitterbuffer, "Pushed packet #%d on queue %d",
- seqnum, async_jitter_queue_length_unlocked (priv->queue));
+ GST_DEBUG_OBJECT (jitterbuffer, "Pushed packet #%d, now %d packets",
+ seqnum, rtp_jitter_buffer_num_packets (priv->jbuf));
finished:
- async_jitter_queue_unlock (priv->queue);
+ JBUF_UNLOCK (priv);
gst_object_unref (jitterbuffer);
}
out_flushing:
{
+ ret = priv->srcresult;
GST_DEBUG_OBJECT (jitterbuffer, "flushing %s", gst_flow_get_name (ret));
gst_buffer_unref (buffer);
goto finished;
}
+have_eos:
+ {
+ ret = GST_FLOW_UNEXPECTED;
+ GST_DEBUG_OBJECT (jitterbuffer, "we are EOS, refusing buffer");
+ gst_buffer_unref (buffer);
+ goto finished;
+ }
too_late:
{
GST_DEBUG_OBJECT (jitterbuffer, "Packet #%d too late as #%d was already"
gst_rtp_jitter_buffer_loop (GstRTPJitterBuffer * jitterbuffer)
{
GstRTPJitterBufferPrivate *priv;
- gpointer elem;
- GstBuffer *outbuf;
+ GstBuffer *outbuf = NULL;
GstFlowReturn result;
guint16 seqnum;
guint32 rtp_time;
priv = jitterbuffer->priv;
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK_CHECK (priv, flushing);
again:
GST_DEBUG_OBJECT (jitterbuffer, "Popping item");
- /* pop a buffer, we will get NULL if the queue was shut down */
- elem = async_jitter_queue_pop_unlocked (priv->queue);
- if (!elem)
- goto no_elem;
-
- /* special code for events */
- if (G_UNLIKELY (GST_IS_EVENT (elem))) {
- GstEvent *event = GST_EVENT_CAST (elem);
-
- switch (GST_EVENT_TYPE (event)) {
- case GST_EVENT_EOS:
- GST_DEBUG_OBJECT (jitterbuffer, "Popped EOS from queue");
- /* we don't expect more data now, makes upstream perform EOS actions */
- priv->srcresult = GST_FLOW_UNEXPECTED;
- break;
- default:
- GST_DEBUG_OBJECT (jitterbuffer, "Popped event %s from queue",
- GST_EVENT_TYPE_NAME (event));
- break;
- }
- async_jitter_queue_unlock (priv->queue);
-
- /* push event */
- gst_pad_push_event (priv->srcpad, event);
- return;
+ /* wait if we are blocked or don't have a packet and eos */
+ while (priv->blocked || !(rtp_jitter_buffer_num_packets (priv->jbuf)
+ || priv->eos)) {
+ JBUF_WAIT_CHECK (priv, flushing);
}
+ if (priv->eos)
+ goto do_eos;
- /* we know it's a buffer now */
- outbuf = GST_BUFFER_CAST (elem);
+ /* pop a buffer, we must have a buffer now */
+ outbuf = rtp_jitter_buffer_pop (priv->jbuf);
seqnum = gst_rtp_buffer_get_seq (outbuf);
- GST_DEBUG_OBJECT (jitterbuffer, "Popped buffer #%d from queue %d",
- gst_rtp_buffer_get_seq (outbuf),
- async_jitter_queue_length_unlocked (priv->queue));
+ GST_DEBUG_OBJECT (jitterbuffer, "Popped buffer #%d, now %d left",
+ seqnum, rtp_jitter_buffer_num_packets (priv->jbuf));
/* If we don't know what the next seqnum should be (== -1) we have to wait
* because it might be possible that we are not receiving this buffer in-order,
GST_OBJECT_UNLOCK (jitterbuffer);
/* release the lock so that the other end can push stuff or unlock */
- async_jitter_queue_unlock (priv->queue);
+ JBUF_UNLOCK (priv);
ret = gst_clock_id_wait (id, &jitter);
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK (priv);
/* and free the entry */
gst_clock_id_unref (id);
priv->clock_id = NULL;
GST_DEBUG_OBJECT (jitterbuffer,
"Wait got unscheduled, will retry to push with new buffer");
/* reinserting popped buffer into queue */
- if (!async_jitter_queue_push_sorted_unlocked (priv->queue, outbuf,
- (GCompareDataFunc) compare_rtp_buffers_seq_num, NULL)) {
+ if (!rtp_jitter_buffer_insert (priv->jbuf, outbuf)) {
GST_DEBUG_OBJECT (jitterbuffer,
"Duplicate packet #%d detected, dropping", seqnum);
priv->num_duplicates++;
* so the other end can push stuff in the queue again. */
priv->last_popped_seqnum = seqnum;
priv->next_seqnum = (seqnum + 1) & 0xffff;
- async_jitter_queue_unlock (priv->queue);
+ JBUF_UNLOCK (priv);
/* push buffer */
GST_DEBUG_OBJECT (jitterbuffer, "Pushing buffer %d", seqnum);
return;
/* ERRORS */
-no_elem:
+do_eos:
{
/* store result, we are flushing now */
- GST_DEBUG_OBJECT (jitterbuffer, "Pop returned NULL, we're flushing");
- priv->srcresult = GST_FLOW_WRONG_STATE;
+ GST_DEBUG_OBJECT (jitterbuffer, "We are EOS, pushing EOS downstream");
+ priv->srcresult = GST_FLOW_UNEXPECTED;
gst_pad_pause_task (priv->srcpad);
- async_jitter_queue_unlock (priv->queue);
+ gst_pad_push_event (priv->srcpad, gst_event_new_eos ());
+ JBUF_UNLOCK (priv);
return;
}
flushing:
{
GST_DEBUG_OBJECT (jitterbuffer, "we are flushing");
- gst_buffer_unref (outbuf);
- async_jitter_queue_unlock (priv->queue);
+ if (outbuf)
+ gst_buffer_unref (outbuf);
+ JBUF_UNLOCK (priv);
return;
}
pause:
GST_DEBUG_OBJECT (jitterbuffer, "pausing task, reason %s", reason);
- async_jitter_queue_lock (priv->queue);
+ JBUF_LOCK (priv);
/* store result */
priv->srcresult = result;
/* we don't post errors or anything because upstream will do that for us
* when we pass the return value upstream. */
gst_pad_pause_task (priv->srcpad);
- async_jitter_queue_unlock (priv->queue);
+ JBUF_UNLOCK (priv);
return;
}
}
old_latency = jitterbuffer->priv->latency_ms;
jitterbuffer->priv->latency_ms = new_latency;
- if (jitterbuffer->priv->clock_rate != -1) {
- async_jitter_queue_set_max_queue_length (jitterbuffer->priv->queue,
- gst_util_uint64_scale_int (new_latency,
- jitterbuffer->priv->clock_rate, 1000));
- }
+
/* post message if latency changed, this will infor the parent pipeline
* that a latency reconfiguration is possible. */
if (new_latency != old_latency) {
{
SIGNAL_REQUEST_PT_MAP,
SIGNAL_CLEAR_PT_MAP,
+
+ SIGNAL_ON_NEW_SSRC,
+ SIGNAL_ON_SSRC_COLLISION,
+ SIGNAL_ON_SSRC_VALIDATED,
+ SIGNAL_ON_BYE_SSRC,
+ SIGNAL_ON_BYE_TIMEOUT,
+ SIGNAL_ON_TIMEOUT,
LAST_SIGNAL
};
static guint gst_rtp_session_signals[LAST_SIGNAL] = { 0 };
+static void
+on_new_ssrc (RTPSession * session, RTPSource * src, GstRTPSession * sess)
+{
+ g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_NEW_SSRC], 0,
+ src->ssrc);
+}
+
+static void
+on_ssrc_collision (RTPSession * session, RTPSource * src, GstRTPSession * sess)
+{
+ g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SSRC_COLLISION], 0,
+ src->ssrc);
+}
+
+static void
+on_ssrc_validated (RTPSession * session, RTPSource * src, GstRTPSession * sess)
+{
+ g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SSRC_VALIDATED], 0,
+ src->ssrc);
+}
+
+static void
+on_bye_ssrc (RTPSession * session, RTPSource * src, GstRTPSession * sess)
+{
+ g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_BYE_SSRC], 0,
+ src->ssrc);
+}
+
+static void
+on_bye_timeout (RTPSession * session, RTPSource * src, GstRTPSession * sess)
+{
+ g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_BYE_TIMEOUT], 0,
+ src->ssrc);
+}
+
+static void
+on_timeout (RTPSession * session, RTPSource * src, GstRTPSession * sess)
+{
+ g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_TIMEOUT], 0,
+ src->ssrc);
+}
+
GST_BOILERPLATE (GstRTPSession, gst_rtp_session, GstElement, GST_TYPE_ELEMENT);
static void
G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRTPSessionClass, clear_pt_map),
NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
+ /**
+ * GstRTPSession::on-new-ssrc:
+ * @sess: the object which received the signal
+ * @ssrc: the SSRC
+ *
+ * Notify of a new SSRC that entered @session.
+ */
+ gst_rtp_session_signals[SIGNAL_ON_NEW_SSRC] =
+ g_signal_new ("on-new-ssrc", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPSessionClass, on_new_ssrc),
+ NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
+ /**
+ * GstRTPSession::on-ssrc_collision:
+ * @sess: the object which received the signal
+ * @ssrc: the SSRC
+ *
+ * Notify when we have an SSRC collision
+ */
+ gst_rtp_session_signals[SIGNAL_ON_SSRC_COLLISION] =
+ g_signal_new ("on-ssrc-collision", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPSessionClass,
+ on_ssrc_collision), NULL, NULL, g_cclosure_marshal_VOID__UINT,
+ G_TYPE_NONE, 1, G_TYPE_UINT);
+ /**
+ * GstRTPSession::on-ssrc_validated:
+ * @sess: the object which received the signal
+ * @ssrc: the SSRC
+ *
+ * Notify of a new SSRC that became validated.
+ */
+ gst_rtp_session_signals[SIGNAL_ON_SSRC_VALIDATED] =
+ g_signal_new ("on-ssrc-validated", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPSessionClass,
+ on_ssrc_validated), NULL, NULL, g_cclosure_marshal_VOID__UINT,
+ G_TYPE_NONE, 1, G_TYPE_UINT);
+
+ /**
+ * GstRTPSession::on-bye-ssrc:
+ * @sess: the object which received the signal
+ * @ssrc: the SSRC
+ *
+ * Notify of an SSRC that became inactive because of a BYE packet.
+ */
+ gst_rtp_session_signals[SIGNAL_ON_BYE_SSRC] =
+ g_signal_new ("on-bye-ssrc", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPSessionClass, on_bye_ssrc),
+ NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
+ /**
+ * GstRTPSession::on-bye-timeout:
+ * @sess: the object which received the signal
+ * @ssrc: the SSRC
+ *
+ * Notify of an SSRC that has timed out because of BYE
+ */
+ gst_rtp_session_signals[SIGNAL_ON_BYE_TIMEOUT] =
+ g_signal_new ("on-bye-timeout", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPSessionClass, on_bye_timeout),
+ NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
+ /**
+ * GstRTPSession::on-timeout:
+ * @sess: the object which received the signal
+ * @ssrc: the SSRC
+ *
+ * Notify of an SSRC that has timed out
+ */
+ gst_rtp_session_signals[SIGNAL_ON_TIMEOUT] =
+ g_signal_new ("on-timeout", G_TYPE_FROM_CLASS (klass),
+ G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTPSessionClass, on_timeout),
+ NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
+
gstelement_class->change_state =
GST_DEBUG_FUNCPTR (gst_rtp_session_change_state);
gstelement_class->request_new_pad =
rtpsession->priv->session = rtp_session_new ();
/* configure callbacks */
rtp_session_set_callbacks (rtpsession->priv->session, &callbacks, rtpsession);
+ /* configure signals */
+ g_signal_connect (rtpsession->priv->session, "on-new-ssrc",
+ (GCallback) on_new_ssrc, rtpsession);
+ g_signal_connect (rtpsession->priv->session, "on-ssrc-collision",
+ (GCallback) on_ssrc_collision, rtpsession);
+ g_signal_connect (rtpsession->priv->session, "on-ssrc-validated",
+ (GCallback) on_ssrc_validated, rtpsession);
+ g_signal_connect (rtpsession->priv->session, "on-bye-ssrc",
+ (GCallback) on_bye_ssrc, rtpsession);
+ g_signal_connect (rtpsession->priv->session, "on-bye-timeout",
+ (GCallback) on_bye_timeout, rtpsession);
+ g_signal_connect (rtpsession->priv->session, "on-timeout",
+ (GCallback) on_timeout, rtpsession);
}
static void
/* signals */
GstCaps* (*request_pt_map) (GstRTPSession *sess, guint pt);
-
void (*clear_pt_map) (GstRTPSession *sess);
+
+ void (*on_new_ssrc) (GstRTPSession *sess, guint32 ssrc);
+ void (*on_ssrc_collision) (GstRTPSession *sess, guint32 ssrc);
+ void (*on_ssrc_validated) (GstRTPSession *sess, guint32 ssrc);
+ void (*on_bye_ssrc) (GstRTPSession *sess, guint32 ssrc);
+ void (*on_bye_timeout) (GstRTPSession *sess, guint32 ssrc);
+ void (*on_timeout) (GstRTPSession *sess, guint32 ssrc);
};
GType gst_rtp_session_get_type (void);
--- /dev/null
+/* GStreamer
+ * Copyright (C) <2007> Wim Taymans <wim@fluendo.com>
+ *
+ * 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.
+ */
+#include <string.h>
+
+#include <gst/rtp/gstrtpbuffer.h>
+#include <gst/rtp/gstrtcpbuffer.h>
+
+#include "rtpjitterbuffer.h"
+
+GST_DEBUG_CATEGORY_STATIC (rtp_jitter_buffer_debug);
+#define GST_CAT_DEFAULT rtp_jitter_buffer_debug
+
+/* signals and args */
+enum
+{
+ LAST_SIGNAL
+};
+
+enum
+{
+ PROP_0
+};
+
+/* GObject vmethods */
+static void rtp_jitter_buffer_finalize (GObject * object);
+
+/* static guint rtp_jitter_buffer_signals[LAST_SIGNAL] = { 0 }; */
+
+G_DEFINE_TYPE (RTPJitterBuffer, rtp_jitter_buffer, G_TYPE_OBJECT);
+
+static void
+rtp_jitter_buffer_class_init (RTPJitterBufferClass * klass)
+{
+ GObjectClass *gobject_class;
+
+ gobject_class = (GObjectClass *) klass;
+
+ gobject_class->finalize = rtp_jitter_buffer_finalize;
+
+ GST_DEBUG_CATEGORY_INIT (rtp_jitter_buffer_debug, "rtpjitterbuffer", 0,
+ "RTP Jitter Buffer");
+}
+
+static void
+rtp_jitter_buffer_init (RTPJitterBuffer * jbuf)
+{
+ jbuf->packets = g_queue_new ();
+}
+
+static void
+rtp_jitter_buffer_finalize (GObject * object)
+{
+ RTPJitterBuffer *jbuf;
+
+ jbuf = RTP_JITTER_BUFFER_CAST (object);
+
+ rtp_jitter_buffer_flush (jbuf);
+ g_queue_free (jbuf->packets);
+
+ G_OBJECT_CLASS (rtp_jitter_buffer_parent_class)->finalize (object);
+}
+
+/**
+ * rtp_jitter_buffer_new:
+ *
+ * Create an #RTPJitterBuffer.
+ *
+ * Returns: a new #RTPJitterBuffer. Use g_object_unref() after usage.
+ */
+RTPJitterBuffer *
+rtp_jitter_buffer_new (void)
+{
+ RTPJitterBuffer *jbuf;
+
+ jbuf = g_object_new (RTP_TYPE_JITTER_BUFFER, NULL);
+
+ return jbuf;
+}
+
+static gint
+compare_seqnum (GstBuffer * a, GstBuffer * b, RTPJitterBuffer * jbuf)
+{
+ guint16 seq1, seq2;
+
+ seq1 = gst_rtp_buffer_get_seq (a);
+ seq2 = gst_rtp_buffer_get_seq (b);
+
+ /* check if diff more than half of the 16bit range */
+ if (abs (seq2 - seq1) > (1 << 15)) {
+ /* one of a/b has wrapped */
+ return seq1 - seq2;
+ } else {
+ return seq2 - seq1;
+ }
+}
+
+/**
+ * rtp_jitter_buffer_insert:
+ * @jbuf: an #RTPJitterBuffer
+ * @buf: a buffer
+ *
+ * Inserts @buf into the packet queue of @jbuf. The sequence number of the
+ * packet will be used to sort the packets. This function takes ownerhip of
+ * @buf when the function returns %TRUE.
+ *
+ * Returns: %FALSE if a packet with the same number already existed.
+ */
+gboolean
+rtp_jitter_buffer_insert (RTPJitterBuffer * jbuf, GstBuffer * buf)
+{
+ GList *list;
+ gint func_ret = 1;
+
+ g_return_val_if_fail (jbuf != NULL, FALSE);
+ g_return_val_if_fail (buf != NULL, FALSE);
+
+ /* loop the list to skip strictly smaller seqnum buffers */
+ list = jbuf->packets->head;
+ while (list
+ && (func_ret =
+ compare_seqnum (GST_BUFFER_CAST (list->data), buf, jbuf)) < 0)
+ list = list->next;
+
+ /* we hit a packet with the same seqnum, return FALSE to notify a duplicate */
+ if (func_ret == 0)
+ return FALSE;
+
+ if (list)
+ g_queue_insert_before (jbuf->packets, list, buf);
+ else
+ g_queue_push_tail (jbuf->packets, buf);
+
+ return TRUE;
+}
+
+/**
+ * rtp_jitter_buffer_pop:
+ * @jbuf: an #RTPJitterBuffer
+ *
+ * Pops the oldest buffer from the packet queue of @jbuf.
+ *
+ * Returns: a #GstBuffer or %NULL when there was no packet in the queue.
+ */
+GstBuffer *
+rtp_jitter_buffer_pop (RTPJitterBuffer * jbuf)
+{
+ GstBuffer *buf;
+
+ g_return_val_if_fail (jbuf != NULL, FALSE);
+
+ buf = g_queue_pop_tail (jbuf->packets);
+
+ return buf;
+}
+
+/**
+ * rtp_jitter_buffer_flush:
+ * @jbuf: an #RTPJitterBuffer
+ *
+ * Flush all packets from the jitterbuffer.
+ */
+void
+rtp_jitter_buffer_flush (RTPJitterBuffer * jbuf)
+{
+ GstBuffer *buffer;
+
+ g_return_if_fail (jbuf != NULL);
+
+ while ((buffer = g_queue_pop_head (jbuf->packets)))
+ gst_buffer_unref (buffer);
+}
+
+/**
+ * rtp_jitter_buffer_num_packets:
+ * @jbuf: an #RTPJitterBuffer
+ *
+ * Get the number of packets currently in "jbuf.
+ *
+ * Returns: The number of packets in @jbuf.
+ */
+guint
+rtp_jitter_buffer_num_packets (RTPJitterBuffer * jbuf)
+{
+ g_return_val_if_fail (jbuf != NULL, 0);
+
+ return jbuf->packets->length;
+}
+
+/**
+ * rtp_jitter_buffer_get_ts_diff:
+ * @jbuf: an #RTPJitterBuffer
+ *
+ * Get the difference between the timestamps of first and last packet in the
+ * jitterbuffer.
+ *
+ * Returns: The difference expressed in the timestamp units of the packets.
+ */
+guint32
+rtp_jitter_buffer_get_ts_diff (RTPJitterBuffer * jbuf)
+{
+ guint32 high_ts, low_ts;
+ GstBuffer *high_buf, *low_buf;
+
+ g_return_val_if_fail (jbuf != NULL, 0);
+
+ high_buf = g_queue_peek_head (jbuf->packets);
+ low_buf = g_queue_peek_tail (jbuf->packets);
+
+ if (!high_buf || !low_buf || high_buf == low_buf)
+ return 0;
+
+ high_ts = gst_rtp_buffer_get_timestamp (high_buf);
+ low_ts = gst_rtp_buffer_get_timestamp (low_buf);
+
+ /* it needs to work if ts wraps */
+ if (high_ts >= low_ts) {
+ return high_ts - low_ts;
+ } else {
+ return high_ts + G_MAXUINT32 + 1 - low_ts;
+ }
+}
--- /dev/null
+/* GStreamer
+ * Copyright (C) <2007> Wim Taymans <wim@fluendo.com>
+ *
+ * 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.
+ */
+
+#ifndef __RTP_JITTER_BUFFER_H__
+#define __RTP_JITTER_BUFFER_H__
+
+#include <gst/gst.h>
+#include <gst/rtp/gstrtcpbuffer.h>
+#include <gst/netbuffer/gstnetbuffer.h>
+
+typedef struct _RTPJitterBuffer RTPJitterBuffer;
+typedef struct _RTPJitterBufferClass RTPJitterBufferClass;
+
+#define RTP_TYPE_JITTER_BUFFER (rtp_jitter_buffer_get_type())
+#define RTP_JITTER_BUFFER(src) (G_TYPE_CHECK_INSTANCE_CAST((src),RTP_TYPE_JITTER_BUFFER,RTPJitterBuffer))
+#define RTP_JITTER_BUFFER_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass),RTP_TYPE_JITTER_BUFFER,RTPJitterBufferClass))
+#define RTP_IS_JITTER_BUFFER(src) (G_TYPE_CHECK_INSTANCE_TYPE((src),RTP_TYPE_JITTER_BUFFER))
+#define RTP_IS_JITTER_BUFFER_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),RTP_TYPE_JITTER_BUFFER))
+#define RTP_JITTER_BUFFER_CAST(src) ((RTPJitterBuffer *)(src))
+
+/**
+ * RTPJitterBuffer:
+ *
+ * A JitterBuffer in the #RTPSession
+ */
+struct _RTPJitterBuffer {
+ GObject object;
+
+ GQueue *packets;
+};
+
+struct _RTPJitterBufferClass {
+ GObjectClass parent_class;
+};
+
+GType rtp_jitter_buffer_get_type (void);
+
+/* managing lifetime */
+RTPJitterBuffer* rtp_jitter_buffer_new (void);
+
+gboolean rtp_jitter_buffer_insert (RTPJitterBuffer *jbuf, GstBuffer *buf);
+GstBuffer * rtp_jitter_buffer_pop (RTPJitterBuffer *jbuf);
+
+void rtp_jitter_buffer_flush (RTPJitterBuffer *jbuf);
+
+guint rtp_jitter_buffer_num_packets (RTPJitterBuffer *jbuf);
+guint32 rtp_jitter_buffer_get_ts_diff (RTPJitterBuffer *jbuf);
+
+
+
+#endif /* __RTP_JITTER_BUFFER_H__ */
static void
on_bye_ssrc (RTPSession * sess, RTPSource * source)
{
- /* notify app that reconsideration should be performed */
g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_BYE_SSRC], 0, source);
}
on_bye_timeout (sess, source);
else
on_timeout (sess, source);
-
}
return remove;
}
projects / platform / upstream / gstreamer.git / commitdiff