From f0d1ab1c1f680f18a2ef9ec09681ad77829e45b1 Mon Sep 17 00:00:00 2001 From: Wim Taymans Date: Tue, 3 Apr 2007 09:13:17 +0000 Subject: Add RTP session management elements. Still in progress. Original commit message from CVS: * configure.ac: * gst/rtpmanager/Makefile.am: * gst/rtpmanager/async_jitter_queue.c: (async_jitter_queue_new), (signal_waiting_threads), (async_jitter_queue_ref), (async_jitter_queue_ref_unlocked), (async_jitter_queue_set_low_threshold), (async_jitter_queue_set_high_threshold), (async_jitter_queue_set_max_queue_length), (async_jitter_queue_get_g_queue), (calculate_ts_diff), (async_jitter_queue_length_ts_units_unlocked), (async_jitter_queue_unref_and_unlock), (async_jitter_queue_unref), (async_jitter_queue_lock), (async_jitter_queue_unlock), (async_jitter_queue_push), (async_jitter_queue_push_unlocked), (async_jitter_queue_push_sorted), (async_jitter_queue_push_sorted_unlocked), (async_jitter_queue_insert_after_unlocked), (async_jitter_queue_pop_intern_unlocked), (async_jitter_queue_pop), (async_jitter_queue_pop_unlocked), (async_jitter_queue_length), (async_jitter_queue_length_unlocked), (async_jitter_queue_set_flushing_unlocked), (async_jitter_queue_unset_flushing_unlocked), (async_jitter_queue_set_blocking_unlocked): * gst/rtpmanager/async_jitter_queue.h: * gst/rtpmanager/gstrtpbin.c: (gst_rtp_bin_base_init), (gst_rtp_bin_class_init), (gst_rtp_bin_init), (gst_rtp_bin_finalize), (gst_rtp_bin_set_property), (gst_rtp_bin_get_property), (gst_rtp_bin_change_state), (gst_rtp_bin_request_new_pad), (gst_rtp_bin_release_pad): * gst/rtpmanager/gstrtpbin.h: * gst/rtpmanager/gstrtpclient.c: (new_pad), (create_stream), (free_stream), (find_stream_by_ssrc), (gst_rtp_client_base_init), (gst_rtp_client_class_init), (gst_rtp_client_init), (gst_rtp_client_finalize), (gst_rtp_client_set_property), (gst_rtp_client_get_property), (gst_rtp_client_change_state), (gst_rtp_client_request_new_pad), (gst_rtp_client_release_pad): * gst/rtpmanager/gstrtpclient.h: * gst/rtpmanager/gstrtpjitterbuffer.c: (gst_rtp_jitter_buffer_base_init), (gst_rtp_jitter_buffer_class_init), (gst_rtp_jitter_buffer_init), (gst_rtp_jitter_buffer_dispose), (gst_rtp_jitter_buffer_getcaps), (gst_jitter_buffer_sink_setcaps), (free_func), (gst_rtp_jitter_buffer_flush_start), (gst_rtp_jitter_buffer_flush_stop), (gst_rtp_jitter_buffer_src_activate_push), (gst_rtp_jitter_buffer_change_state), (priv_compare_rtp_seq_lt), (compare_rtp_buffers_seq_num), (gst_rtp_jitter_buffer_sink_event), (gst_rtp_jitter_buffer_chain), (gst_rtp_jitter_buffer_loop), (gst_rtp_jitter_buffer_query), (gst_rtp_jitter_buffer_set_property), (gst_rtp_jitter_buffer_get_property): * gst/rtpmanager/gstrtpjitterbuffer.h: * gst/rtpmanager/gstrtpmanager.c: (plugin_init): * gst/rtpmanager/gstrtpptdemux.c: (gst_rtp_pt_demux_base_init), (gst_rtp_pt_demux_class_init), (gst_rtp_pt_demux_init), (gst_rtp_pt_demux_finalize), (gst_rtp_pt_demux_chain), (gst_rtp_pt_demux_getcaps), (find_pad_for_pt), (gst_rtp_pt_demux_setup), (gst_rtp_pt_demux_release), (gst_rtp_pt_demux_change_state): * gst/rtpmanager/gstrtpptdemux.h: * gst/rtpmanager/gstrtpsession.c: (gst_rtp_session_base_init), (gst_rtp_session_class_init), (gst_rtp_session_init), (gst_rtp_session_finalize), (gst_rtp_session_set_property), (gst_rtp_session_get_property), (gst_rtp_session_change_state), (gst_rtp_session_chain_recv_rtp), (gst_rtp_session_chain_recv_rtcp), (gst_rtp_session_chain_send_rtp), (create_recv_rtp_sink), (create_recv_rtcp_sink), (create_send_rtp_sink), (create_rtcp_src), (gst_rtp_session_request_new_pad), (gst_rtp_session_release_pad): * gst/rtpmanager/gstrtpsession.h: Add RTP session management elements. Still in progress. --- gst/rtpmanager/Makefile.am | 23 + gst/rtpmanager/async_jitter_queue.c | 679 ++++++++++++++++++++++ gst/rtpmanager/async_jitter_queue.h | 130 +++++ gst/rtpmanager/gstrtpbin.c | 279 +++++++++ gst/rtpmanager/gstrtpbin.h | 56 ++ gst/rtpmanager/gstrtpclient.c | 482 ++++++++++++++++ gst/rtpmanager/gstrtpclient.h | 56 ++ gst/rtpmanager/gstrtpjitterbuffer.c | 1085 +++++++++++++++++++++++++++++++++++ gst/rtpmanager/gstrtpjitterbuffer.h | 74 +++ gst/rtpmanager/gstrtpmanager.c | 55 ++ gst/rtpmanager/gstrtpptdemux.c | 350 +++++++++++ gst/rtpmanager/gstrtpptdemux.h | 57 ++ gst/rtpmanager/gstrtpsession.c | 453 +++++++++++++++ gst/rtpmanager/gstrtpsession.h | 62 ++ 14 files changed, 3841 insertions(+) create mode 100644 gst/rtpmanager/Makefile.am create mode 100644 gst/rtpmanager/async_jitter_queue.c create mode 100644 gst/rtpmanager/async_jitter_queue.h create mode 100644 gst/rtpmanager/gstrtpbin.c create mode 100644 gst/rtpmanager/gstrtpbin.h create mode 100644 gst/rtpmanager/gstrtpclient.c create mode 100644 gst/rtpmanager/gstrtpclient.h create mode 100644 gst/rtpmanager/gstrtpjitterbuffer.c create mode 100644 gst/rtpmanager/gstrtpjitterbuffer.h create mode 100644 gst/rtpmanager/gstrtpmanager.c create mode 100644 gst/rtpmanager/gstrtpptdemux.c create mode 100644 gst/rtpmanager/gstrtpptdemux.h create mode 100644 gst/rtpmanager/gstrtpsession.c create mode 100644 gst/rtpmanager/gstrtpsession.h (limited to 'gst') diff --git a/gst/rtpmanager/Makefile.am b/gst/rtpmanager/Makefile.am new file mode 100644 index 00000000..561bf52d --- /dev/null +++ b/gst/rtpmanager/Makefile.am @@ -0,0 +1,23 @@ + +plugin_LTLIBRARIES = libgstrtpmanager.la + +libgstrtpmanager_la_SOURCES = gstrtpmanager.c \ + gstrtpbin.c \ + gstrtpclient.c \ + async_jitter_queue.c \ + gstrtpjitterbuffer.c \ + gstrtpptdemux.c \ + gstrtpsession.c + +noinst_HEADERS = gstrtpbin.h \ + gstrtpclient.h \ + async_jitter_queue.h \ + gstrtpjitterbuffer.h \ + gstrtpsession.h + +libgstrtpmanager_la_CFLAGS = $(GST_CFLAGS) $(GST_PLUGINS_BASE_CFLAGS) $(ERROR_CFLAGS) +libgstrtpmanager_la_LIBADD = $(GST_LIBS_LIBS) +libgstrtpmanager_la_LDFLAGS = $(GST_PLUGIN_LDFLAGS) $(GST_BASE_LIBS) $(GST_PLUGINS_BASE_LIBS) -lgstrtp-@GST_MAJORMINOR@ + +EXTRA_DIST = + diff --git a/gst/rtpmanager/async_jitter_queue.c b/gst/rtpmanager/async_jitter_queue.c new file mode 100644 index 00000000..22a8ed0e --- /dev/null +++ b/gst/rtpmanager/async_jitter_queue.c @@ -0,0 +1,679 @@ +/* + * 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 . + * + * 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 +#include + +#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) { + 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 + * async_jitter_queue_*_unlocked() 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) +{ + g_return_val_if_fail (queue != NULL, FALSE); + gboolean ret; + + 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; + + 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; + } + + if (async_jitter_queue_length_ts_units_unlocked (queue) <= + queue->low_threshold * queue->max_queue_length + && queue->pops_remaining == 0) { + if (!queue->buffering) { + queue->buffering = TRUE; + queue->pops_remaining = queue->queue->length; + } else { + 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) +{ + 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 */ + g_queue_foreach (queue->queue, free_func, 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); +} diff --git a/gst/rtpmanager/async_jitter_queue.h b/gst/rtpmanager/async_jitter_queue.h new file mode 100644 index 00000000..bea76d6d --- /dev/null +++ b/gst/rtpmanager/async_jitter_queue.h @@ -0,0 +1,130 @@ +/* Async Jitter Queue based on g_async_queue + * + * Farsight Voice+Video library + * Copyright 2007 Collabora Ltd, + * Copyright 2007 Nokia Corporation + * @author: Philippe Khalaf . + */ + +/* 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 +#include + +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__ */ + diff --git a/gst/rtpmanager/gstrtpbin.c b/gst/rtpmanager/gstrtpbin.c new file mode 100644 index 00000000..629f098d --- /dev/null +++ b/gst/rtpmanager/gstrtpbin.c @@ -0,0 +1,279 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +/** + * SECTION:element-rtpbin + * @short_description: handle media from one RTP bin + * @see_also: rtpjitterbuffer, rtpclient, rtpsession + * + * + * + * + * Example pipelines + * + * + * gst-launch -v filesrc location=sine.ogg ! oggdemux ! vorbisdec ! audioconvert ! alsasink + * + * + * + * + * Last reviewed on 2007-04-02 (0.10.6) + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif +#include + +#include "gstrtpbin.h" + +/* elementfactory information */ +static const GstElementDetails rtpbin_details = GST_ELEMENT_DETAILS ("RTP Bin", + "Filter/Editor/Video", + "Implement an RTP bin", + "Wim Taymans "); + +/* sink pads */ +static GstStaticPadTemplate rtpbin_recv_rtp_sink_template = +GST_STATIC_PAD_TEMPLATE ("recv_rtp_sink_%d", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtp") + ); + +static GstStaticPadTemplate rtpbin_recv_rtcp_sink_template = +GST_STATIC_PAD_TEMPLATE ("recv_rtcp_sink_%d", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtcp") + ); + +static GstStaticPadTemplate rtpbin_send_rtp_sink_template = +GST_STATIC_PAD_TEMPLATE ("send_rtp_sink_%d", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtp") + ); + +/* src pads */ +static GstStaticPadTemplate rtpbin_recv_rtp_src_template = +GST_STATIC_PAD_TEMPLATE ("recv_rtp_src_%d_%d_%d", + GST_PAD_SRC, + GST_PAD_SOMETIMES, + GST_STATIC_CAPS ("application/x-rtp") + ); + +static GstStaticPadTemplate rtpbin_send_rtcp_src_template = +GST_STATIC_PAD_TEMPLATE ("send_rtcp_src_%d", + GST_PAD_SRC, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtcp") + ); + +static GstStaticPadTemplate rtpbin_send_rtp_src_template = +GST_STATIC_PAD_TEMPLATE ("send_rtp_src_%d", + GST_PAD_SRC, + GST_PAD_SOMETIMES, + GST_STATIC_CAPS ("application/x-rtp") + ); + +#define GST_RTP_BIN_GET_PRIVATE(obj) \ + (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTP_BIN, GstRTPBinPrivate)) + +struct _GstRTPBinPrivate +{ +}; + +/* signals and args */ +enum +{ + /* FILL ME */ + LAST_SIGNAL +}; + +enum +{ + PROP_0 +}; + +/* GObject vmethods */ +static void gst_rtp_bin_finalize (GObject * object); +static void gst_rtp_bin_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec); +static void gst_rtp_bin_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec); + +/* GstElement vmethods */ +static GstStateChangeReturn gst_rtp_bin_change_state (GstElement * element, + GstStateChange transition); +static GstPad *gst_rtp_bin_request_new_pad (GstElement * element, + GstPadTemplate * templ, const gchar * name); +static void gst_rtp_bin_release_pad (GstElement * element, GstPad * pad); + +/*static guint gst_rtp_bin_signals[LAST_SIGNAL] = { 0 }; */ + +GST_BOILERPLATE (GstRTPBin, gst_rtp_bin, GstBin, GST_TYPE_BIN); + +static void +gst_rtp_bin_base_init (gpointer klass) +{ + GstElementClass *element_class = GST_ELEMENT_CLASS (klass); + + /* sink pads */ + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpbin_recv_rtp_sink_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpbin_recv_rtcp_sink_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpbin_send_rtp_sink_template)); + + /* src pads */ + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpbin_recv_rtp_src_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpbin_send_rtcp_src_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpbin_send_rtp_src_template)); + + gst_element_class_set_details (element_class, &rtpbin_details); +} + +static void +gst_rtp_bin_class_init (GstRTPBinClass * klass) +{ + GObjectClass *gobject_class; + GstElementClass *gstelement_class; + + gobject_class = (GObjectClass *) klass; + gstelement_class = (GstElementClass *) klass; + + g_type_class_add_private (klass, sizeof (GstRTPBinPrivate)); + + gobject_class->finalize = gst_rtp_bin_finalize; + gobject_class->set_property = gst_rtp_bin_set_property; + gobject_class->get_property = gst_rtp_bin_get_property; + + gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_rtp_bin_change_state); + gstelement_class->request_new_pad = + GST_DEBUG_FUNCPTR (gst_rtp_bin_request_new_pad); + gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_rtp_bin_release_pad); +} + +static void +gst_rtp_bin_init (GstRTPBin * rtpbin, GstRTPBinClass * klass) +{ + rtpbin->priv = GST_RTP_BIN_GET_PRIVATE (rtpbin); +} + +static void +gst_rtp_bin_finalize (GObject * object) +{ + GstRTPBin *rtpbin; + + rtpbin = GST_RTP_BIN (object); + + G_OBJECT_CLASS (parent_class)->finalize (object); +} + +static void +gst_rtp_bin_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstRTPBin *rtpbin; + + rtpbin = GST_RTP_BIN (object); + + switch (prop_id) { + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +gst_rtp_bin_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstRTPBin *rtpbin; + + rtpbin = GST_RTP_BIN (object); + + switch (prop_id) { + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static GstStateChangeReturn +gst_rtp_bin_change_state (GstElement * element, GstStateChange transition) +{ + GstStateChangeReturn res; + GstRTPBin *rtpbin; + + rtpbin = GST_RTP_BIN (element); + + switch (transition) { + case GST_STATE_CHANGE_NULL_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_PLAYING: + break; + default: + break; + } + + res = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); + + switch (transition) { + case GST_STATE_CHANGE_PLAYING_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_NULL: + break; + default: + break; + } + return res; +} + +/* + */ +static GstPad * +gst_rtp_bin_request_new_pad (GstElement * element, + GstPadTemplate * templ, const gchar * name) +{ + GstRTPBin *rtpbin; + GstElementClass *klass; + + g_return_val_if_fail (templ != NULL, NULL); + g_return_val_if_fail (GST_IS_RTP_BIN (element), NULL); + + rtpbin = GST_RTP_BIN (element); + klass = GST_ELEMENT_GET_CLASS (element); + + return NULL; +} + +static void +gst_rtp_bin_release_pad (GstElement * element, GstPad * pad) +{ +} diff --git a/gst/rtpmanager/gstrtpbin.h b/gst/rtpmanager/gstrtpbin.h new file mode 100644 index 00000000..5b3d7954 --- /dev/null +++ b/gst/rtpmanager/gstrtpbin.h @@ -0,0 +1,56 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * 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 __GST_RTP_BIN_H__ +#define __GST_RTP_BIN_H__ + +#include + +#define GST_TYPE_RTP_BIN \ + (gst_rtp_bin_get_type()) +#define GST_RTP_BIN(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_RTP_BIN,GstRTPBin)) +#define GST_RTP_BIN_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_RTP_BIN,GstRTPBinClass)) +#define GST_IS_RTP_BIN(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_RTP_BIN)) +#define GST_IS_RTP_BIN_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_RTP_BIN)) + +typedef struct _GstRTPBin GstRTPBin; +typedef struct _GstRTPBinClass GstRTPBinClass; +typedef struct _GstRTPBinPrivate GstRTPBinPrivate; + +struct _GstRTPBin { + GstBin element; + + /* a list of streams from a client */ + GList *streams; + + /*< private >*/ + GstRTPBinPrivate *priv; +}; + +struct _GstRTPBinClass { + GstBinClass parent_class; +}; + +GType gst_rtp_bin_get_type (void); + +#endif /* __GST_RTP_BIN_H__ */ diff --git a/gst/rtpmanager/gstrtpclient.c b/gst/rtpmanager/gstrtpclient.c new file mode 100644 index 00000000..2984d3f9 --- /dev/null +++ b/gst/rtpmanager/gstrtpclient.c @@ -0,0 +1,482 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +/** + * SECTION:element-rtpclient + * @short_description: handle media from one RTP client + * @see_also: rtpjitterbuffer, rtpbin, rtpsession + * + * + * + * This element handles RTP data from one client. It accepts multiple RTP streams that + * should be synchronized together. + * + * Example pipelines + * + * + * gst-launch -v filesrc location=sine.ogg ! oggdemux ! vorbisdec ! audioconvert ! alsasink + * + * + * + * + * Last reviewed on 2007-04-02 (0.10.6) + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif +#include + +#include "gstrtpclient.h" + +/* elementfactory information */ +static const GstElementDetails rtpclient_details = +GST_ELEMENT_DETAILS ("RTP Client", + "Filter/Editor/Video", + "Implement an RTP client", + "Wim Taymans "); + +/* sink pads */ +static GstStaticPadTemplate rtpclient_rtp_sink_template = +GST_STATIC_PAD_TEMPLATE ("rtp_sink_%d", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtp") + ); + +static GstStaticPadTemplate rtpclient_sync_sink_template = +GST_STATIC_PAD_TEMPLATE ("sync_sink_%d", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtcp") + ); + +/* src pads */ +static GstStaticPadTemplate rtpclient_rtp_src_template = +GST_STATIC_PAD_TEMPLATE ("rtp_src_%d_%d", + GST_PAD_SRC, + GST_PAD_SOMETIMES, + GST_STATIC_CAPS ("application/x-rtp") + ); + +#define GST_RTP_CLIENT_GET_PRIVATE(obj) \ + (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTP_CLIENT, GstRTPClientPrivate)) + +struct _GstRTPClientPrivate +{ +}; + +/* all the info needed to handle the stream with SSRC */ +typedef struct +{ + GstRTPClient *client; + + /* the SSRC of this stream */ + guint32 ssrc; + + /* RTP and RTCP in */ + GstPad *rtp_sink; + GstPad *sync_sink; + + /* the jitterbuffer */ + GstElement *jitterbuffer; + /* the payload demuxer */ + GstElement *ptdemux; + /* the new-pad signal */ + gulong new_pad_sig; +} GstRTPClientStream; + +/* the PT demuxer found a new payload type */ +static void +new_pad (GstElement * element, GstPad * pad, GstRTPClientStream * stream) +{ +} + +/* create a new stream for SSRC. + * + * We create a jitterbuffer and an payload demuxer for the SSRC. The sinkpad of + * the jitterbuffer is ghosted to the bin. We connect a pad-added signal to + * rtpptdemux so that we can ghost the payload pads outside. + * + * +-----------------+ +---------------+ + * | rtpjitterbuffer | | rtpptdemux | + * +- sink src - sink | + * / +-----------------+ +---------------+ + * + */ +static GstRTPClientStream * +create_stream (GstRTPClient * rtpclient, guint32 ssrc) +{ + GstRTPClientStream *stream; + gchar *name; + GstPad *srcpad, *sinkpad; + GstPadLinkReturn res; + + stream = g_new0 (GstRTPClientStream, 1); + stream->ssrc = ssrc; + stream->client = rtpclient; + + stream->jitterbuffer = gst_element_factory_make ("rtpjitterbuffer", NULL); + if (!stream->jitterbuffer) + goto no_jitterbuffer; + + stream->ptdemux = gst_element_factory_make ("rtpptdemux", NULL); + if (!stream->ptdemux) + goto no_ptdemux; + + /* add elements to bin */ + gst_bin_add (GST_BIN_CAST (rtpclient), stream->jitterbuffer); + gst_bin_add (GST_BIN_CAST (rtpclient), stream->ptdemux); + + /* link jitterbuffer and PT demuxer */ + srcpad = gst_element_get_pad (stream->jitterbuffer, "src"); + sinkpad = gst_element_get_pad (stream->ptdemux, "sink"); + res = gst_pad_link (srcpad, sinkpad); + gst_object_unref (srcpad); + gst_object_unref (sinkpad); + + if (res != GST_PAD_LINK_OK) + goto could_not_link; + + /* add stream to list */ + rtpclient->streams = g_list_prepend (rtpclient->streams, stream); + + /* ghost sinkpad */ + name = g_strdup_printf ("rtp_sink_%d", ssrc); + sinkpad = gst_element_get_pad (stream->jitterbuffer, "sink"); + stream->rtp_sink = gst_ghost_pad_new (name, sinkpad); + gst_object_unref (sinkpad); + g_free (name); + gst_element_add_pad (GST_ELEMENT_CAST (rtpclient), stream->rtp_sink); + + /* add signal to ptdemuxer */ + stream->new_pad_sig = + g_signal_connect (G_OBJECT (stream->ptdemux), "pad-added", + G_CALLBACK (new_pad), stream); + + return stream; + + /* ERRORS */ +no_jitterbuffer: + { + g_free (stream); + g_warning ("could not create rtpjitterbuffer element"); + return NULL; + } +no_ptdemux: + { + gst_object_unref (stream->jitterbuffer); + g_free (stream); + g_warning ("could not create rtpptdemux element"); + return NULL; + } +could_not_link: + { + gst_bin_remove (GST_BIN_CAST (rtpclient), stream->jitterbuffer); + gst_bin_remove (GST_BIN_CAST (rtpclient), stream->ptdemux); + g_free (stream); + g_warning ("could not link jitterbuffer and rtpptdemux element"); + return NULL; + } +} + +#if 0 +static void +free_stream (GstRTPClientStream * stream) +{ + gst_object_unref (stream->jitterbuffer); + g_free (stream); +} +#endif + +/* find the stream for the given SSRC, return NULL if the stream did not exist + */ +static GstRTPClientStream * +find_stream_by_ssrc (GstRTPClient * client, guint32 ssrc) +{ + GstRTPClientStream *stream; + GList *walk; + + for (walk = client->streams; walk; walk = g_list_next (walk)) { + stream = (GstRTPClientStream *) walk->data; + if (stream->ssrc == ssrc) + return stream; + } + return NULL; +} + +/* signals and args */ +enum +{ + /* FILL ME */ + LAST_SIGNAL +}; + +enum +{ + PROP_0 +}; + +/* GObject vmethods */ +static void gst_rtp_client_finalize (GObject * object); +static void gst_rtp_client_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec); +static void gst_rtp_client_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec); + +/* GstElement vmethods */ +static GstStateChangeReturn gst_rtp_client_change_state (GstElement * element, + GstStateChange transition); +static GstPad *gst_rtp_client_request_new_pad (GstElement * element, + GstPadTemplate * templ, const gchar * name); +static void gst_rtp_client_release_pad (GstElement * element, GstPad * pad); + +/*static guint gst_rtp_client_signals[LAST_SIGNAL] = { 0 }; */ + +GST_BOILERPLATE (GstRTPClient, gst_rtp_client, GstBin, GST_TYPE_BIN); + +static void +gst_rtp_client_base_init (gpointer klass) +{ + GstElementClass *element_class = GST_ELEMENT_CLASS (klass); + + /* sink pads */ + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpclient_rtp_sink_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpclient_sync_sink_template)); + + /* src pads */ + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpclient_rtp_src_template)); + + gst_element_class_set_details (element_class, &rtpclient_details); +} + +static void +gst_rtp_client_class_init (GstRTPClientClass * klass) +{ + GObjectClass *gobject_class; + GstElementClass *gstelement_class; + + gobject_class = (GObjectClass *) klass; + gstelement_class = (GstElementClass *) klass; + + g_type_class_add_private (klass, sizeof (GstRTPClientPrivate)); + + gobject_class->finalize = gst_rtp_client_finalize; + gobject_class->set_property = gst_rtp_client_set_property; + gobject_class->get_property = gst_rtp_client_get_property; + + gstelement_class->change_state = + GST_DEBUG_FUNCPTR (gst_rtp_client_change_state); + gstelement_class->request_new_pad = + GST_DEBUG_FUNCPTR (gst_rtp_client_request_new_pad); + gstelement_class->release_pad = + GST_DEBUG_FUNCPTR (gst_rtp_client_release_pad); +} + +static void +gst_rtp_client_init (GstRTPClient * rtpclient, GstRTPClientClass * klass) +{ + rtpclient->priv = GST_RTP_CLIENT_GET_PRIVATE (rtpclient); +} + +static void +gst_rtp_client_finalize (GObject * object) +{ + GstRTPClient *rtpclient; + + rtpclient = GST_RTP_CLIENT (object); + + G_OBJECT_CLASS (parent_class)->finalize (object); +} + +static void +gst_rtp_client_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstRTPClient *rtpclient; + + rtpclient = GST_RTP_CLIENT (object); + + switch (prop_id) { + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +gst_rtp_client_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstRTPClient *rtpclient; + + rtpclient = GST_RTP_CLIENT (object); + + switch (prop_id) { + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static GstStateChangeReturn +gst_rtp_client_change_state (GstElement * element, GstStateChange transition) +{ + GstStateChangeReturn res; + GstRTPClient *rtpclient; + + rtpclient = GST_RTP_CLIENT (element); + + switch (transition) { + case GST_STATE_CHANGE_NULL_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_PLAYING: + break; + default: + break; + } + + res = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); + + switch (transition) { + case GST_STATE_CHANGE_PLAYING_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_NULL: + break; + default: + break; + } + return res; +} + +/* We have 2 request pads (rtp_sink_%d and sync_sink_%d), the %d is assumed to + * be the SSRC of the stream. + * + * We require that the rtp pad is requested first for a particular SSRC, then + * (optionaly) the sync pad can be requested. If no sync pad is requested, no + * sync information can be exchanged for this stream. + */ +static GstPad * +gst_rtp_client_request_new_pad (GstElement * element, + GstPadTemplate * templ, const gchar * name) +{ + GstRTPClient *rtpclient; + GstElementClass *klass; + GstPadTemplate *rtp_sink_templ, *sync_sink_templ; + guint32 ssrc; + GstRTPClientStream *stream; + GstPad *result; + + g_return_val_if_fail (templ != NULL, NULL); + g_return_val_if_fail (GST_IS_RTP_CLIENT (element), NULL); + + if (templ->direction != GST_PAD_SINK) + goto wrong_direction; + + rtpclient = GST_RTP_CLIENT (element); + klass = GST_ELEMENT_GET_CLASS (element); + + /* figure out the template */ + rtp_sink_templ = gst_element_class_get_pad_template (klass, "rtp_sink_%d"); + sync_sink_templ = gst_element_class_get_pad_template (klass, "sync_sink_%d"); + + if (templ != rtp_sink_templ && templ != sync_sink_templ) + goto wrong_template; + + if (templ == rtp_sink_templ) { + /* create new rtp sink pad. If a stream with the pad number already exists + * we have an error, else we create the sinkpad, add a jitterbuffer and + * ptdemuxer. */ + if (name == NULL || strlen (name) < 9) + goto no_name; + + ssrc = atoi (&name[9]); + + /* see if a stream with that name exists, if so we have an error. */ + stream = find_stream_by_ssrc (rtpclient, ssrc); + if (stream != NULL) + goto stream_exists; + + /* ok, create new stream */ + stream = create_stream (rtpclient, ssrc); + if (stream == NULL) + goto stream_not_found; + + result = stream->rtp_sink; + } else { + /* create new rtp sink pad. We can only do this if the RTP pad was + * requested before, meaning the session with the padnumber must exist. */ + if (name == NULL || strlen (name) < 10) + goto no_name; + + ssrc = atoi (&name[10]); + + /* find stream */ + stream = find_stream_by_ssrc (rtpclient, ssrc); + if (stream == NULL) + goto stream_not_found; + + stream->sync_sink = + gst_pad_new_from_static_template (&rtpclient_sync_sink_template, name); + gst_element_add_pad (GST_ELEMENT_CAST (rtpclient), stream->sync_sink); + + result = stream->sync_sink; + } + + return result; + + /* ERRORS */ +wrong_direction: + { + g_warning ("rtpclient: request pad that is not a SINK pad"); + return NULL; + } +wrong_template: + { + g_warning ("rtpclient: this is not our template"); + return NULL; + } +no_name: + { + g_warning ("rtpclient: no padname was specified"); + return NULL; + } +stream_exists: + { + g_warning ("rtpclient: stream with SSRC %d already registered", ssrc); + return NULL; + } +stream_not_found: + { + g_warning ("rtpclient: stream with SSRC %d not yet registered", ssrc); + return NULL; + } +} + +static void +gst_rtp_client_release_pad (GstElement * element, GstPad * pad) +{ +} diff --git a/gst/rtpmanager/gstrtpclient.h b/gst/rtpmanager/gstrtpclient.h new file mode 100644 index 00000000..de837fd7 --- /dev/null +++ b/gst/rtpmanager/gstrtpclient.h @@ -0,0 +1,56 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * 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 __GST_RTP_CLIENT_H__ +#define __GST_RTP_CLIENT_H__ + +#include + +#define GST_TYPE_RTP_CLIENT \ + (gst_rtp_client_get_type()) +#define GST_RTP_CLIENT(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_RTP_CLIENT,GstRTPClient)) +#define GST_RTP_CLIENT_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_RTP_CLIENT,GstRTPClientClass)) +#define GST_IS_RTP_CLIENT(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_RTP_CLIENT)) +#define GST_IS_RTP_CLIENT_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_RTP_CLIENT)) + +typedef struct _GstRTPClient GstRTPClient; +typedef struct _GstRTPClientClass GstRTPClientClass; +typedef struct _GstRTPClientPrivate GstRTPClientPrivate; + +struct _GstRTPClient { + GstBin parent_bin; + + /* a list of streams from a client */ + GList *streams; + + /*< private >*/ + GstRTPClientPrivate *priv; +}; + +struct _GstRTPClientClass { + GstBinClass parent_class; +}; + +GType gst_rtp_client_get_type (void); + +#endif /* __GST_RTP_CLIENT_H__ */ diff --git a/gst/rtpmanager/gstrtpjitterbuffer.c b/gst/rtpmanager/gstrtpjitterbuffer.c new file mode 100644 index 00000000..58b48a34 --- /dev/null +++ b/gst/rtpmanager/gstrtpjitterbuffer.c @@ -0,0 +1,1085 @@ +/* + * Farsight Voice+Video library + * + * Copyright 2007 Collabora Ltd, + * Copyright 2007 Nokia Corporation + * @author: Philippe Kalaf . + * Copyright 2007 Wim Taymans + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + * + */ + +/** + * SECTION:element-rtpjitterbuffer + * @short_description: buffer, reorder and remove duplicate RTP packets to + * compensate for network oddities. + * + * + * + * This element reorders and removes duplicate RTP packets as they are received + * from a network source. It will also wait for missing packets up to a + * configurable time limit using the ::latency property. Packets arriving too + * late are considered as lost packets. + * + * + * This element acts as a live element and so adds ::latency to the pipeline. + * + * Example pipelines + * + * + * gst-launch rtspsrc location=rtsp://192.168.1.133:8554/mpeg1or2AudioVideoTest ! rtpjitterbuffer ! rtpmpvdepay ! mpeg2dec ! xvimagesink + * + * Connect to a streaming server and decode the MPEG video. The jitterbuffer is + * inserted into the pipeline to smooth out network jitter and to reorder the + * out-of-order RTP packets. + * + * + * + * Last reviewed on 2007-03-27 (0.10.13) + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include +#include +#include "gstrtpjitterbuffer.h" + +#include "async_jitter_queue.h" + +GST_DEBUG_CATEGORY (rtpjitterbuffer_debug); +#define GST_CAT_DEFAULT (rtpjitterbuffer_debug) + +/* low and high threshold tell the queue when to start and stop buffering */ +#define LOW_THRESHOLD 0.2 +#define HIGH_THRESHOLD 0.8 + +/* elementfactory information */ +static const GstElementDetails gst_rtp_jitter_buffer_details = +GST_ELEMENT_DETAILS ("RTP packet jitter-buffer", + "Filter/Network", + "A buffer that deals with network jitter and other transmission faults", + "Philippe Kalaf , " + "Wim Taymans "); + +/* RTPJitterBuffer signals and args */ +enum +{ + /* FILL ME */ + LAST_SIGNAL +}; + +#define DEFAULT_LATENCY_MS 200 +#define DEFAULT_DROP_ON_LATENCY FALSE + +enum +{ + ARG_0, + ARG_LATENCY, + ARG_DROP_ON_LATENCY +}; + +struct _GstRTPJitterBufferPrivate +{ + GstPad *sinkpad, *srcpad; + + AsyncJitterQueue *queue; + + /* properties */ + guint latency_ms; + gboolean drop_on_latency; + + /* the last seqnum we pushed out */ + guint32 last_popped_seqnum; + /* the next expected seqnum */ + guint32 next_seqnum; + + /* clock rate and rtp timestamp offset */ + gint32 clock_rate; + guint64 clock_base; + + /* when we are shutting down */ + GstFlowReturn srcresult; + + /* for sync */ + GstSegment segment; + GstClockID clock_id; + guint32 waiting_seqnum; + + /* some accounting */ + guint64 num_late; + guint64 num_duplicates; +}; + +#define GST_RTP_JITTER_BUFFER_GET_PRIVATE(o) \ + (G_TYPE_INSTANCE_GET_PRIVATE ((o), GST_TYPE_RTP_JITTER_BUFFER, \ + GstRTPJitterBufferPrivate)) + +static GstStaticPadTemplate gst_rtp_jitter_buffer_sink_template = +GST_STATIC_PAD_TEMPLATE ("sink", + GST_PAD_SINK, + GST_PAD_ALWAYS, + GST_STATIC_CAPS ("application/x-rtp, " + "clock-rate = (int) [ 1, 2147483647 ]" + /* "payload = (int) , " + * "encoding-name = (string) " + */ ) + ); + +static GstStaticPadTemplate gst_rtp_jitter_buffer_src_template = +GST_STATIC_PAD_TEMPLATE ("src", + GST_PAD_SRC, + GST_PAD_ALWAYS, + GST_STATIC_CAPS ("application/x-rtp" + /* "payload = (int) , " + * "clock-rate = (int) , " + * "encoding-name = (string) " + */ ) + ); + +GST_BOILERPLATE (GstRTPJitterBuffer, gst_rtp_jitter_buffer, GstElement, + GST_TYPE_ELEMENT); + +/* object overrides */ +static void gst_rtp_jitter_buffer_set_property (GObject * object, + guint prop_id, const GValue * value, GParamSpec * pspec); +static void gst_rtp_jitter_buffer_get_property (GObject * object, + guint prop_id, GValue * value, GParamSpec * pspec); +static void gst_rtp_jitter_buffer_dispose (GObject * object); + +/* element overrides */ +static GstStateChangeReturn gst_rtp_jitter_buffer_change_state (GstElement + * element, GstStateChange transition); + +/* pad overrides */ +static GstCaps *gst_rtp_jitter_buffer_getcaps (GstPad * pad); + +/* sinkpad overrides */ +static gboolean gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps); +static gboolean gst_rtp_jitter_buffer_sink_event (GstPad * pad, + GstEvent * event); +static GstFlowReturn gst_rtp_jitter_buffer_chain (GstPad * pad, + GstBuffer * buffer); + +/* srcpad overrides */ +static gboolean +gst_rtp_jitter_buffer_src_activate_push (GstPad * pad, gboolean active); +static void gst_rtp_jitter_buffer_loop (GstRTPJitterBuffer * jitterbuffer); +static gboolean gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query); + +static void +gst_rtp_jitter_buffer_base_init (gpointer klass) +{ + GstElementClass *element_class = GST_ELEMENT_CLASS (klass); + + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&gst_rtp_jitter_buffer_src_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&gst_rtp_jitter_buffer_sink_template)); + gst_element_class_set_details (element_class, &gst_rtp_jitter_buffer_details); +} + +static void +gst_rtp_jitter_buffer_class_init (GstRTPJitterBufferClass * klass) +{ + GObjectClass *gobject_class; + GstElementClass *gstelement_class; + + gobject_class = (GObjectClass *) klass; + gstelement_class = (GstElementClass *) klass; + + g_type_class_add_private (klass, sizeof (GstRTPJitterBufferPrivate)); + + gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_dispose); + + gobject_class->set_property = gst_rtp_jitter_buffer_set_property; + gobject_class->get_property = gst_rtp_jitter_buffer_get_property; + + g_object_class_install_property (gobject_class, ARG_LATENCY, + g_param_spec_uint ("latency", "Buffer latency in ms", + "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS, + G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, ARG_DROP_ON_LATENCY, + g_param_spec_boolean ("drop_on_latency", + "Drop buffers when maximum latency is reached", + "Tells the jitterbuffer to never exceed the given latency in size", + DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE)); + + gstelement_class->change_state = gst_rtp_jitter_buffer_change_state; + + GST_DEBUG_CATEGORY_INIT + (rtpjitterbuffer_debug, "rtpjitterbuffer", 0, "RTP Jitter Buffer"); +} + +static void +gst_rtp_jitter_buffer_init (GstRTPJitterBuffer * jitterbuffer, + GstRTPJitterBufferClass * klass) +{ + GstRTPJitterBufferPrivate *priv; + + priv = GST_RTP_JITTER_BUFFER_GET_PRIVATE (jitterbuffer); + jitterbuffer->priv = priv; + + 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->waiting_seqnum = -1; + + priv->srcpad = + gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_src_template, + "src"); + + gst_pad_set_activatepush_function (priv->srcpad, + GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_src_activate_push)); + gst_pad_set_query_function (priv->srcpad, + GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_query)); + gst_pad_set_getcaps_function (priv->srcpad, + GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_getcaps)); + + priv->sinkpad = + gst_pad_new_from_static_template (&gst_rtp_jitter_buffer_sink_template, + "sink"); + + gst_pad_set_chain_function (priv->sinkpad, + GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_chain)); + gst_pad_set_event_function (priv->sinkpad, + GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_sink_event)); + gst_pad_set_setcaps_function (priv->sinkpad, + GST_DEBUG_FUNCPTR (gst_jitter_buffer_sink_setcaps)); + gst_pad_set_getcaps_function (priv->sinkpad, + GST_DEBUG_FUNCPTR (gst_rtp_jitter_buffer_getcaps)); + + gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->srcpad); + gst_element_add_pad (GST_ELEMENT (jitterbuffer), priv->sinkpad); +} + +static void +gst_rtp_jitter_buffer_dispose (GObject * object) +{ + GstRTPJitterBuffer *jitterbuffer; + + jitterbuffer = GST_RTP_JITTER_BUFFER (object); + if (jitterbuffer->priv->queue) { + async_jitter_queue_unref (jitterbuffer->priv->queue); + jitterbuffer->priv->queue = NULL; + } + + G_OBJECT_CLASS (parent_class)->dispose (object); +} + +static GstCaps * +gst_rtp_jitter_buffer_getcaps (GstPad * pad) +{ + GstRTPJitterBuffer *jitterbuffer; + GstRTPJitterBufferPrivate *priv; + GstPad *other; + GstCaps *caps; + const GstCaps *templ; + + jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad)); + priv = jitterbuffer->priv; + + other = (pad == priv->srcpad ? priv->sinkpad : priv->srcpad); + + caps = gst_pad_peer_get_caps (other); + + templ = gst_pad_get_pad_template_caps (pad); + if (caps == NULL) { + GST_DEBUG_OBJECT (jitterbuffer, "copy template"); + caps = gst_caps_copy (templ); + } else { + GstCaps *intersect; + + GST_DEBUG_OBJECT (jitterbuffer, "intersect with template"); + + intersect = gst_caps_intersect (caps, templ); + gst_caps_unref (caps); + + caps = intersect; + } + gst_object_unref (jitterbuffer); + + return caps; +} + +static gboolean +gst_jitter_buffer_sink_setcaps (GstPad * pad, GstCaps * caps) +{ + GstRTPJitterBuffer *jitterbuffer; + GstRTPJitterBufferPrivate *priv; + GstStructure *caps_struct; + const GValue *value; + + jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad)); + priv = jitterbuffer->priv; + + /* first parse the caps */ + caps_struct = gst_caps_get_structure (caps, 0); + + /* we need a clock-rate to convert the rtp timestamps to GStreamer time and to + * measure the amount of data in the buffer */ + if (!gst_structure_get_int (caps_struct, "clock-rate", &priv->clock_rate)) + goto error; + + if (priv->clock_rate <= 0) + goto wrong_rate; + + /* gah, clock-base is uint. If we don't have a base, we will use the first + * buffer timestamp as the base time. This will screw up sync but it's better + * than nothing. */ + value = gst_structure_get_value (caps_struct, "clock-base"); + if (value && G_VALUE_HOLDS_UINT (value)) + priv->clock_base = g_value_get_uint (value); + else + priv->clock_base = -1; + + /* first expected seqnum */ + value = gst_structure_get_value (caps_struct, "seqnum-base"); + if (value && G_VALUE_HOLDS_UINT (value)) + priv->next_seqnum = g_value_get_uint (value); + else + priv->next_seqnum = -1; + + async_jitter_queue_set_max_queue_length (priv->queue, + priv->latency_ms * priv->clock_rate / 1000); + + /* set same caps on srcpad */ + gst_pad_set_caps (priv->srcpad, caps); + + gst_object_unref (jitterbuffer); + + return TRUE; + + /* ERRORS */ +error: + { + GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!"); + gst_object_unref (jitterbuffer); + return FALSE; + } +wrong_rate: + { + GST_DEBUG_OBJECT (jitterbuffer, "Invalid clock-rate %d", priv->clock_rate); + gst_object_unref (jitterbuffer); + return FALSE; + } +} + +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); + /* 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); + /* 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); +} + +static void +gst_rtp_jitter_buffer_flush_stop (GstRTPJitterBuffer * jitterbuffer) +{ + GstRTPJitterBufferPrivate *priv; + + priv = jitterbuffer->priv; + + async_jitter_queue_lock (priv->queue); + GST_DEBUG_OBJECT (jitterbuffer, "Enabling pop on queue"); + /* Mark as non flushing */ + priv->srcresult = GST_FLOW_OK; + gst_segment_init (&priv->segment, GST_FORMAT_TIME); + priv->last_popped_seqnum = -1; + priv->next_seqnum = -1; + /* allow pops from the src pad task */ + async_jitter_queue_unset_flushing_unlocked (jitterbuffer->priv->queue); + async_jitter_queue_unlock (priv->queue); +} + +static gboolean +gst_rtp_jitter_buffer_src_activate_push (GstPad * pad, gboolean active) +{ + gboolean result = TRUE; + GstRTPJitterBuffer *jitterbuffer = NULL; + + jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad)); + + if (active) { + /* allow data processing */ + gst_rtp_jitter_buffer_flush_stop (jitterbuffer); + + /* start pushing out buffers */ + GST_DEBUG_OBJECT (jitterbuffer, "Starting task on srcpad"); + gst_pad_start_task (jitterbuffer->priv->srcpad, + (GstTaskFunction) gst_rtp_jitter_buffer_loop, jitterbuffer); + } else { + /* make sure all data processing stops ASAP */ + gst_rtp_jitter_buffer_flush_start (jitterbuffer); + + /* NOTE this will hardlock if the state change is called from the src pad + * task thread because we will _join() the thread. */ + GST_DEBUG_OBJECT (jitterbuffer, "Stopping task on srcpad"); + result = gst_pad_stop_task (pad); + } + + gst_object_unref (jitterbuffer); + + return result; +} + +static GstStateChangeReturn +gst_rtp_jitter_buffer_change_state (GstElement * element, + GstStateChange transition) +{ + GstRTPJitterBuffer *jitterbuffer; + GstRTPJitterBufferPrivate *priv; + GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS; + + jitterbuffer = GST_RTP_JITTER_BUFFER (element); + priv = jitterbuffer->priv; + + switch (transition) { + case GST_STATE_CHANGE_NULL_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_PAUSED: + async_jitter_queue_lock (priv->queue); + /* 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); + break; + case GST_STATE_CHANGE_PAUSED_TO_PLAYING: + async_jitter_queue_lock (priv->queue); + /* unblock to allow streaming in PLAYING */ + async_jitter_queue_set_blocking_unlocked (jitterbuffer->priv->queue, + FALSE); + async_jitter_queue_unlock (priv->queue); + break; + default: + break; + } + + ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); + + switch (transition) { + case GST_STATE_CHANGE_READY_TO_PAUSED: + /* we are a live element because we sync to the clock, which we can only + * do in the PLAYING state */ + if (ret != GST_STATE_CHANGE_FAILURE) + ret = GST_STATE_CHANGE_NO_PREROLL; + break; + case GST_STATE_CHANGE_PLAYING_TO_PAUSED: + async_jitter_queue_lock (priv->queue); + /* block to stop streaming when PAUSED */ + async_jitter_queue_set_blocking_unlocked (jitterbuffer->priv->queue, + TRUE); + async_jitter_queue_unlock (priv->queue); + break; + case GST_STATE_CHANGE_PAUSED_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_NULL: + break; + default: + break; + } + + return ret; +} + +/** + * Performs comparison 'b - a' with check for overflows. + */ +static inline gint +priv_compare_rtp_seq_lt (guint16 a, guint16 b) +{ + /* check if diff more than half of the 16bit range */ + if (abs (b - a) > (1 << 15)) { + /* one of a/b has wrapped */ + return a - b; + } else { + return b - a; + } +} + +/** + * 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) +{ + gboolean ret = TRUE; + GstRTPJitterBuffer *jitterbuffer; + GstRTPJitterBufferPrivate *priv; + + jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad)); + priv = jitterbuffer->priv; + + switch (GST_EVENT_TYPE (event)) { + case GST_EVENT_NEWSEGMENT: + { + GstFormat format; + gdouble rate, arate; + gint64 start, stop, time; + gboolean update; + + gst_event_parse_new_segment_full (event, &update, &rate, &arate, &format, + &start, &stop, &time); + + /* we need time for now */ + if (format != GST_FORMAT_TIME) + goto newseg_wrong_format; + + GST_DEBUG_OBJECT (jitterbuffer, + "newsegment: update %d, rate %g, arate %g, start %" GST_TIME_FORMAT + ", stop %" GST_TIME_FORMAT ", time %" GST_TIME_FORMAT, + update, rate, arate, GST_TIME_ARGS (start), GST_TIME_ARGS (stop), + GST_TIME_ARGS (time)); + + /* now configure the values, we need these to time the release of the + * buffers on the srcpad. */ + gst_segment_set_newsegment_full (&priv->segment, update, + rate, arate, format, start, stop, time); + + /* FIXME, push SEGMENT in the queue. Sorting order might be difficult. */ + ret = gst_pad_push_event (priv->srcpad, event); + break; + } + case GST_EVENT_FLUSH_START: + gst_rtp_jitter_buffer_flush_start (jitterbuffer); + break; + case GST_EVENT_FLUSH_STOP: + gst_rtp_jitter_buffer_flush_stop (jitterbuffer); + break; + case GST_EVENT_EOS: + { + /* push EOS in queue. We always push it at the head */ + async_jitter_queue_lock (priv->queue); + /* 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 + gst_event_unref (event); + async_jitter_queue_unlock (priv->queue); + break; + } + default: + ret = gst_pad_push_event (priv->srcpad, event); + break; + } + +done: + gst_object_unref (jitterbuffer); + + return ret; + + /* ERRORS */ +newseg_wrong_format: + { + GST_DEBUG_OBJECT (jitterbuffer, "received non TIME newsegment"); + ret = FALSE; + goto done; + } +} + +static GstFlowReturn +gst_rtp_jitter_buffer_chain (GstPad * pad, GstBuffer * buffer) +{ + GstRTPJitterBuffer *jitterbuffer; + GstRTPJitterBufferPrivate *priv; + guint16 seqnum; + GstFlowReturn ret; + + + g_return_val_if_fail (gst_rtp_buffer_validate (buffer), GST_FLOW_ERROR); + + jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad)); + priv = jitterbuffer->priv; + + if (priv->clock_rate == -1) + goto not_negotiated; + + 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; + + /* let's check if this buffer is too late, we cannot accept packets with + * bigger seqnum than the one we already pushed. */ + if (priv->last_popped_seqnum != -1) { + if (priv_compare_rtp_seq_lt (priv->last_popped_seqnum, seqnum) < 0) + goto too_late; + } + + /* 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) { + GST_DEBUG_OBJECT (jitterbuffer, "Queue full, dropping old packet #%d", + seqnum); + GstBuffer *old_buf; + + old_buf = async_jitter_queue_pop_unlocked (priv->queue); + 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)) + goto duplicate; + + /* 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) { + if (priv->waiting_seqnum > seqnum) { + gst_clock_id_unschedule (priv->clock_id); + GST_DEBUG_OBJECT (jitterbuffer, "Unscheduling waiting buffer"); + } + } + + GST_DEBUG_OBJECT (jitterbuffer, "Pushed packet #%d on queue %d", + seqnum, async_jitter_queue_length_unlocked (priv->queue)); + +finished: + async_jitter_queue_unlock (priv->queue); + + gst_object_unref (jitterbuffer); + + return ret; + + /* ERRORS */ +not_negotiated: + { + GST_DEBUG_OBJECT (jitterbuffer, "No clock-rate in caps!"); + gst_buffer_unref (buffer); + gst_object_unref (jitterbuffer); + return GST_FLOW_NOT_NEGOTIATED; + } +out_flushing: + { + GST_DEBUG_OBJECT (jitterbuffer, "flushing %s", gst_flow_get_name (ret)); + gst_buffer_unref (buffer); + gst_object_unref (jitterbuffer); + goto finished; + } +too_late: + { + GST_DEBUG_OBJECT (jitterbuffer, "Packet #%d too late as #%d was already" + " popped, dropping", seqnum, priv->last_popped_seqnum); + priv->num_late++; + gst_buffer_unref (buffer); + goto finished; + } +duplicate: + { + GST_DEBUG_OBJECT (jitterbuffer, "Duplicate packet #%d detected, dropping", + seqnum); + priv->num_duplicates++; + gst_buffer_unref (buffer); + goto finished; + } +} + +/** + * This funcion will push out buffers on the source pad. + * + * For each pushed buffer, the seqnum is recorded, if the next buffer B has a + * different seqnum (missing packets before B), this function will wait for the + * missing packet to arrive up to the rtp timestamp of buffer B. + */ +static void +gst_rtp_jitter_buffer_loop (GstRTPJitterBuffer * jitterbuffer) +{ + GstRTPJitterBufferPrivate *priv; + gpointer elem; + GstBuffer *outbuf; + GstFlowReturn result; + guint16 seqnum; + guint32 rtp_time; + GstClockTime timestamp; + gint64 running_time; + + priv = jitterbuffer->priv; + + async_jitter_queue_lock (priv->queue); +again: + 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; + } + + /* pop a buffer, we will get NULL if the queue was shut down */ + outbuf = GST_BUFFER_CAST (elem); + + 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)); + + /* 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, + * a buffer with a lower seqnum could arrive later and we want to push that + * earlier buffer before this buffer then. + * If we know the expected seqnum, we can compare it to the current seqnum to + * determine if we have missing a packet. If we have a missing packet (which + * must be before this packet) we can wait for it until the deadline for this + * packet expires. */ + if (priv->next_seqnum == -1 || priv->next_seqnum != seqnum) { + GstClockID id; + GstClockTimeDiff jitter; + GstClockReturn ret; + GstClock *clock; + + if (priv->next_seqnum != -1) { + /* we expected next_seqnum but received something else, that's a gap */ + GST_DEBUG_OBJECT (jitterbuffer, + "Sequence number GAP detected -> %d instead of %d", priv->next_seqnum, + seqnum); + } else { + /* we don't know what the next_seqnum should be, wait for the last + * possible moment to push this buffer, maybe we get an earlier seqnum + * while we wait */ + GST_DEBUG_OBJECT (jitterbuffer, "First buffer %d, do sync", seqnum); + } + + /* get the max deadline to wait for the missing packets, this is the time + * of the currently popped packet */ + rtp_time = gst_rtp_buffer_get_timestamp (outbuf); + + GST_DEBUG_OBJECT (jitterbuffer, "rtp_time %u, base %u", rtp_time, + priv->clock_base); + + /* if no clock_base was given, take first ts as base */ + if (priv->clock_base == -1) + priv->clock_base = rtp_time; + + /* take rtp timestamp offset into account, this can wrap around */ + rtp_time -= priv->clock_base; + + /* bring timestamp to gst time */ + timestamp = + gst_util_uint64_scale_int (GST_SECOND, rtp_time, priv->clock_rate); + + GST_DEBUG_OBJECT (jitterbuffer, "timestamp %" GST_TIME_FORMAT, + GST_TIME_ARGS (timestamp)); + + /* bring to running time */ + running_time = gst_segment_to_running_time (&priv->segment, GST_FORMAT_TIME, + timestamp); + + /* correct for sync against the gstreamer clock, add latency */ + GST_OBJECT_LOCK (jitterbuffer); + clock = GST_ELEMENT_CLOCK (jitterbuffer); + if (!clock) { + GST_OBJECT_UNLOCK (jitterbuffer); + /* let's just push if there is no clock */ + goto push_buffer; + } + + /* add latency */ + running_time += (priv->latency_ms * GST_MSECOND); + + GST_DEBUG_OBJECT (jitterbuffer, "sync to running_time %" GST_TIME_FORMAT, + GST_TIME_ARGS (running_time)); + + /* prepare for sync against clock */ + running_time += GST_ELEMENT_CAST (jitterbuffer)->base_time; + + /* create an entry for the clock */ + id = priv->clock_id = gst_clock_new_single_shot_id (clock, running_time); + priv->waiting_seqnum = seqnum; + GST_OBJECT_UNLOCK (jitterbuffer); + + /* release the lock so that the other end can push stuff or unlock */ + async_jitter_queue_unlock (priv->queue); + + ret = gst_clock_id_wait (id, &jitter); + + async_jitter_queue_lock (priv->queue); + /* and free the entry */ + gst_clock_id_unref (id); + priv->clock_id = NULL; + priv->waiting_seqnum = -1; + + /* at this point, the clock could have been unlocked by a timeout, a new + * tail element was added to the queue or because we are shutting down. Check + * for shutdown first. */ + if (priv->srcresult != GST_FLOW_OK) + goto flushing; + + /* if we got unscheduled and we are not flushing, it's because a new tail + * element became available in the queue. Grab it and try to push or sync. */ + if (ret == GST_CLOCK_UNSCHEDULED) { + 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)) { + GST_DEBUG_OBJECT (jitterbuffer, + "Duplicate packet #%d detected, dropping", seqnum); + priv->num_duplicates++; + gst_buffer_unref (outbuf); + } + goto again; + } + } +push_buffer: + /* check if we are pushing something unexpected */ + if (priv->next_seqnum != -1 && priv->next_seqnum != seqnum) { + gint dropped; + + /* calc number of missing packets, careful for wraparounds */ + dropped = priv_compare_rtp_seq_lt (priv->next_seqnum, seqnum); + + GST_DEBUG_OBJECT (jitterbuffer, + "Pushing DISCONT after dropping %d (%d to %d)", dropped, + priv->next_seqnum, seqnum); + + /* update stats */ + priv->num_late += dropped; + + /* set DISCONT flag */ + outbuf = gst_buffer_make_metadata_writable (outbuf); + GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT); + } + /* now we are ready to push the buffer. Save the seqnum and release the lock + * 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); + + /* push buffer */ + GST_DEBUG_OBJECT (jitterbuffer, "Pushing buffer %d", seqnum); + result = gst_pad_push (priv->srcpad, outbuf); + if (result != GST_FLOW_OK) + goto pause; + + return; + + /* ERRORS */ +no_elem: + { + /* store result, we are flushing now */ + GST_DEBUG_OBJECT (jitterbuffer, "Pop returned NULL, we're flushing"); + priv->srcresult = GST_FLOW_WRONG_STATE; + gst_pad_pause_task (priv->srcpad); + async_jitter_queue_unlock (priv->queue); + return; + } +flushing: + { + GST_DEBUG_OBJECT (jitterbuffer, "we are flushing"); + gst_buffer_unref (outbuf); + async_jitter_queue_unlock (priv->queue); + return; + } +pause: + { + const gchar *reason = gst_flow_get_name (result); + + GST_DEBUG_OBJECT (jitterbuffer, "pausing task, reason %s", reason); + + async_jitter_queue_lock (priv->queue); + /* 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); + return; + } +} + +static gboolean +gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query) +{ + GstRTPJitterBuffer *jitterbuffer; + GstRTPJitterBufferPrivate *priv; + gboolean res = FALSE; + + jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad)); + priv = jitterbuffer->priv; + + switch (GST_QUERY_TYPE (query)) { + case GST_QUERY_LATENCY: + { + /* We need to send the query upstream and add the returned latency to our + * own */ + GstClockTime min_latency, max_latency; + gboolean us_live; + GstPad *peer; + + if ((peer = gst_pad_get_peer (priv->sinkpad))) { + if ((res = gst_pad_query (peer, query))) { + gst_query_parse_latency (query, &us_live, &min_latency, &max_latency); + + min_latency += priv->latency_ms * GST_MSECOND; + max_latency += priv->latency_ms * GST_MSECOND; + + GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %" + GST_TIME_FORMAT " max %" GST_TIME_FORMAT, + GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency)); + + gst_query_set_latency (query, TRUE, min_latency, max_latency); + } + gst_object_unref (peer); + } + break; + } + default: + break; + } + return res; +} + +static void +gst_rtp_jitter_buffer_set_property (GObject * object, + guint prop_id, const GValue * value, GParamSpec * pspec) +{ + GstRTPJitterBuffer *jitterbuffer = GST_RTP_JITTER_BUFFER (object); + + switch (prop_id) { + case ARG_LATENCY: + { + guint new_latency, old_latency; + + /* FIXME, not threadsafe */ + new_latency = g_value_get_uint (value); + 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) { + gst_element_post_message (GST_ELEMENT_CAST (jitterbuffer), + gst_message_new_latency (GST_OBJECT_CAST (jitterbuffer))); + } + break; + } + case ARG_DROP_ON_LATENCY: + { + jitterbuffer->priv->drop_on_latency = g_value_get_boolean (value); + break; + } + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +gst_rtp_jitter_buffer_get_property (GObject * object, + guint prop_id, GValue * value, GParamSpec * pspec) +{ + GstRTPJitterBuffer *jitterbuffer = GST_RTP_JITTER_BUFFER (object); + + switch (prop_id) { + case ARG_LATENCY: + g_value_set_uint (value, jitterbuffer->priv->latency_ms); + break; + case ARG_DROP_ON_LATENCY: + g_value_set_boolean (value, jitterbuffer->priv->drop_on_latency); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} diff --git a/gst/rtpmanager/gstrtpjitterbuffer.h b/gst/rtpmanager/gstrtpjitterbuffer.h new file mode 100644 index 00000000..a8671440 --- /dev/null +++ b/gst/rtpmanager/gstrtpjitterbuffer.h @@ -0,0 +1,74 @@ +/* + * Farsight Voice+Video library + * + * Copyright 2007 Collabora Ltd, + * Copyright 2007 Nokia Corporation + * @author: Philippe Kalaf . + * Copyright 2007 Wim Taymans + * + * 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 __GST_RTP_JITTER_BUFFER_H__ +#define __GST_RTP_JITTER_BUFFER_H__ + +#include +#include + +G_BEGIN_DECLS + +/* #define's don't like whitespacey bits */ +#define GST_TYPE_RTP_JITTER_BUFFER \ + (gst_rtp_jitter_buffer_get_type()) +#define GST_RTP_JITTER_BUFFER(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj), \ + GST_TYPE_RTP_JITTER_BUFFER,GstRTPJitterBuffer)) +#define GST_RTP_JITTER_BUFFER_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass), \ + GST_TYPE_RTP_JITTER_BUFFER,GstRTPJitterBufferClass)) +#define GST_IS_RTP_JITTER_BUFFER(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_RTP_JITTER_BUFFER)) +#define GST_IS_RTP_JITTER_BUFFER_CLASS(obj) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_RTP_JITTER_BUFFER)) + +typedef struct _GstRTPJitterBuffer GstRTPJitterBuffer; +typedef struct _GstRTPJitterBufferClass GstRTPJitterBufferClass; +typedef struct _GstRTPJitterBufferPrivate GstRTPJitterBufferPrivate; + +struct _GstRTPJitterBuffer +{ + GstElement parent; + + GstRTPJitterBufferPrivate *priv; + + /*< private > */ + gpointer _gst_reserved[GST_PADDING]; +}; + +struct _GstRTPJitterBufferClass +{ + GstElementClass parent_class; + + /*< private > */ + gpointer _gst_reserved[GST_PADDING]; +}; + +GType gst_rtp_jitter_buffer_get_type (void); + +G_END_DECLS + +#endif /* __GST_RTP_JITTER_BUFFER_H__ */ diff --git a/gst/rtpmanager/gstrtpmanager.c b/gst/rtpmanager/gstrtpmanager.c new file mode 100644 index 00000000..a059cad8 --- /dev/null +++ b/gst/rtpmanager/gstrtpmanager.c @@ -0,0 +1,55 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * 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. + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "gstrtpclient.h" +#include "gstrtpjitterbuffer.h" +#include "gstrtpptdemux.h" +#include "gstrtpsession.h" + +static gboolean +plugin_init (GstPlugin * plugin) +{ + if (!gst_element_register (plugin, "rtpclient", GST_RANK_NONE, + GST_TYPE_RTP_CLIENT)) + return FALSE; + + if (!gst_element_register (plugin, "rtpjitterbuffer", GST_RANK_NONE, + GST_TYPE_RTP_JITTER_BUFFER)) + return FALSE; + + if (!gst_element_register (plugin, "rtpptdemux", GST_RANK_NONE, + GST_TYPE_RTP_PT_DEMUX)) + return FALSE; + + if (!gst_element_register (plugin, "rtpsession", GST_RANK_NONE, + GST_TYPE_RTP_SESSION)) + return FALSE; + + return TRUE; +} + +GST_PLUGIN_DEFINE (GST_VERSION_MAJOR, + GST_VERSION_MINOR, + "rtpmanager", + "RTP session management plugin library", + plugin_init, VERSION, "LGPL", GST_PACKAGE_NAME, GST_PACKAGE_ORIGIN) diff --git a/gst/rtpmanager/gstrtpptdemux.c b/gst/rtpmanager/gstrtpptdemux.c new file mode 100644 index 00000000..5950f619 --- /dev/null +++ b/gst/rtpmanager/gstrtpptdemux.c @@ -0,0 +1,350 @@ +/* + * RTP Demux element + * + * Copyright (C) 2005 Nokia Corporation. + * @author Kai Vehmanen + * + * Loosely based on GStreamer gstdecodebin + * Copyright (C) <2004> Wim Taymans + * + * 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. + */ + +/* + * Contributors: + * Andre Moreira Magalhaes + */ + +/* + * Status: + * - works with the test_rtpdemux.c tool + * + * Check: + * - is emitting a signal enough, or should we + * use GstEvent to notify downstream elements + * of the new packet... no? + * + * Notes: + * - emits event both for new PTs, and whenever + * a PT is changed + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include +#include +#include "gstrtpptdemux.h" +#include + +/* generic templates */ +static GstStaticPadTemplate rtp_pt_demux_sink_template = +GST_STATIC_PAD_TEMPLATE ("sink", + GST_PAD_SINK, + GST_PAD_ALWAYS, + GST_STATIC_CAPS ("application/x-rtp, " + "payload = (int) [ 0, 255 ], " "clock-rate = (int) [ 0, 2147483647 ]") + ); + +static GstStaticPadTemplate rtp_pt_demux_src_template = +GST_STATIC_PAD_TEMPLATE ("src%d", + GST_PAD_SRC, + GST_PAD_SOMETIMES, + GST_STATIC_CAPS_ANY); + +GST_DEBUG_CATEGORY_STATIC (gst_rtp_pt_demux_debug); +#define GST_CAT_DEFAULT gst_rtp_pt_demux_debug + +/** + * Item for storing GstPad<->pt pairs. + */ +struct _GstRTPPtDemuxPad +{ + GstPad *pad; /**< pointer to the actual pad */ + gint pt; /**< RTP payload-type attached to pad */ +}; + +/* signals */ +enum +{ + SIGNAL_NEW_PAYLOAD_TYPE, + SIGNAL_PAYLOAD_TYPE_CHANGE, + LAST_SIGNAL +}; + +GST_BOILERPLATE (GstRTPPtDemux, gst_rtp_pt_demux, GstElement, GST_TYPE_ELEMENT); + +static void gst_rtp_pt_demux_finalize (GObject * object); + +static void gst_rtp_pt_demux_release (GstElement * element); +static gboolean gst_rtp_pt_demux_setup (GstElement * element); + +static GstFlowReturn gst_rtp_pt_demux_chain (GstPad * pad, GstBuffer * buf); +static GstCaps *gst_rtp_pt_demux_getcaps (GstPad * pad); +static GstStateChangeReturn gst_rtp_pt_demux_change_state (GstElement * element, + GstStateChange transition); + +static GstPad *find_pad_for_pt (GstRTPPtDemux * rtpdemux, guint8 pt); + +static guint gst_rtp_pt_demux_signals[LAST_SIGNAL] = { 0 }; + +static GstElementDetails gst_rtp_pt_demux_details = { + "RTP Demux", + /* XXX: what's the correct hierarchy? */ + "Codec/Demux/Network", + "Parses codec streams transmitted in the same RTP session", + "Kai Vehmanen " +}; + +static void +gst_rtp_pt_demux_base_init (gpointer g_class) +{ + GstElementClass *gstelement_klass = GST_ELEMENT_CLASS (g_class); + + gst_element_class_add_pad_template (gstelement_klass, + gst_static_pad_template_get (&rtp_pt_demux_sink_template)); + gst_element_class_add_pad_template (gstelement_klass, + gst_static_pad_template_get (&rtp_pt_demux_src_template)); + + gst_element_class_set_details (gstelement_klass, &gst_rtp_pt_demux_details); +} + +static void +gst_rtp_pt_demux_class_init (GstRTPPtDemuxClass * klass) +{ + GObjectClass *gobject_klass; + GstElementClass *gstelement_klass; + + gobject_klass = (GObjectClass *) klass; + gstelement_klass = (GstElementClass *) klass; + + gst_rtp_pt_demux_signals[SIGNAL_NEW_PAYLOAD_TYPE] = + g_signal_new ("new-payload-type", + G_TYPE_FROM_CLASS (klass), + G_SIGNAL_RUN_LAST, + G_STRUCT_OFFSET (GstRTPPtDemuxClass, new_payload_type), + NULL, NULL, + g_cclosure_marshal_VOID__UINT_POINTER, + G_TYPE_NONE, 2, G_TYPE_INT, GST_TYPE_PAD); + gst_rtp_pt_demux_signals[SIGNAL_PAYLOAD_TYPE_CHANGE] = + g_signal_new ("payload-type-change", + G_TYPE_FROM_CLASS (klass), + G_SIGNAL_RUN_LAST, + G_STRUCT_OFFSET (GstRTPPtDemuxClass, payload_type_change), + NULL, NULL, gst_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT); + + gobject_klass->finalize = GST_DEBUG_FUNCPTR (gst_rtp_pt_demux_finalize); + + gstelement_klass->change_state = + GST_DEBUG_FUNCPTR (gst_rtp_pt_demux_change_state); + + GST_DEBUG_CATEGORY_INIT (gst_rtp_pt_demux_debug, + "rtpptdemux", 0, "RTP codec demuxer"); + +} + +static void +gst_rtp_pt_demux_init (GstRTPPtDemux * ptdemux, GstRTPPtDemuxClass * g_class) +{ + GstElementClass *klass = GST_ELEMENT_GET_CLASS (ptdemux); + + ptdemux->sink = + gst_pad_new_from_template (gst_element_class_get_pad_template (klass, + "sink"), "sink"); + g_assert (ptdemux->sink != NULL); + + gst_pad_set_chain_function (ptdemux->sink, gst_rtp_pt_demux_chain); + + gst_element_add_pad (GST_ELEMENT (ptdemux), ptdemux->sink); +} + +static void +gst_rtp_pt_demux_finalize (GObject * object) +{ + gst_rtp_pt_demux_release (GST_ELEMENT (object)); + + G_OBJECT_CLASS (parent_class)->finalize (object); +} + +static GstFlowReturn +gst_rtp_pt_demux_chain (GstPad * pad, GstBuffer * buf) +{ + GstFlowReturn ret = GST_FLOW_OK; + GstRTPPtDemux *rtpdemux; + GstElement *element = GST_ELEMENT (GST_OBJECT_PARENT (pad)); + guint8 pt; + GstPad *srcpad; + + rtpdemux = GST_RTP_PT_DEMUX (GST_OBJECT_PARENT (pad)); + + g_return_val_if_fail (gst_rtp_buffer_validate (buf), GST_FLOW_ERROR); + + pt = gst_rtp_buffer_get_payload_type (buf); + + srcpad = find_pad_for_pt (rtpdemux, pt); + if (srcpad == NULL) { + /* new PT, create a src pad */ + GstElementClass *klass; + GstPadTemplate *templ; + gchar *padname; + GstCaps *caps; + GstRTPPtDemuxPad *rtpdemuxpad; + + klass = GST_ELEMENT_GET_CLASS (rtpdemux); + templ = gst_element_class_get_pad_template (klass, "src%d"); + padname = g_strdup_printf ("src%d", pt); + srcpad = gst_pad_new_from_template (templ, padname); + g_free (padname); + + caps = gst_pad_get_caps (srcpad); + caps = gst_caps_make_writable (caps); + gst_caps_append_structure (caps, + gst_structure_new ("payload", "payload", G_TYPE_INT, pt, NULL)); + gst_pad_set_caps (srcpad, caps); + + /* XXX: set _link () function */ + gst_pad_set_getcaps_function (srcpad, gst_rtp_pt_demux_getcaps); + gst_pad_set_active (srcpad, TRUE); + gst_element_add_pad (element, srcpad); + + if (srcpad) { + GST_DEBUG ("Adding pt=%d to the list.", pt); + rtpdemuxpad = g_new0 (GstRTPPtDemuxPad, 1); + rtpdemuxpad->pt = pt; + rtpdemuxpad->pad = srcpad; + rtpdemux->srcpads = g_slist_append (rtpdemux->srcpads, rtpdemuxpad); + + GST_DEBUG ("emitting new-payload_type for pt %d", pt); + g_signal_emit (G_OBJECT (rtpdemux), + gst_rtp_pt_demux_signals[SIGNAL_NEW_PAYLOAD_TYPE], 0, pt, srcpad); + } + } + + if (pt != rtpdemux->last_pt) { + gint emit_pt = pt; + + /* our own signal with an extra flag that this is the only pad */ + rtpdemux->last_pt = pt; + GST_DEBUG ("emitting payload-type-changed for pt %d", emit_pt); + g_signal_emit (G_OBJECT (rtpdemux), + gst_rtp_pt_demux_signals[SIGNAL_PAYLOAD_TYPE_CHANGE], 0, emit_pt); + } + + /* push to srcpad */ + if (srcpad) + gst_pad_push (srcpad, GST_BUFFER (buf)); + + return ret; +} + +static GstCaps * +gst_rtp_pt_demux_getcaps (GstPad * pad) +{ + GstCaps *caps; + + GST_OBJECT_LOCK (pad); + if ((caps = GST_PAD_CAPS (pad))) + caps = gst_caps_ref (caps); + GST_OBJECT_UNLOCK (pad); + + return caps; +} + +static GstPad * +find_pad_for_pt (GstRTPPtDemux * rtpdemux, guint8 pt) +{ + GstPad *respad = NULL; + GSList *item = rtpdemux->srcpads; + + for (; item; item = g_slist_next (item)) { + GstRTPPtDemuxPad *pad = item->data; + + if (pad->pt == pt) { + respad = pad->pad; + break; + } + } + + return respad; +} + +/** + * Reserves resources for the object. + */ +static gboolean +gst_rtp_pt_demux_setup (GstElement * element) +{ + GstRTPPtDemux *ptdemux = GST_RTP_PT_DEMUX (element); + gboolean res = TRUE; + + if (ptdemux) { + ptdemux->srcpads = NULL; + ptdemux->last_pt = 0xFFFF; + } + + return res; +} + +/** + * Free resources for the object. + */ +static void +gst_rtp_pt_demux_release (GstElement * element) +{ + GstRTPPtDemux *ptdemux = GST_RTP_PT_DEMUX (element); + + if (ptdemux) { + /* note: GstElement's dispose() will handle the pads */ + g_slist_free (ptdemux->srcpads); + ptdemux->srcpads = NULL; + } +} + +static GstStateChangeReturn +gst_rtp_pt_demux_change_state (GstElement * element, GstStateChange transition) +{ + GstStateChangeReturn ret; + GstRTPPtDemux *ptdemux; + + ptdemux = GST_RTP_PT_DEMUX (element); + + switch (transition) { + case GST_STATE_CHANGE_NULL_TO_READY: + if (gst_rtp_pt_demux_setup (element) != TRUE) + ret = GST_STATE_CHANGE_FAILURE; + break; + case GST_STATE_CHANGE_READY_TO_PAUSED: + case GST_STATE_CHANGE_PAUSED_TO_PLAYING: + default: + break; + } + + ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); + + switch (transition) { + case GST_STATE_CHANGE_PLAYING_TO_PAUSED: + case GST_STATE_CHANGE_PAUSED_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_NULL: + gst_rtp_pt_demux_release (element); + break; + default: + break; + } + + return ret; +} diff --git a/gst/rtpmanager/gstrtpptdemux.h b/gst/rtpmanager/gstrtpptdemux.h new file mode 100644 index 00000000..93be3959 --- /dev/null +++ b/gst/rtpmanager/gstrtpptdemux.h @@ -0,0 +1,57 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * 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 __GST_RTP_PT_DEMUX_H__ +#define __GST_RTP_PT_DEMUX_H__ + +#include + +#define GST_TYPE_RTP_PT_DEMUX (gst_rtp_pt_demux_get_type()) +#define GST_RTP_PT_DEMUX(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_RTP_PT_DEMUX,GstRTPPtDemux)) +#define GST_RTP_PT_DEMUX_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_RTP_PT_DEMUX,GstRTPPtDemuxClass)) +#define GST_IS_RTP_PT_DEMUX(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_RTP_PT_DEMUX)) +#define GST_IS_RTP_PT_DEMUX_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_RTP_PT_DEMUX)) + +typedef struct _GstRTPPtDemux GstRTPPtDemux; +typedef struct _GstRTPPtDemuxClass GstRTPPtDemuxClass; +typedef struct _GstRTPPtDemuxPad GstRTPPtDemuxPad; + +struct _GstRTPPtDemux +{ + GstElement parent; /**< parent class */ + + GstPad *sink; /**< the sink pad */ + guint16 last_pt; /**< pt of the last packet 0xFFFF if none */ + GSList *srcpads; /**< a linked list of GstRTPPtDemuxPad objects */ +}; + +struct _GstRTPPtDemuxClass +{ + GstElementClass parent_class; + + /* signal emmited when a new PT is found from the incoming stream */ + void (*new_payload_type) (GstElement * element, gint pt, GstPad * pad); + + /* signal emitted when the payload type changes */ + void (*payload_type_change) (GstElement * element, gint pt); +}; + +GType gst_rtp_pt_demux_get_type (void); + +#endif /* __GST_RTP_PT_DEMUX_H__ */ diff --git a/gst/rtpmanager/gstrtpsession.c b/gst/rtpmanager/gstrtpsession.c new file mode 100644 index 00000000..47df756f --- /dev/null +++ b/gst/rtpmanager/gstrtpsession.c @@ -0,0 +1,453 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Library General Public + * License as published by the Free Software Foundation; either + * version 2 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Library General Public License for more details. + * + * You should have received a copy of the GNU Library General Public + * License along with this library; if not, write to the + * Free Software Foundation, Inc., 59 Temple Place - Suite 330, + * Boston, MA 02111-1307, USA. + */ + +/** + * SECTION:element-rtpsession + * @short_description: an RTP session manager + * @see_also: rtpjitterbuffer, rtpbin + * + * + * + * + * Example pipelines + * + * + * gst-launch -v filesrc location=sine.ogg ! oggdemux ! vorbisdec ! audioconvert ! alsasink + * + * + * + * + * Last reviewed on 2007-04-02 (0.10.6) + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif +#include "gstrtpsession.h" + +/* elementfactory information */ +static const GstElementDetails rtpsession_details = +GST_ELEMENT_DETAILS ("RTP Session", + "Filter/Editor/Video", + "Implement an RTP session", + "Wim Taymans "); + +/* sink pads */ +static GstStaticPadTemplate rtpsession_recv_rtp_sink_template = +GST_STATIC_PAD_TEMPLATE ("recv_rtp_sink", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtp") + ); + +static GstStaticPadTemplate rtpsession_recv_rtcp_sink_template = +GST_STATIC_PAD_TEMPLATE ("recv_rtcp_sink", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtcp") + ); + +static GstStaticPadTemplate rtpsession_send_rtp_sink_template = +GST_STATIC_PAD_TEMPLATE ("send_rtp_sink", + GST_PAD_SINK, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtp") + ); + +/* src pads */ +static GstStaticPadTemplate rtpsession_recv_rtp_src_template = +GST_STATIC_PAD_TEMPLATE ("recv_rtp_src", + GST_PAD_SRC, + GST_PAD_SOMETIMES, + GST_STATIC_CAPS ("application/x-rtp") + ); + +static GstStaticPadTemplate rtpsession_sync_src_template = +GST_STATIC_PAD_TEMPLATE ("sync_src", + GST_PAD_SRC, + GST_PAD_SOMETIMES, + GST_STATIC_CAPS ("application/x-rtcp") + ); + +static GstStaticPadTemplate rtpsession_send_rtp_src_template = +GST_STATIC_PAD_TEMPLATE ("send_rtp_src", + GST_PAD_SRC, + GST_PAD_SOMETIMES, + GST_STATIC_CAPS ("application/x-rtp") + ); + +static GstStaticPadTemplate rtpsession_rtcp_src_template = +GST_STATIC_PAD_TEMPLATE ("rtcp_src", + GST_PAD_SRC, + GST_PAD_REQUEST, + GST_STATIC_CAPS ("application/x-rtcp") + ); + +/* signals and args */ +enum +{ + /* FILL ME */ + LAST_SIGNAL +}; + +enum +{ + PROP_0 +}; + +/* GObject vmethods */ +static void gst_rtp_session_finalize (GObject * object); +static void gst_rtp_session_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec); +static void gst_rtp_session_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec); + +/* GstElement vmethods */ +static GstStateChangeReturn gst_rtp_session_change_state (GstElement * element, + GstStateChange transition); +static GstPad *gst_rtp_session_request_new_pad (GstElement * element, + GstPadTemplate * templ, const gchar * name); +static void gst_rtp_session_release_pad (GstElement * element, GstPad * pad); + +/*static guint gst_rtp_session_signals[LAST_SIGNAL] = { 0 }; */ + +GST_BOILERPLATE (GstRTPSession, gst_rtp_session, GstElement, GST_TYPE_ELEMENT); + +static void +gst_rtp_session_base_init (gpointer klass) +{ + GstElementClass *element_class = GST_ELEMENT_CLASS (klass); + + /* sink pads */ + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpsession_recv_rtp_sink_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpsession_recv_rtcp_sink_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpsession_send_rtp_sink_template)); + + /* src pads */ + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpsession_recv_rtp_src_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpsession_sync_src_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpsession_send_rtp_src_template)); + gst_element_class_add_pad_template (element_class, + gst_static_pad_template_get (&rtpsession_rtcp_src_template)); + + gst_element_class_set_details (element_class, &rtpsession_details); +} + +static void +gst_rtp_session_class_init (GstRTPSessionClass * klass) +{ + GObjectClass *gobject_class; + GstElementClass *gstelement_class; + + gobject_class = (GObjectClass *) klass; + gstelement_class = (GstElementClass *) klass; + + gobject_class->finalize = gst_rtp_session_finalize; + gobject_class->set_property = gst_rtp_session_set_property; + gobject_class->get_property = gst_rtp_session_get_property; + + gstelement_class->change_state = + GST_DEBUG_FUNCPTR (gst_rtp_session_change_state); + gstelement_class->request_new_pad = + GST_DEBUG_FUNCPTR (gst_rtp_session_request_new_pad); + gstelement_class->release_pad = + GST_DEBUG_FUNCPTR (gst_rtp_session_release_pad); +} + +static void +gst_rtp_session_init (GstRTPSession * rtpsession, GstRTPSessionClass * klass) +{ +} + +static void +gst_rtp_session_finalize (GObject * object) +{ + GstRTPSession *rtpsession; + + rtpsession = GST_RTP_SESSION (object); + + G_OBJECT_CLASS (parent_class)->finalize (object); +} + +static void +gst_rtp_session_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstRTPSession *rtpsession; + + rtpsession = GST_RTP_SESSION (object); + + switch (prop_id) { + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +gst_rtp_session_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstRTPSession *rtpsession; + + rtpsession = GST_RTP_SESSION (object); + + switch (prop_id) { + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static GstStateChangeReturn +gst_rtp_session_change_state (GstElement * element, GstStateChange transition) +{ + GstStateChangeReturn res; + GstRTPSession *rtpsession; + + rtpsession = GST_RTP_SESSION (element); + + switch (transition) { + case GST_STATE_CHANGE_NULL_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_PLAYING: + break; + default: + break; + } + + res = parent_class->change_state (element, transition); + + switch (transition) { + case GST_STATE_CHANGE_PLAYING_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_READY: + break; + case GST_STATE_CHANGE_READY_TO_NULL: + break; + default: + break; + } + return res; +} + +/* receive a packet from a sender, send it to the RTP session manager and + * forward the packet on the rtp_src pad + */ +static GstFlowReturn +gst_rtp_session_chain_recv_rtp (GstPad * pad, GstBuffer * buffer) +{ + GstRTPSession *rtpsession; + GstFlowReturn ret; + + rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad)); + + /* FIXME, do something */ + ret = gst_pad_push (rtpsession->recv_rtp_src, buffer); + + gst_object_unref (rtpsession); + + return ret; +} + +/* Receive an RTCP packet from a sender, send it to the RTP session manager and + * forward the SR packets to the sync_src pad. + */ +static GstFlowReturn +gst_rtp_session_chain_recv_rtcp (GstPad * pad, GstBuffer * buffer) +{ + GstRTPSession *rtpsession; + GstFlowReturn ret; + + rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad)); + + /* FIXME, do something */ + ret = gst_pad_push (rtpsession->sync_src, buffer); + + gst_object_unref (rtpsession); + + return ret; +} + +/* Recieve an RTP packet to be send to the receivers, send to RTP session + * manager and forward to send_rtp_src. + */ +static GstFlowReturn +gst_rtp_session_chain_send_rtp (GstPad * pad, GstBuffer * buffer) +{ + GstRTPSession *rtpsession; + GstFlowReturn ret; + + rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad)); + + /* FIXME, do something */ + ret = gst_pad_push (rtpsession->send_rtp_src, buffer); + + gst_object_unref (rtpsession); + + return ret; +} + + +/* Create sinkpad to receive RTP packets from senders. This will also create a + * srcpad for the RTP packets. + */ +static GstPad * +create_recv_rtp_sink (GstRTPSession * rtpsession) +{ + rtpsession->recv_rtp_sink = + gst_pad_new_from_static_template (&rtpsession_recv_rtp_sink_template, + NULL); + gst_pad_set_chain_function (rtpsession->recv_rtp_sink, + gst_rtp_session_chain_recv_rtp); + gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), + rtpsession->recv_rtp_sink); + + rtpsession->recv_rtp_src = + gst_pad_new_from_static_template (&rtpsession_recv_rtp_src_template, + NULL); + gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->recv_rtp_src); + + return rtpsession->recv_rtp_sink; +} + +/* Create a sinkpad to receive RTCP messages from senders, this will also create a + * sync_src pad for the SR packets. + */ +static GstPad * +create_recv_rtcp_sink (GstRTPSession * rtpsession) +{ + rtpsession->recv_rtcp_sink = + gst_pad_new_from_static_template (&rtpsession_recv_rtcp_sink_template, + NULL); + gst_pad_set_chain_function (rtpsession->recv_rtcp_sink, + gst_rtp_session_chain_recv_rtcp); + gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), + rtpsession->recv_rtcp_sink); + + rtpsession->sync_src = + gst_pad_new_from_static_template (&rtpsession_sync_src_template, NULL); + gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->sync_src); + + return rtpsession->recv_rtcp_sink; +} + +/* Create a sinkpad to receive RTP packets for receivers. This will also create a + * send_rtp_src pad. + */ +static GstPad * +create_send_rtp_sink (GstRTPSession * rtpsession) +{ + rtpsession->send_rtp_sink = + gst_pad_new_from_static_template (&rtpsession_send_rtp_sink_template, + NULL); + gst_pad_set_chain_function (rtpsession->send_rtp_sink, + gst_rtp_session_chain_send_rtp); + gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), + rtpsession->recv_rtcp_sink); + + rtpsession->send_rtp_src = + gst_pad_new_from_static_template (&rtpsession_send_rtp_src_template, + NULL); + gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->send_rtp_src); + + return rtpsession->send_rtp_sink; +} + +/* Create a srcpad with the RTCP packets to send out. + * This pad will be driven by the RTP session manager when it wants to send out + * RTCP packets. + */ +static GstPad * +create_rtcp_src (GstRTPSession * rtpsession) +{ + rtpsession->rtcp_src = + gst_pad_new_from_static_template (&rtpsession_rtcp_src_template, NULL); + gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->rtcp_src); + + return rtpsession->rtcp_src; +} + +static GstPad * +gst_rtp_session_request_new_pad (GstElement * element, + GstPadTemplate * templ, const gchar * name) +{ + GstRTPSession *rtpsession; + GstElementClass *klass; + GstPad *result; + + g_return_val_if_fail (templ != NULL, NULL); + g_return_val_if_fail (GST_IS_RTP_SESSION (element), NULL); + + rtpsession = GST_RTP_SESSION (element); + klass = GST_ELEMENT_GET_CLASS (element); + + /* figure out the template */ + if (templ == gst_element_class_get_pad_template (klass, "recv_rtp_sink")) { + if (rtpsession->recv_rtp_sink != NULL) + goto exists; + + result = create_recv_rtp_sink (rtpsession); + } else if (templ == gst_element_class_get_pad_template (klass, + "recv_rtcp_sink")) { + if (rtpsession->recv_rtcp_sink != NULL) + goto exists; + + result = create_recv_rtcp_sink (rtpsession); + } else if (templ == gst_element_class_get_pad_template (klass, + "send_rtp_sink")) { + if (rtpsession->send_rtp_sink != NULL) + goto exists; + + result = create_send_rtp_sink (rtpsession); + } else if (templ == gst_element_class_get_pad_template (klass, "rtcp_src")) { + if (rtpsession->rtcp_src != NULL) + goto exists; + + result = create_rtcp_src (rtpsession); + } else + goto wrong_template; + + return result; + + /* ERRORS */ +wrong_template: + { + g_warning ("rtpsession: this is not our template"); + return NULL; + } +exists: + { + g_warning ("rtpsession: pad already requested"); + return NULL; + } +} + +static void +gst_rtp_session_release_pad (GstElement * element, GstPad * pad) +{ +} diff --git a/gst/rtpmanager/gstrtpsession.h b/gst/rtpmanager/gstrtpsession.h new file mode 100644 index 00000000..8b343064 --- /dev/null +++ b/gst/rtpmanager/gstrtpsession.h @@ -0,0 +1,62 @@ +/* GStreamer + * Copyright (C) <2007> Wim Taymans + * + * 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 __GST_RTP_SESSION_H__ +#define __GST_RTP_SESSION_H__ + +#include + +#define GST_TYPE_RTP_SESSION \ + (gst_rtp_session_get_type()) +#define GST_RTP_SESSION(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_RTP_SESSION,GstRTPSession)) +#define GST_RTP_SESSION_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_RTP_SESSION,GstRTPSessionClass)) +#define GST_IS_RTP_SESSION(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_RTP_SESSION)) +#define GST_IS_RTP_SESSION_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_RTP_SESSION)) + +typedef struct _GstRTPSession GstRTPSession; +typedef struct _GstRTPSessionClass GstRTPSessionClass; +typedef struct _GstRTPSessionPrivate GstRTPSessionPrivate; + +struct _GstRTPSession { + GstElement element; + + /*< private >*/ + GstPad *recv_rtp_sink; + GstPad *recv_rtcp_sink; + GstPad *send_rtp_sink; + + GstPad *recv_rtp_src; + GstPad *sync_src; + GstPad *send_rtp_src; + GstPad *rtcp_src; + + GstRTPSessionPrivate *priv; +}; + +struct _GstRTPSessionClass { + GstElementClass parent_class; +}; + +GType gst_rtp_session_get_type (void); + +#endif /* __GST_RTP_SESSION_H__ */ -- cgit