mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-11-20 08:41:07 +00:00
Code Issues 1 Projects Releases Wiki Activity

rtprtxsend: Allow SSRC-multiplexing and multiple payload types in the original stream

Browse source 
Conflicts:
	tests/examples/rtp/server-rtpaux.c
This commit is contained in:
Wim Taymans 2014-01-02 15:18:52 +01:00
parent 41285697ac
commit 130ad1b1fa
6 changed files with 252 additions and 100 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

272
gst/rtpmanager/gstrtprtxsend.c
View file

@ -42,6 +42,7 @@
#include <gst/gst.h> #include <gst/gst.h>
#include <gst/rtp/gstrtpbuffer.h> #include <gst/rtp/gstrtpbuffer.h>
#include <string.h> #include <string.h>
#include <stdlib.h>
#include "gstrtprtxsend.h" #include "gstrtprtxsend.h"
@ -55,8 +56,8 @@ GST_DEBUG_CATEGORY_STATIC (gst_rtp_rtx_send_debug);
enum enum
{ {
PROP_0, PROP_0,
PROP_RTX_SSRC, PROP_SSRC_MAP,
PROP_RTX_PAYLOAD_TYPE, PROP_PAYLOAD_TYPE_MAP,
PROP_MAX_SIZE_TIME, PROP_MAX_SIZE_TIME,
PROP_MAX_SIZE_PACKETS, PROP_MAX_SIZE_PACKETS,
PROP_NUM_RTX_REQUESTS, PROP_NUM_RTX_REQUESTS,
@ -108,6 +109,35 @@ buffer_queue_item_free (BufferQueueItem * item)
g_free (item); g_free (item);
} }
typedef struct
{
guint32 rtx_ssrc;
guint16 next_seqnum;
gint clock_rate;
/* history of rtp packets */
GSequence *queue;
} SSRCRtxData;
static SSRCRtxData *
ssrc_rtx_data_new (guint32 rtx_ssrc)
{
SSRCRtxData *data = g_new0 (SSRCRtxData, 1);
data->rtx_ssrc = rtx_ssrc;
data->next_seqnum = g_random_int_range (0, G_MAXUINT16);
data->queue = g_sequence_new ((GDestroyNotify) buffer_queue_item_free);
return data;
}
static void
ssrc_rtx_data_free (SSRCRtxData * data)
{
g_sequence_free (data->queue);
g_free (data);
}
static void static void
gst_rtp_rtx_send_class_init (GstRtpRtxSendClass * klass) gst_rtp_rtx_send_class_init (GstRtpRtxSendClass * klass)
{ {
@ -121,17 +151,16 @@ gst_rtp_rtx_send_class_init (GstRtpRtxSendClass * klass)
gobject_class->set_property = gst_rtp_rtx_send_set_property; gobject_class->set_property = gst_rtp_rtx_send_set_property;
gobject_class->finalize = gst_rtp_rtx_send_finalize; gobject_class->finalize = gst_rtp_rtx_send_finalize;
g_object_class_install_property (gobject_class, PROP_RTX_SSRC, g_object_class_install_property (gobject_class, PROP_SSRC_MAP,
g_param_spec_uint ("rtx-ssrc", "Retransmission SSRC", g_param_spec_boxed ("ssrc-map", "SSRC Map",
"SSRC of the retransmission stream for SSRC-multiplexed mode " "Map of SSRCs to their retransmission SSRCs for SSRC-multiplexed mode"
"(default = random)", 0, G_MAXUINT, -1, " (default = random)", GST_TYPE_STRUCTURE,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); G_PARAM_WRITABLE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_RTX_PAYLOAD_TYPE, g_object_class_install_property (gobject_class, PROP_PAYLOAD_TYPE_MAP,
g_param_spec_uint ("rtx-payload-type", "RTX Payload Type", g_param_spec_boxed ("payload-type-map", "Payload Type Map",
"Payload type of the retransmission stream (fmtp in SDP)", 0, "Map of original payload types to their retransmission payload types",
G_MAXUINT, DEFAULT_RTX_PAYLOAD_TYPE, GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME, g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
g_param_spec_uint ("max-size-time", "Max Size Time", g_param_spec_uint ("max-size-time", "Max Size Time",
@ -172,13 +201,10 @@ static void
gst_rtp_rtx_send_reset (GstRtpRtxSend * rtx, gboolean full) gst_rtp_rtx_send_reset (GstRtpRtxSend * rtx, gboolean full)
{ {
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
g_sequence_remove_range (g_sequence_get_begin_iter (rtx->queue),
g_sequence_get_end_iter (rtx->queue));
g_queue_foreach (rtx->pending, (GFunc) gst_buffer_unref, NULL); g_queue_foreach (rtx->pending, (GFunc) gst_buffer_unref, NULL);
g_queue_clear (rtx->pending); g_queue_clear (rtx->pending);
rtx->master_ssrc = 0; g_hash_table_remove_all (rtx->ssrc_data);
rtx->next_seqnum = g_random_int_range (0, G_MAXUINT16); g_hash_table_remove_all (rtx->rtx_ssrcs);
rtx->rtx_ssrc = g_random_int ();
rtx->num_rtx_requests = 0; rtx->num_rtx_requests = 0;
rtx->num_rtx_packets = 0; rtx->num_rtx_packets = 0;
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
@ -190,7 +216,14 @@ gst_rtp_rtx_send_finalize (GObject * object)
GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (object); GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (object);
gst_rtp_rtx_send_reset (rtx, TRUE); gst_rtp_rtx_send_reset (rtx, TRUE);
g_sequence_free (rtx->queue);
g_hash_table_unref (rtx->ssrc_data);
g_hash_table_unref (rtx->rtx_ssrcs);
if (rtx->external_ssrc_map)
gst_structure_free (rtx->external_ssrc_map);
g_hash_table_unref (rtx->rtx_pt_map);
if (rtx->pending_rtx_pt_map)
gst_structure_free (rtx->pending_rtx_pt_map);
g_queue_free (rtx->pending); g_queue_free (rtx->pending);
g_mutex_clear (&rtx->lock); g_mutex_clear (&rtx->lock);
@ -222,32 +255,60 @@ gst_rtp_rtx_send_init (GstRtpRtxSend * rtx)
GST_DEBUG_FUNCPTR (gst_rtp_rtx_send_chain)); GST_DEBUG_FUNCPTR (gst_rtp_rtx_send_chain));
gst_element_add_pad (GST_ELEMENT (rtx), rtx->sinkpad); gst_element_add_pad (GST_ELEMENT (rtx), rtx->sinkpad);
rtx->queue = g_sequence_new ((GDestroyNotify) buffer_queue_item_free);
rtx->pending = g_queue_new ();
g_mutex_init (&rtx->lock); g_mutex_init (&rtx->lock);
rtx->pending = g_queue_new ();
rtx->next_seqnum = g_random_int_range (0, G_MAXUINT16); rtx->ssrc_data = g_hash_table_new_full (g_direct_hash, g_direct_equal,
rtx->rtx_ssrc = g_random_int (); NULL, (GDestroyNotify) ssrc_rtx_data_free);
rtx->rtx_ssrcs = g_hash_table_new (g_direct_hash, g_direct_equal);
rtx->rtx_pt_map = g_hash_table_new (g_direct_hash, g_direct_equal);
rtx->rtx_pt_map_changed = FALSE;
rtx->max_size_time = DEFAULT_MAX_SIZE_TIME; rtx->max_size_time = DEFAULT_MAX_SIZE_TIME;
rtx->max_size_packets = DEFAULT_MAX_SIZE_PACKETS; rtx->max_size_packets = DEFAULT_MAX_SIZE_PACKETS;
} }
static guint32 static guint32
choose_ssrc (GstRtpRtxSend * rtx) gst_rtp_rtx_send_choose_ssrc (GstRtpRtxSend * rtx, guint32 choice,
gboolean consider_choice)
{ {
guint32 ssrc; guint32 ssrc = consider_choice ? choice : g_random_int ();
while (TRUE) { /* make sure to be different than any other */
while (g_hash_table_contains (rtx->ssrc_data, GUINT_TO_POINTER (ssrc)) ||
g_hash_table_contains (rtx->rtx_ssrcs, GUINT_TO_POINTER (ssrc))) {
ssrc = g_random_int (); ssrc = g_random_int ();
/* make sure to be different than master */
if (ssrc != rtx->master_ssrc)
break;
} }
return ssrc; return ssrc;
} }
static SSRCRtxData *
gst_rtp_rtx_send_get_ssrc_data (GstRtpRtxSend * rtx, guint32 ssrc)
{
SSRCRtxData *data;
guint32 rtx_ssrc;
gboolean consider = FALSE;
if (G_UNLIKELY (!g_hash_table_contains (rtx->ssrc_data,
GUINT_TO_POINTER (ssrc)))) {
if (rtx->external_ssrc_map) {
gchar *ssrc_str;
ssrc_str = g_strdup_printf ("%" G_GUINT32_FORMAT, ssrc);
consider = gst_structure_get_uint (rtx->external_ssrc_map, ssrc_str,
&rtx_ssrc);
g_free (ssrc_str);
}
rtx_ssrc = gst_rtp_rtx_send_choose_ssrc (rtx, rtx_ssrc, consider);
data = ssrc_rtx_data_new (rtx_ssrc);
g_hash_table_insert (rtx->ssrc_data, GUINT_TO_POINTER (ssrc), data);
g_hash_table_insert (rtx->rtx_ssrcs, GUINT_TO_POINTER (rtx_ssrc),
GUINT_TO_POINTER (ssrc));
} else {
data = g_hash_table_lookup (rtx->ssrc_data, GUINT_TO_POINTER (ssrc));
}
return data;
}
static gint static gint
buffer_queue_items_cmp (BufferQueueItem * a, BufferQueueItem * b, buffer_queue_items_cmp (BufferQueueItem * a, BufferQueueItem * b,
gpointer user_data) gpointer user_data)
@ -271,7 +332,7 @@ gst_rtp_rtx_send_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
/* This event usually comes from the downstream gstrtpsession */ /* This event usually comes from the downstream gstrtpsession */
if (gst_structure_has_name (s, "GstRTPRetransmissionRequest")) { if (gst_structure_has_name (s, "GstRTPRetransmissionRequest")) {
guint32 seqnum = 0; guint seqnum = 0;
guint ssrc = 0; guint ssrc = 0;
/* retrieve seqnum of the packet that need to be restransmisted */ /* retrieve seqnum of the packet that need to be restransmisted */
@ -283,20 +344,23 @@ gst_rtp_rtx_send_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
ssrc = -1; ssrc = -1;
GST_DEBUG_OBJECT (rtx, GST_DEBUG_OBJECT (rtx,
"request seqnum: %" G_GUINT16_FORMAT ", ssrc: %" G_GUINT32_FORMAT, "request seqnum: %" G_GUINT32_FORMAT ", ssrc: %" G_GUINT32_FORMAT,
seqnum, ssrc); seqnum, ssrc);
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
/* check if request is for us */ /* check if request is for us */
if (rtx->master_ssrc == ssrc) { if (g_hash_table_contains (rtx->ssrc_data, GUINT_TO_POINTER (ssrc))) {
SSRCRtxData *data;
GSequenceIter *iter; GSequenceIter *iter;
BufferQueueItem search_item; BufferQueueItem search_item;
/* update statistics */ /* update statistics */
++rtx->num_rtx_requests; ++rtx->num_rtx_requests;
data = gst_rtp_rtx_send_get_ssrc_data (rtx, ssrc);
search_item.seqnum = seqnum; search_item.seqnum = seqnum;
iter = g_sequence_lookup (rtx->queue, &search_item, iter = g_sequence_lookup (data->queue, &search_item,
(GCompareDataFunc) buffer_queue_items_cmp, NULL); (GCompareDataFunc) buffer_queue_items_cmp, NULL);
if (iter) { if (iter) {
BufferQueueItem *item = g_sequence_get (iter); BufferQueueItem *item = g_sequence_get (iter);
@ -321,16 +385,20 @@ gst_rtp_rtx_send_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
/* choose another ssrc for our retransmited stream */ /* choose another ssrc for our retransmited stream */
if (ssrc == rtx->rtx_ssrc) { if (g_hash_table_contains (rtx->rtx_ssrcs, GUINT_TO_POINTER (ssrc))) {
rtx->rtx_ssrc = choose_ssrc (rtx); guint master_ssrc;
SSRCRtxData *data;
/* clear buffers we already saved */ master_ssrc = GPOINTER_TO_UINT (g_hash_table_lookup (rtx->rtx_ssrcs,
g_sequence_remove_range (g_sequence_get_begin_iter (rtx->queue), GUINT_TO_POINTER (ssrc)));
g_sequence_get_end_iter (rtx->queue)); data = gst_rtp_rtx_send_get_ssrc_data (rtx, master_ssrc);
/* clear buffers that are about to be retransmited */ /* change rtx_ssrc and update the reverse map */
g_queue_foreach (rtx->pending, (GFunc) gst_buffer_unref, NULL); data->rtx_ssrc = gst_rtp_rtx_send_choose_ssrc (rtx, 0, FALSE);
g_queue_clear (rtx->pending); g_hash_table_remove (rtx->rtx_ssrcs, GUINT_TO_POINTER (ssrc));
g_hash_table_insert (rtx->rtx_ssrcs,
GUINT_TO_POINTER (data->rtx_ssrc),
GUINT_TO_POINTER (master_ssrc));
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
@ -340,6 +408,16 @@ gst_rtp_rtx_send_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
gst_event_unref (event); gst_event_unref (event);
res = TRUE; res = TRUE;
} else { } else {
/* if master ssrc has collided, remove it from our data, as it
* is not going to be used any longer */
if (g_hash_table_contains (rtx->ssrc_data, GUINT_TO_POINTER (ssrc))) {
SSRCRtxData *data;
data = gst_rtp_rtx_send_get_ssrc_data (rtx, ssrc);
g_hash_table_remove (rtx->rtx_ssrcs,
GUINT_TO_POINTER (data->rtx_ssrc));
g_hash_table_remove (rtx->ssrc_data, GUINT_TO_POINTER (ssrc));
}
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
/* forward event to payloader in case collided ssrc is /* forward event to payloader in case collided ssrc is
@ -368,14 +446,19 @@ gst_rtp_rtx_send_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
{ {
GstCaps *caps; GstCaps *caps;
GstStructure *s; GstStructure *s;
guint ssrc;
SSRCRtxData *data;
gst_event_parse_caps (event, &caps); gst_event_parse_caps (event, &caps);
g_assert (gst_caps_is_fixed (caps)); g_assert (gst_caps_is_fixed (caps));
s = gst_caps_get_structure (caps, 0); s = gst_caps_get_structure (caps, 0);
gst_structure_get_int (s, "clock-rate", &rtx->clock_rate); gst_structure_get_uint (s, "ssrc", &ssrc);
data = gst_rtp_rtx_send_get_ssrc_data (rtx, ssrc);
gst_structure_get_int (s, "clock-rate", &data->clock_rate);
GST_DEBUG_OBJECT (rtx, "got clock-rate from caps: %d", rtx->clock_rate); GST_DEBUG_OBJECT (rtx, "got clock-rate from caps: %d for ssrc: %u",
data->clock_rate, ssrc);
break; break;
} }
@ -385,9 +468,25 @@ gst_rtp_rtx_send_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
return gst_pad_event_default (pad, parent, event); return gst_pad_event_default (pad, parent, event);
} }
static gboolean
structure_to_hash_table (GQuark field_id, const GValue * value, gpointer hash)
{
const gchar *field_str;
guint field_uint;
guint value_uint;
field_str = g_quark_to_string (field_id);
field_uint = atoi (field_str);
value_uint = g_value_get_uint (value);
g_hash_table_insert ((GHashTable *) hash, GUINT_TO_POINTER (field_uint),
GUINT_TO_POINTER (value_uint));
return TRUE;
}
/* like rtp_jitter_buffer_get_ts_diff() */ /* like rtp_jitter_buffer_get_ts_diff() */
static guint32 static guint32
gst_rtp_rtx_send_get_ts_diff (GstRtpRtxSend * self) gst_rtp_rtx_send_get_ts_diff (SSRCRtxData * data)
{ {
guint64 high_ts, low_ts; guint64 high_ts, low_ts;
BufferQueueItem *high_buf, *low_buf; BufferQueueItem *high_buf, *low_buf;
@ -395,8 +494,8 @@ gst_rtp_rtx_send_get_ts_diff (GstRtpRtxSend * self)
high_buf = high_buf =
g_sequence_get (g_sequence_iter_prev (g_sequence_get_end_iter g_sequence_get (g_sequence_iter_prev (g_sequence_get_end_iter
(self->queue))); (data->queue)));
low_buf = g_sequence_get (g_sequence_get_begin_iter (self->queue)); low_buf = g_sequence_get (g_sequence_get_begin_iter (data->queue));
if (!high_buf || !low_buf || high_buf == low_buf) if (!high_buf || !low_buf || high_buf == low_buf)
return 0; return 0;
@ -412,15 +511,14 @@ gst_rtp_rtx_send_get_ts_diff (GstRtpRtxSend * self)
} }
/* return value in ms instead of clock ticks */ /* return value in ms instead of clock ticks */
return (guint32) gst_util_uint64_scale_int (result, 1000, self->clock_rate); return (guint32) gst_util_uint64_scale_int (result, 1000, data->clock_rate);
} }
/* Copy fixed header and extension. Add OSN before to copy payload /* Copy fixed header and extension. Add OSN before to copy payload
* Copy memory to avoid to manually copy each rtp buffer field. * Copy memory to avoid to manually copy each rtp buffer field.
*/ */
static GstBuffer * static GstBuffer *
_gst_rtp_rtx_buffer_new (GstBuffer * buffer, guint32 ssrc, guint16 seqnum, _gst_rtp_rtx_buffer_new (GstRtpRtxSend * rtx, GstBuffer * buffer)
guint8 fmtp)
{ {
GstMemory *mem = NULL; GstMemory *mem = NULL;
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
@ -428,9 +526,25 @@ _gst_rtp_rtx_buffer_new (GstBuffer * buffer, guint32 ssrc, guint16 seqnum,
GstBuffer *new_buffer = gst_buffer_new (); GstBuffer *new_buffer = gst_buffer_new ();
GstMapInfo map; GstMapInfo map;
guint payload_len = 0; guint payload_len = 0;
SSRCRtxData *data;
guint32 ssrc;
guint16 seqnum;
guint8 fmtp;
gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp); gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp);
/* get needed data from GstRtpRtxSend */
ssrc = gst_rtp_buffer_get_ssrc (&rtp);
data = gst_rtp_rtx_send_get_ssrc_data (rtx, ssrc);
ssrc = data->rtx_ssrc;
seqnum = data->next_seqnum++;
fmtp = GPOINTER_TO_UINT (g_hash_table_lookup (rtx->rtx_pt_map,
GUINT_TO_POINTER (gst_rtp_buffer_get_payload_type (&rtp))));
GST_DEBUG_OBJECT (rtx,
"retransmit seqnum: %" G_GUINT16_FORMAT ", ssrc: %" G_GUINT32_FORMAT,
seqnum, ssrc);
/* gst_rtp_buffer_map does not map the payload so do it now */ /* gst_rtp_buffer_map does not map the payload so do it now */
gst_rtp_buffer_get_payload (&rtp); gst_rtp_buffer_get_payload (&rtp);
@ -476,17 +590,11 @@ _gst_rtp_rtx_buffer_new (GstBuffer * buffer, guint32 ssrc, guint16 seqnum,
} }
/* push pending retransmission packet. /* push pending retransmission packet.
* it constructs rtx packet from original paclets */ * it constructs rtx packet from original packets */
static void static void
do_push (GstBuffer * buffer, GstRtpRtxSend * rtx) do_push (GstBuffer * buffer, GstRtpRtxSend * rtx)
{ {
/* RFC4588 two streams multiplexed by sending them in the same session using gst_pad_push (rtx->srcpad, _gst_rtp_rtx_buffer_new (rtx, buffer));
* different SSRC values, i.e., SSRC-multiplexing. */
GST_DEBUG_OBJECT (rtx,
"retransmit seqnum: %" G_GUINT16_FORMAT ", ssrc: %" G_GUINT32_FORMAT,
rtx->next_seqnum, rtx->rtx_ssrc);
gst_pad_push (rtx->srcpad, _gst_rtp_rtx_buffer_new (buffer, rtx->rtx_ssrc,
rtx->next_seqnum++, rtx->rtx_payload_type));
} }
static GstFlowReturn static GstFlowReturn
@ -497,6 +605,7 @@ gst_rtp_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
GQueue *pending = NULL; GQueue *pending = NULL;
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
BufferQueueItem *item; BufferQueueItem *item;
SSRCRtxData *data;
guint16 seqnum; guint16 seqnum;
guint32 ssrc, rtptime; guint32 ssrc, rtptime;
@ -511,27 +620,23 @@ gst_rtp_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
/* retrieve master stream ssrc */ data = gst_rtp_rtx_send_get_ssrc_data (rtx, ssrc);
rtx->master_ssrc = ssrc;
/* check if our initial aux ssrc is equal to master */
if (rtx->rtx_ssrc == rtx->master_ssrc)
choose_ssrc (rtx);
/* add current rtp buffer to queue history */ /* add current rtp buffer to queue history */
item = g_new0 (BufferQueueItem, 1); item = g_new0 (BufferQueueItem, 1);
item->seqnum = seqnum; item->seqnum = seqnum;
item->timestamp = rtptime; item->timestamp = rtptime;
item->buffer = gst_buffer_ref (buffer); item->buffer = gst_buffer_ref (buffer);
g_sequence_append (rtx->queue, item); g_sequence_append (data->queue, item);
/* remove oldest packets from history if they are too many */ /* remove oldest packets from history if they are too many */
if (rtx->max_size_packets) { if (rtx->max_size_packets) {
while (g_sequence_get_length (rtx->queue) > rtx->max_size_packets) while (g_sequence_get_length (data->queue) > rtx->max_size_packets)
g_sequence_remove (g_sequence_get_begin_iter (rtx->queue)); g_sequence_remove (g_sequence_get_begin_iter (data->queue));
} }
if (rtx->max_size_time) { if (rtx->max_size_time) {
while (gst_rtp_rtx_send_get_ts_diff (rtx) > rtx->max_size_time) while (gst_rtp_rtx_send_get_ts_diff (data) > rtx->max_size_time)
g_sequence_remove (g_sequence_get_begin_iter (rtx->queue)); g_sequence_remove (g_sequence_get_begin_iter (data->queue));
} }
/* within lock, get packets that have to be retransmited */ /* within lock, get packets that have to be retransmited */
@ -544,7 +649,12 @@ gst_rtp_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
} }
/* transfer payload type while holding the lock */ /* transfer payload type while holding the lock */
rtx->rtx_payload_type = rtx->rtx_payload_type_pending; if (rtx->rtx_pt_map_changed) {
g_hash_table_remove_all (rtx->rtx_pt_map);
gst_structure_foreach (rtx->pending_rtx_pt_map, structure_to_hash_table,
rtx->rtx_pt_map);
rtx->rtx_pt_map_changed = FALSE;
}
/* no need to hold the lock to push rtx packets */ /* no need to hold the lock to push rtx packets */
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
@ -557,7 +667,7 @@ gst_rtp_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
GST_LOG_OBJECT (rtx, GST_LOG_OBJECT (rtx,
"push seqnum: %" G_GUINT16_FORMAT ", ssrc: %" G_GUINT32_FORMAT, seqnum, "push seqnum: %" G_GUINT16_FORMAT ", ssrc: %" G_GUINT32_FORMAT, seqnum,
rtx->master_ssrc); ssrc);
/* push current rtp packet */ /* push current rtp packet */
ret = gst_pad_push (rtx->srcpad, buffer); ret = gst_pad_push (rtx->srcpad, buffer);
@ -572,14 +682,9 @@ gst_rtp_rtx_send_get_property (GObject * object,
GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (object); GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (object);
switch (prop_id) { switch (prop_id) {
case PROP_RTX_SSRC: case PROP_PAYLOAD_TYPE_MAP:
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
g_value_set_uint (value, rtx->rtx_ssrc); g_value_set_boxed (value, rtx->pending_rtx_pt_map);
g_mutex_unlock (&rtx->lock);
break;
case PROP_RTX_PAYLOAD_TYPE:
g_mutex_lock (&rtx->lock);
g_value_set_uint (value, rtx->rtx_payload_type_pending);
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
break; break;
case PROP_MAX_SIZE_TIME: case PROP_MAX_SIZE_TIME:
@ -615,14 +720,19 @@ gst_rtp_rtx_send_set_property (GObject * object,
GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (object); GstRtpRtxSend *rtx = GST_RTP_RTX_SEND (object);
switch (prop_id) { switch (prop_id) {
case PROP_RTX_SSRC: case PROP_SSRC_MAP:
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
rtx->rtx_ssrc = g_value_get_uint (value); if (rtx->external_ssrc_map)
gst_structure_free (rtx->external_ssrc_map);
rtx->external_ssrc_map = g_value_dup_boxed (value);
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
break; break;
case PROP_RTX_PAYLOAD_TYPE: case PROP_PAYLOAD_TYPE_MAP:
g_mutex_lock (&rtx->lock); g_mutex_lock (&rtx->lock);
rtx->rtx_payload_type_pending = g_value_get_uint (value); if (rtx->pending_rtx_pt_map)
gst_structure_free (rtx->pending_rtx_pt_map);
rtx->pending_rtx_pt_map = g_value_dup_boxed (value);
rtx->rtx_pt_map_changed = TRUE;
g_mutex_unlock (&rtx->lock); g_mutex_unlock (&rtx->lock);
break; break;
case PROP_MAX_SIZE_TIME: case PROP_MAX_SIZE_TIME:

20
gst/rtpmanager/gstrtprtxsend.h
View file

@ -47,20 +47,22 @@ struct _GstRtpRtxSend
GMutex lock; GMutex lock;
/* history of rtp packets */
GSequence *queue;
/* rtp packets that will be pushed upon next buffer */ /* rtp packets that will be pushed upon next buffer */
GQueue *pending; GQueue *pending;
guint32 master_ssrc; /* ssrc -> SSRCRtxData */
guint32 rtx_ssrc; GHashTable *ssrc_data;
guint16 next_seqnum; /* rtx ssrc -> master ssrc */
guint8 rtx_payload_type; GHashTable *rtx_ssrcs;
gint clock_rate; /* master ssrc -> rtx ssrc (property) */
GstStructure *external_ssrc_map;
/* retreived from SDP */ /* orig pt (uint) -> rtx pt (uint) */
guint rtx_payload_type_pending; GHashTable *rtx_pt_map;
/* orig pt (string) -> rtx pt (uint) */
GstStructure *pending_rtx_pt_map;
gboolean rtx_pt_map_changed;
/* buffering control properties */ /* buffering control properties */
guint max_size_time; guint max_size_time;

6
tests/check/elements/rtpaux.c
View file

@ -185,6 +185,7 @@ GST_START_TEST (test_simple_rtpbin_aux)
GstBus *busreceive; GstBus *busreceive;
gboolean res; gboolean res;
GstCaps *rtpcaps = NULL; GstCaps *rtpcaps = NULL;
GstStructure *pt_map;
GstStateChangeReturn state_res = GST_STATE_CHANGE_FAILURE; GstStateChangeReturn state_res = GST_STATE_CHANGE_FAILURE;
GstPad *srcpad = NULL; GstPad *srcpad = NULL;
guint nb_rtx_send_packets = 0; guint nb_rtx_send_packets = 0;
@ -254,10 +255,13 @@ GST_START_TEST (test_simple_rtpbin_aux)
g_signal_connect (rtpbinreceive, "pad-added", g_signal_connect (rtpbinreceive, "pad-added",
G_CALLBACK (on_rtpbinreceive_pad_added), rtpdepayloader); G_CALLBACK (on_rtpbinreceive_pad_added), rtpdepayloader);
pt_map = gst_structure_new ("application/x-rtp-pt-map",
"96", G_TYPE_UINT, 99, NULL);
g_object_set (rtppayloader, "pt", 96, NULL); g_object_set (rtppayloader, "pt", 96, NULL);
g_object_set (rtppayloader, "seqnum-offset", 1, NULL); g_object_set (rtppayloader, "seqnum-offset", 1, NULL);
g_object_set (rtprtxsend, "rtx-payload-type", 99, NULL); g_object_set (rtprtxsend, "payload-type-map", pt_map, NULL);
g_object_set (rtprtxreceive, "rtx-payload-types", "99:111:125", NULL); g_object_set (rtprtxreceive, "rtx-payload-types", "99:111:125", NULL);
gst_structure_free (pt_map);
/* set rtp aux receive */ /* set rtp aux receive */
g_signal_connect (rtpbinreceive, "request-aux-receiver", (GCallback) g_signal_connect (rtpbinreceive, "request-aux-receiver", (GCallback)

6
tests/check/elements/rtpcollision.c
View file

@ -343,6 +343,7 @@ GST_START_TEST (test_rtx_ssrc_collision)
GstPad *fake_udp_sinkpad = NULL; GstPad *fake_udp_sinkpad = NULL;
GstPad *rtcp_srcpad = NULL; GstPad *rtcp_srcpad = NULL;
GstStateChangeReturn state_res = GST_STATE_CHANGE_FAILURE; GstStateChangeReturn state_res = GST_STATE_CHANGE_FAILURE;
GstStructure *pt_map;
GST_INFO ("preparing test"); GST_INFO ("preparing test");
@ -357,7 +358,10 @@ GST_START_TEST (test_rtx_ssrc_collision)
rtppayloader = gst_element_factory_make ("rtpspeexpay", NULL); rtppayloader = gst_element_factory_make ("rtpspeexpay", NULL);
g_object_set (rtppayloader, "pt", 96, NULL); g_object_set (rtppayloader, "pt", 96, NULL);
rtprtxsend = gst_element_factory_make ("rtprtxsend", NULL); rtprtxsend = gst_element_factory_make ("rtprtxsend", NULL);
g_object_set (rtprtxsend, "rtx-payload-type", 99, NULL); pt_map = gst_structure_new ("application/x-rtp-pt-map",
"96", G_TYPE_UINT, 99, NULL);
g_object_set (rtprtxsend, "payload-type-map", pt_map, NULL);
gst_structure_free (pt_map);
rtpsession = gst_element_factory_make ("rtpsession", NULL); rtpsession = gst_element_factory_make ("rtpsession", NULL);
sink = gst_element_factory_make ("fakesink", "sink"); sink = gst_element_factory_make ("fakesink", "sink");
gst_bin_add_many (GST_BIN (bin), src, encoder, rtppayloader, rtprtxsend, gst_bin_add_many (GST_BIN (bin), src, encoder, rtppayloader, rtprtxsend,

42
tests/check/elements/rtprtx.c
View file

@ -200,6 +200,7 @@ GST_START_TEST (test_push_forward_seq)
GList *node; GList *node;
gint i = 0; gint i = 0;
GstCaps *caps = NULL; GstCaps *caps = NULL;
GstStructure *pt_map;
rtprtxsend = gst_check_setup_element ("rtprtxsend"); rtprtxsend = gst_check_setup_element ("rtprtxsend");
rtprtxreceive = gst_check_setup_element ("rtprtxreceive"); rtprtxreceive = gst_check_setup_element ("rtprtxreceive");
@ -216,8 +217,11 @@ GST_START_TEST (test_push_forward_seq)
gst_check_setup_events (srcpad, rtprtxsend, caps, GST_FORMAT_TIME); gst_check_setup_events (srcpad, rtprtxsend, caps, GST_FORMAT_TIME);
gst_caps_unref (caps); gst_caps_unref (caps);
g_object_set (rtprtxsend, "rtx-payload-type", 97, NULL); pt_map = gst_structure_new ("application/x-rtp-pt-map",
"0", G_TYPE_UINT, 97, NULL);
g_object_set (rtprtxsend, "payload-type-map", pt_map, NULL);
g_object_set (rtprtxreceive, "rtx-payload-types", "97", NULL); g_object_set (rtprtxreceive, "rtx-payload-types", "97", NULL);
gst_structure_free (pt_map);
/* push buffers: 0,1,2, */ /* push buffers: 0,1,2, */
for (node = inbuffers; node; node = g_list_next (node)) { for (node = inbuffers; node; node = g_list_next (node)) {
@ -395,13 +399,17 @@ start_test_drop_and_check_results (GstElement * bin, GstElement * rtppayloader,
guint nbrtxrequests = 0; guint nbrtxrequests = 0;
guint nbrtxpackets = 0; guint nbrtxpackets = 0;
guint nb_expected_requests = 0; guint nb_expected_requests = 0;
GstStructure *pt_map;
GST_INFO ("starting test"); GST_INFO ("starting test");
pt_map = gst_structure_new ("application/x-rtp-pt-map",
"96", G_TYPE_UINT, 99, NULL);
g_object_set (rtppayloader, "pt", 96, NULL); g_object_set (rtppayloader, "pt", 96, NULL);
g_object_set (rtppayloader, "seqnum-offset", 1, NULL); g_object_set (rtppayloader, "seqnum-offset", 1, NULL);
g_object_set (rtprtxsend, "rtx-payload-type", 99, NULL); g_object_set (rtprtxsend, "payload-type-map", pt_map, NULL);
g_object_set (rtprtxreceive, "rtx-payload-types", "99:111:125", NULL); g_object_set (rtprtxreceive, "rtx-payload-types", "99:111:125", NULL);
gst_structure_free (pt_map);
send_rtxdata->count = 1; send_rtxdata->count = 1;
send_rtxdata->nb_packets = 0; send_rtxdata->nb_packets = 0;
@ -686,6 +694,8 @@ add_sender (GstElement * bin, const gchar * src_name,
GstPad *srcpad = NULL; GstPad *srcpad = NULL;
gboolean res = FALSE; gboolean res = FALSE;
RTXSendMultipleData *send_rtxdata = g_slice_new0 (RTXSendMultipleData); RTXSendMultipleData *send_rtxdata = g_slice_new0 (RTXSendMultipleData);
gchar *pt_master;
GstStructure *pt_map;
send_rtxdata->count = 1; send_rtxdata->count = 1;
send_rtxdata->nb_packets = 0; send_rtxdata->nb_packets = 0;
@ -698,14 +708,21 @@ add_sender (GstElement * bin, const gchar * src_name,
queue = gst_element_factory_make ("queue", NULL); queue = gst_element_factory_make ("queue", NULL);
funnel = gst_bin_get_by_name (GST_BIN (bin), "funnel"); funnel = gst_bin_get_by_name (GST_BIN (bin), "funnel");
pt_master = g_strdup_printf ("%" G_GUINT32_FORMAT, payload_type_master);
pt_map = gst_structure_new ("application/x-rtp-pt-map",
pt_master, G_TYPE_UINT, payload_type_aux, NULL);
g_free (pt_master);
g_object_set (src, "num-buffers", 25, NULL); g_object_set (src, "num-buffers", 25, NULL);
g_object_set (rtppayloader, "pt", payload_type_master, NULL); g_object_set (rtppayloader, "pt", payload_type_master, NULL);
g_object_set (rtppayloader, "seqnum-offset", 1, NULL); g_object_set (rtppayloader, "seqnum-offset", 1, NULL);
g_object_set (rtprtxsend, "rtx-payload-type", payload_type_aux, NULL); g_object_set (rtprtxsend, "payload-type-map", pt_map, NULL);
/* we want that every drop packet be resent fast */ /* we want that every drop packet be resent fast */
g_object_set (queue, "max-size-buffers", 1, NULL); g_object_set (queue, "max-size-buffers", 1, NULL);
g_object_set (queue, "flush-on-eos", FALSE, NULL); g_object_set (queue, "flush-on-eos", FALSE, NULL);
gst_structure_free (pt_map);
gst_bin_add_many (GST_BIN (bin), src, rtppayloader, rtprtxsend, queue, NULL); gst_bin_add_many (GST_BIN (bin), src, rtppayloader, rtprtxsend, queue, NULL);
res = gst_element_link (src, rtppayloader); res = gst_element_link (src, rtppayloader);
@ -1167,6 +1184,7 @@ test_rtxsender_packet_retention (gboolean test_with_time)
const gint half_buffers = num_buffers / 2; const gint half_buffers = num_buffers / 2;
const guint ssrc = 1234567; const guint ssrc = 1234567;
const guint rtx_payload_type = 99; const guint rtx_payload_type = 99;
GstStructure *pt_map;
GList *in_buffers, *node; GList *in_buffers, *node;
guint payload_type; guint payload_type;
GstElement *rtxsend; GstElement *rtxsend;
@ -1185,6 +1203,9 @@ test_rtxsender_packet_retention (gboolean test_with_time)
/* setup element & pads */ /* setup element & pads */
rtxsend = gst_check_setup_element ("rtprtxsend"); rtxsend = gst_check_setup_element ("rtprtxsend");
pt_map = gst_structure_new ("application/x-rtp-pt-map",
"96", G_TYPE_UINT, rtx_payload_type, NULL);
/* in both cases we want the rtxsend queue to store 'half_buffers' /* in both cases we want the rtxsend queue to store 'half_buffers'
* amount of buffers at most. In max-size-packets mode, it's trivial. * amount of buffers at most. In max-size-packets mode, it's trivial.
* In max-size-time mode, we specify almost half a second, which is * In max-size-time mode, we specify almost half a second, which is
@ -1192,7 +1213,8 @@ test_rtxsender_packet_retention (gboolean test_with_time)
g_object_set (rtxsend, g_object_set (rtxsend,
"max-size-packets", test_with_time ? 0 : half_buffers, "max-size-packets", test_with_time ? 0 : half_buffers,
"max-size-time", test_with_time ? 499 : 0, "max-size-time", test_with_time ? 499 : 0,
"rtx-payload-type", rtx_payload_type, NULL); "payload-type-map", pt_map, NULL);
gst_structure_free (pt_map);
srcpad = gst_check_setup_src_pad (rtxsend, &srctemplate); srcpad = gst_check_setup_src_pad (rtxsend, &srctemplate);
fail_unless_equals_int (gst_pad_set_active (srcpad, TRUE), TRUE); fail_unless_equals_int (gst_pad_set_active (srcpad, TRUE), TRUE);
@ -1204,7 +1226,8 @@ test_rtxsender_packet_retention (gboolean test_with_time)
caps = gst_caps_from_string ("application/x-rtp, " caps = gst_caps_from_string ("application/x-rtp, "
"media = (string)video, payload = (int)96, " "media = (string)video, payload = (int)96, "
"clock-rate = (int)90000, encoding-name = (string)RAW"); "ssrc = (uint)1234567, clock-rate = (int)90000, "
"encoding-name = (string)RAW");
gst_check_setup_events (srcpad, rtxsend, caps, GST_FORMAT_TIME); gst_check_setup_events (srcpad, rtxsend, caps, GST_FORMAT_TIME);
gst_caps_unref (caps); gst_caps_unref (caps);
@ -1364,6 +1387,7 @@ GST_START_TEST (test_rtxreceive_data_reconstruction)
GstCaps *caps; GstCaps *caps;
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
GstBuffer *buffer; GstBuffer *buffer;
GstStructure *pt_map;
/* generate test data */ /* generate test data */
in_buffers = generate_test_buffers (1, ssrc, &payload_type); in_buffers = generate_test_buffers (1, ssrc, &payload_type);
@ -1375,8 +1399,11 @@ GST_START_TEST (test_rtxreceive_data_reconstruction)
rtxsend = gst_check_setup_element ("rtprtxsend"); rtxsend = gst_check_setup_element ("rtprtxsend");
rtxrecv = gst_check_setup_element ("rtprtxreceive"); rtxrecv = gst_check_setup_element ("rtprtxreceive");
g_object_set (rtxsend, "rtx-payload-type", 99, NULL); pt_map = gst_structure_new ("application/x-rtp-pt-map",
"96", G_TYPE_UINT, 99, NULL);
g_object_set (rtxsend, "payload-type-map", pt_map, NULL);
g_object_set (rtxrecv, "rtx-payload-types", "99", NULL); g_object_set (rtxrecv, "rtx-payload-types", "99", NULL);
gst_structure_free (pt_map);
fail_unless_equals_int (gst_element_link (rtxsend, rtxrecv), TRUE); fail_unless_equals_int (gst_element_link (rtxsend, rtxrecv), TRUE);
@ -1391,7 +1418,8 @@ GST_START_TEST (test_rtxreceive_data_reconstruction)
caps = gst_caps_from_string ("application/x-rtp, " caps = gst_caps_from_string ("application/x-rtp, "
"media = (string)video, payload = (int)96, " "media = (string)video, payload = (int)96, "
"clock-rate = (int)90000, encoding-name = (string)RAW"); "ssrc = (uint)1234567, clock-rate = (int)90000, "
"encoding-name = (string)RAW");
gst_check_setup_events (srcpad, rtxsend, caps, GST_FORMAT_TIME); gst_check_setup_events (srcpad, rtxsend, caps, GST_FORMAT_TIME);
gst_caps_unref (caps); gst_caps_unref (caps);

6
tests/examples/rtp/server-rtpaux.c
View file

@ -167,11 +167,15 @@ request_aux_sender (GstElement * rtpbin, guint sessid, SessionData * session)
GstElement *rtx, *bin; GstElement *rtx, *bin;
GstPad *pad; GstPad *pad;
gchar *name; gchar *name;
GstStructure *pt_map;
GST_INFO ("creating AUX sender"); GST_INFO ("creating AUX sender");
bin = gst_bin_new (NULL); bin = gst_bin_new (NULL);
rtx = gst_element_factory_make ("rtprtxsend", NULL); rtx = gst_element_factory_make ("rtprtxsend", NULL);
g_object_set (rtx, "rtx-payload-type", 99, NULL); pt_map = gst_structure_new ("application/x-rtp-pt-map",
"96", G_TYPE_UINT, 99, NULL);
g_object_set (rtx, "payload-type-map", pt_map, NULL);
gst_structure_free (pt_map);
gst_bin_add (GST_BIN (bin), rtx); gst_bin_add (GST_BIN (bin), rtx);
pad = gst_element_get_static_pad (rtx, "src"); pad = gst_element_get_static_pad (rtx, "src");