gstreamer/gst/rtpmanager/gstrtprtxreceive.c
Nicolas Dufresne bf5cbce3b4 rtprtxreceive: Add memory and boudary checks
This element was not checking if mapping the RTP buffer and the payload
worked, and was not checking if the RTX payload was large enough.

https://bugzilla.gnome.org/show_bug.cgi?id=784484
2017-07-04 09:58:15 -04:00

786 lines
28 KiB
C

/* RTP Retransmission receiver element for GStreamer
*
* gstrtprtxreceive.c:
*
* Copyright (C) 2013 Collabora Ltd.
* @author Julien Isorce <julien.isorce@collabora.co.uk>
*
* 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., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
/**
* SECTION:element-rtprtxreceive
* @see_also: rtprtxsend, rtpsession, rtpjitterbuffer
*
* rtprtxreceive listens to the retransmission events from the
* downstream rtpjitterbuffer and remembers the SSRC (ssrc1) of the stream and
* the sequence number that was requested. When it receives a packet with
* a sequence number equal to one of the ones stored and with a different SSRC,
* it identifies the new SSRC (ssrc2) as the retransmission stream of ssrc1.
* From this point on, it replaces ssrc2 with ssrc1 in all packets of the
* ssrc2 stream and flags them as retransmissions, so that rtpjitterbuffer
* can reconstruct the original stream.
*
* This algorithm is implemented as specified in RFC 4588.
*
* This element is meant to be used with rtprtxsend on the sender side.
* See #GstRtpRtxSend
*
* Below you can see some examples that illustrate how rtprtxreceive and
* rtprtxsend fit among the other rtp elements and how they work internally.
* Normally, hoewever, you should avoid using such pipelines and use
* rtpbin instead, with its #GstRtpBin::request-aux-sender and
* #GstRtpBin::request-aux-receiver signals. See #GstRtpBin.
*
* # Example pipelines
* |[
* gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
* audiotestsrc is-live=true ! opusenc ! rtpopuspay pt=96 ! \
* rtprtxsend payload-type-map="application/x-rtp-pt-map,96=(uint)97" ! \
* rtpsession.send_rtp_sink \
* rtpsession.send_rtp_src ! identity drop-probability=0.01 ! \
* udpsink host="127.0.0.1" port=5000 \
* udpsrc port=5001 ! rtpsession.recv_rtcp_sink \
* rtpsession.send_rtcp_src ! udpsink host="127.0.0.1" port=5002 \
* sync=false async=false
* ]| Send audio stream through port 5000 (5001 and 5002 are just the rtcp
* link with the receiver)
* |[
* gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
* udpsrc port=5000 caps="application/x-rtp,media=(string)audio,clock-rate=(int)48000,encoding-name=(string)OPUS,payload=(int)96" ! \
* rtpsession.recv_rtp_sink \
* rtpsession.recv_rtp_src ! \
* rtprtxreceive payload-type-map="application/x-rtp-pt-map,96=(uint)97" ! \
* rtpssrcdemux ! rtpjitterbuffer do-retransmission=true ! \
* rtpopusdepay ! opusdec ! audioconvert ! audioresample ! autoaudiosink \
* rtpsession.send_rtcp_src ! \
* udpsink host="127.0.0.1" port=5001 sync=false async=false \
* udpsrc port=5002 ! rtpsession.recv_rtcp_sink
* ]| Receive audio stream from port 5000 (5001 and 5002 are just the rtcp
* link with the sender)
*
* In this example we can see a simple streaming of an OPUS stream with some
* of the packets being artificially dropped by the identity element.
* Thanks to retransmission, you should still hear a clear sound when setting
* drop-probability to something greater than 0.
*
* Internally, the rtpjitterbuffer will generate a custom upstream event,
* GstRTPRetransmissionRequest, when it detects that one packet is missing.
* Then this request is translated to a FB NACK in the rtcp link by rtpsession.
* Finally the rtpsession of the sender side will re-convert it in a
* GstRTPRetransmissionRequest that will be handled by rtprtxsend. rtprtxsend
* will then re-send the missing packet with a new srrc and a different payload
* type (here, 97), but with the same original sequence number. On the receiver
* side, rtprtxreceive will associate this new stream with the original and
* forward the retransmission packets to rtpjitterbuffer with the original
* ssrc and payload type.
*
* |[
* gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
* audiotestsrc is-live=true ! opusenc ! rtpopuspay pt=97 seqnum-offset=1 ! \
* rtprtxsend payload-type-map="application/x-rtp-pt-map,97=(uint)99" ! \
* funnel name=f ! rtpsession.send_rtp_sink \
* audiotestsrc freq=660.0 is-live=true ! opusenc ! \
* rtpopuspay pt=97 seqnum-offset=100 ! \
* rtprtxsend payload-type-map="application/x-rtp-pt-map,97=(uint)99" ! \
* f. \
* rtpsession.send_rtp_src ! identity drop-probability=0.01 ! \
* udpsink host="127.0.0.1" port=5000 \
* udpsrc port=5001 ! rtpsession.recv_rtcp_sink \
* rtpsession.send_rtcp_src ! udpsink host="127.0.0.1" port=5002 \
* sync=false async=false
* ]| Send two audio streams to port 5000.
* |[
* gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
* udpsrc port=5000 caps="application/x-rtp,media=(string)audio,clock-rate=(int)48000,encoding-name=(string)OPUS,payload=(int)97" ! \
* rtpsession.recv_rtp_sink \
* rtpsession.recv_rtp_src ! \
* rtprtxreceive payload-type-map="application/x-rtp-pt-map,97=(uint)99" ! \
* rtpssrcdemux name=demux \
* demux. ! queue ! rtpjitterbuffer do-retransmission=true ! rtpopusdepay ! \
* opusdec ! audioconvert ! autoaudiosink \
* demux. ! queue ! rtpjitterbuffer do-retransmission=true ! rtpopusdepay ! \
* opusdec ! audioconvert ! autoaudiosink \
* udpsrc port=5002 ! rtpsession.recv_rtcp_sink \
* rtpsession.send_rtcp_src ! udpsink host="127.0.0.1" port=5001 \
* sync=false async=false
* ]| Receive two audio streams from port 5000.
*
* In this example we are streaming two streams of the same type through the
* same port. They, however, are using a different SSRC (ssrc is randomly
* generated on each payloader - rtpopuspay in this example), so they can be
* identified and demultiplexed by rtpssrcdemux on the receiver side. This is
* an example of SSRC-multiplexing.
*
* It is important here to use a different starting sequence number
* (seqnum-offset), since this is the only means of identification that
* rtprtxreceive uses the very first time to identify retransmission streams.
* It is an error, according to RFC4588 to have two retransmission requests for
* packets belonging to two different streams but with the same sequence number.
* Note that the default seqnum-offset value (-1, which means random) would
* work just fine, but it is overriden here for illustration purposes.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <gst/gst.h>
#include <gst/rtp/gstrtpbuffer.h>
#include <string.h>
#include <stdlib.h>
#include "gstrtprtxreceive.h"
#define ASSOC_TIMEOUT (GST_SECOND)
GST_DEBUG_CATEGORY_STATIC (gst_rtp_rtx_receive_debug);
#define GST_CAT_DEFAULT gst_rtp_rtx_receive_debug
enum
{
PROP_0,
PROP_PAYLOAD_TYPE_MAP,
PROP_NUM_RTX_REQUESTS,
PROP_NUM_RTX_PACKETS,
PROP_NUM_RTX_ASSOC_PACKETS
};
static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
GST_PAD_SRC,
GST_PAD_ALWAYS,
GST_STATIC_CAPS ("application/x-rtp")
);
static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink",
GST_PAD_SINK,
GST_PAD_ALWAYS,
GST_STATIC_CAPS ("application/x-rtp")
);
static gboolean gst_rtp_rtx_receive_src_event (GstPad * pad, GstObject * parent,
GstEvent * event);
static GstFlowReturn gst_rtp_rtx_receive_chain (GstPad * pad,
GstObject * parent, GstBuffer * buffer);
static GstStateChangeReturn gst_rtp_rtx_receive_change_state (GstElement *
element, GstStateChange transition);
static void gst_rtp_rtx_receive_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_rtp_rtx_receive_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
static void gst_rtp_rtx_receive_finalize (GObject * object);
G_DEFINE_TYPE (GstRtpRtxReceive, gst_rtp_rtx_receive, GST_TYPE_ELEMENT);
static void
gst_rtp_rtx_receive_class_init (GstRtpRtxReceiveClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *gstelement_class;
gobject_class = (GObjectClass *) klass;
gstelement_class = (GstElementClass *) klass;
gobject_class->get_property = gst_rtp_rtx_receive_get_property;
gobject_class->set_property = gst_rtp_rtx_receive_set_property;
gobject_class->finalize = gst_rtp_rtx_receive_finalize;
g_object_class_install_property (gobject_class, PROP_PAYLOAD_TYPE_MAP,
g_param_spec_boxed ("payload-type-map", "Payload Type Map",
"Map of original payload types to their retransmission payload types",
GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_NUM_RTX_REQUESTS,
g_param_spec_uint ("num-rtx-requests", "Num RTX Requests",
"Number of retransmission events received", 0, G_MAXUINT,
0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_NUM_RTX_PACKETS,
g_param_spec_uint ("num-rtx-packets", "Num RTX Packets",
" Number of retransmission packets received", 0, G_MAXUINT,
0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_NUM_RTX_ASSOC_PACKETS,
g_param_spec_uint ("num-rtx-assoc-packets",
"Num RTX Associated Packets", "Number of retransmission packets "
"correctly associated with retransmission requests", 0, G_MAXUINT,
0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
gst_element_class_add_static_pad_template (gstelement_class, &src_factory);
gst_element_class_add_static_pad_template (gstelement_class, &sink_factory);
gst_element_class_set_static_metadata (gstelement_class,
"RTP Retransmission receiver", "Codec",
"Receive retransmitted RTP packets according to RFC4588",
"Julien Isorce <julien.isorce@collabora.co.uk>");
gstelement_class->change_state =
GST_DEBUG_FUNCPTR (gst_rtp_rtx_receive_change_state);
}
static void
gst_rtp_rtx_receive_reset (GstRtpRtxReceive * rtx)
{
GST_OBJECT_LOCK (rtx);
g_hash_table_remove_all (rtx->ssrc2_ssrc1_map);
g_hash_table_remove_all (rtx->seqnum_ssrc1_map);
rtx->num_rtx_requests = 0;
rtx->num_rtx_packets = 0;
rtx->num_rtx_assoc_packets = 0;
GST_OBJECT_UNLOCK (rtx);
}
static void
gst_rtp_rtx_receive_finalize (GObject * object)
{
GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (object);
g_hash_table_unref (rtx->ssrc2_ssrc1_map);
g_hash_table_unref (rtx->seqnum_ssrc1_map);
g_hash_table_unref (rtx->rtx_pt_map);
if (rtx->rtx_pt_map_structure)
gst_structure_free (rtx->rtx_pt_map_structure);
G_OBJECT_CLASS (gst_rtp_rtx_receive_parent_class)->finalize (object);
}
typedef struct
{
guint32 ssrc;
GstClockTime time;
} SsrcAssoc;
static SsrcAssoc *
ssrc_assoc_new (guint32 ssrc, GstClockTime time)
{
SsrcAssoc *assoc = g_slice_new (SsrcAssoc);
assoc->ssrc = ssrc;
assoc->time = time;
return assoc;
}
static void
ssrc_assoc_free (SsrcAssoc * assoc)
{
g_slice_free (SsrcAssoc, assoc);
}
static void
gst_rtp_rtx_receive_init (GstRtpRtxReceive * rtx)
{
GstElementClass *klass = GST_ELEMENT_GET_CLASS (rtx);
rtx->srcpad =
gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
"src"), "src");
GST_PAD_SET_PROXY_CAPS (rtx->srcpad);
GST_PAD_SET_PROXY_ALLOCATION (rtx->srcpad);
gst_pad_set_event_function (rtx->srcpad,
GST_DEBUG_FUNCPTR (gst_rtp_rtx_receive_src_event));
gst_element_add_pad (GST_ELEMENT (rtx), rtx->srcpad);
rtx->sinkpad =
gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
"sink"), "sink");
GST_PAD_SET_PROXY_CAPS (rtx->sinkpad);
GST_PAD_SET_PROXY_ALLOCATION (rtx->sinkpad);
gst_pad_set_chain_function (rtx->sinkpad,
GST_DEBUG_FUNCPTR (gst_rtp_rtx_receive_chain));
gst_element_add_pad (GST_ELEMENT (rtx), rtx->sinkpad);
rtx->ssrc2_ssrc1_map = g_hash_table_new (g_direct_hash, g_direct_equal);
rtx->seqnum_ssrc1_map = g_hash_table_new_full (g_direct_hash, g_direct_equal,
NULL, (GDestroyNotify) ssrc_assoc_free);
rtx->rtx_pt_map = g_hash_table_new (g_direct_hash, g_direct_equal);
}
static gboolean
gst_rtp_rtx_receive_src_event (GstPad * pad, GstObject * parent,
GstEvent * event)
{
GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (parent);
gboolean res;
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_CUSTOM_UPSTREAM:
{
const GstStructure *s = gst_event_get_structure (event);
/* This event usually comes from the downstream gstrtpjitterbuffer */
if (gst_structure_has_name (s, "GstRTPRetransmissionRequest")) {
guint seqnum = 0;
guint ssrc = 0;
gpointer ssrc2 = 0;
/* retrieve seqnum of the packet that need to be retransmitted */
if (!gst_structure_get_uint (s, "seqnum", &seqnum))
seqnum = -1;
/* retrieve ssrc of the packet that need to be retransmitted
* it's useful when reconstructing the original packet from the rtx packet */
if (!gst_structure_get_uint (s, "ssrc", &ssrc))
ssrc = -1;
GST_DEBUG_OBJECT (rtx,
"request seqnum: %" G_GUINT32_FORMAT ", ssrc: %" G_GUINT32_FORMAT,
seqnum, ssrc);
GST_OBJECT_LOCK (rtx);
/* increase number of seen requests for our statistics */
++rtx->num_rtx_requests;
/* First, we lookup in our map to see if we have already associate this
* master stream ssrc with its retransmitted stream.
* Every ssrc are unique so we can use the same hash table
* for both retrieving the ssrc1 from ssrc2 and also ssrc2 from ssrc1
*/
if (g_hash_table_lookup_extended (rtx->ssrc2_ssrc1_map,
GUINT_TO_POINTER (ssrc), NULL, &ssrc2)
&& GPOINTER_TO_UINT (ssrc2) != GPOINTER_TO_UINT (ssrc)) {
GST_DEBUG_OBJECT (rtx, "Retransmited stream %" G_GUINT32_FORMAT
" already associated to its master", GPOINTER_TO_UINT (ssrc2));
} else {
SsrcAssoc *assoc;
/* not already associated but also we have to check that we have not
* already considered this request.
*/
if (g_hash_table_lookup_extended (rtx->seqnum_ssrc1_map,
GUINT_TO_POINTER (seqnum), NULL, (gpointer *) & assoc)) {
if (assoc->ssrc == ssrc) {
/* do nothing because we have already considered this request
* The jitter may be too impatient of the rtx packet has been
* lost too.
* It does not mean we reject the event, we still want to forward
* the request to the gstrtpsession to be translater into a FB NACK
*/
GST_DEBUG_OBJECT (rtx, "Duplicated request seqnum: %"
G_GUINT32_FORMAT ", ssrc1: %" G_GUINT32_FORMAT, seqnum, ssrc);
} else {
/* If the association attempt is larger than ASSOC_TIMEOUT,
* then we give up on it, and try this one.
*/
if (!GST_CLOCK_TIME_IS_VALID (rtx->last_time) ||
!GST_CLOCK_TIME_IS_VALID (assoc->time) ||
assoc->time + ASSOC_TIMEOUT < rtx->last_time) {
/* From RFC 4588:
* the receiver MUST NOT have two outstanding requests for the
* same packet sequence number in two different original streams
* before the association is resolved. Otherwise it's impossible
* to associate a rtx stream and its master stream
*/
/* remove seqnum in order to reuse the spot */
g_hash_table_remove (rtx->seqnum_ssrc1_map,
GUINT_TO_POINTER (seqnum));
goto retransmit;
} else {
GST_DEBUG_OBJECT (rtx,
"reject request for seqnum %" G_GUINT32_FORMAT
" of master stream %" G_GUINT32_FORMAT, seqnum, ssrc);
/* do not forward the event as we are rejecting this request */
GST_OBJECT_UNLOCK (rtx);
gst_event_unref (event);
return TRUE;
}
}
} else {
retransmit:
/* the request has not been already considered
* insert it for the first time */
g_hash_table_insert (rtx->seqnum_ssrc1_map,
GUINT_TO_POINTER (seqnum),
ssrc_assoc_new (ssrc, rtx->last_time));
}
}
GST_DEBUG_OBJECT (rtx,
"packet number %" G_GUINT32_FORMAT " of master stream %"
G_GUINT32_FORMAT " needs to be retransmitted", seqnum, ssrc);
GST_OBJECT_UNLOCK (rtx);
}
/* Transfer event upstream so that the request can acutally by translated
* through gstrtpsession through the network */
res = gst_pad_event_default (pad, parent, event);
break;
}
default:
res = gst_pad_event_default (pad, parent, event);
break;
}
return res;
}
/* Copy fixed header and extension. Replace current ssrc by ssrc1,
* remove OSN and replace current seq num by OSN.
* Copy memory to avoid to manually copy each rtp buffer field.
*/
static GstBuffer *
_gst_rtp_buffer_new_from_rtx (GstRTPBuffer * rtp, guint32 ssrc1,
guint16 orign_seqnum, guint8 origin_payload_type)
{
GstMemory *mem = NULL;
GstRTPBuffer new_rtp = GST_RTP_BUFFER_INIT;
GstBuffer *new_buffer = gst_buffer_new ();
GstMapInfo map;
guint payload_len = 0;
/* copy fixed header */
mem = gst_memory_copy (rtp->map[0].memory,
(guint8 *) rtp->data[0] - rtp->map[0].data, rtp->size[0]);
gst_buffer_append_memory (new_buffer, mem);
/* copy extension if any */
if (rtp->size[1]) {
mem = gst_memory_copy (rtp->map[1].memory,
(guint8 *) rtp->data[1] - rtp->map[1].data, rtp->size[1]);
gst_buffer_append_memory (new_buffer, mem);
}
/* copy payload and remove OSN */
payload_len = rtp->size[2] - 2;
mem = gst_allocator_alloc (NULL, payload_len, NULL);
gst_memory_map (mem, &map, GST_MAP_WRITE);
if (rtp->size[2])
memcpy (map.data, (guint8 *) rtp->data[2] + 2, payload_len);
gst_memory_unmap (mem, &map);
gst_buffer_append_memory (new_buffer, mem);
/* the sender always constructs rtx packets without padding,
* But the receiver can still receive rtx packets with padding.
* So just copy it.
*/
if (rtp->size[3]) {
guint pad_len = rtp->size[3];
mem = gst_allocator_alloc (NULL, pad_len, NULL);
gst_memory_map (mem, &map, GST_MAP_WRITE);
map.data[pad_len - 1] = pad_len;
gst_memory_unmap (mem, &map);
gst_buffer_append_memory (new_buffer, mem);
}
/* set ssrc and seq num */
gst_rtp_buffer_map (new_buffer, GST_MAP_WRITE, &new_rtp);
gst_rtp_buffer_set_ssrc (&new_rtp, ssrc1);
gst_rtp_buffer_set_seq (&new_rtp, orign_seqnum);
gst_rtp_buffer_set_payload_type (&new_rtp, origin_payload_type);
gst_rtp_buffer_unmap (&new_rtp);
gst_buffer_copy_into (new_buffer, rtp->buffer,
GST_BUFFER_COPY_FLAGS | GST_BUFFER_COPY_TIMESTAMPS, 0, -1);
GST_BUFFER_FLAG_SET (new_buffer, GST_RTP_BUFFER_FLAG_RETRANSMISSION);
return new_buffer;
}
static GstFlowReturn
gst_rtp_rtx_receive_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{
GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (parent);
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
GstFlowReturn ret = GST_FLOW_OK;
GstBuffer *new_buffer = NULL;
guint32 ssrc = 0;
gpointer ssrc1 = 0;
guint32 ssrc2 = 0;
guint16 seqnum = 0;
guint16 orign_seqnum = 0;
guint8 payload_type = 0;
gpointer payload = NULL;
guint8 origin_payload_type = 0;
gboolean is_rtx;
gboolean drop = FALSE;
/* map current rtp packet to parse its header */
if (!gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp))
goto invalid_buffer;
ssrc = gst_rtp_buffer_get_ssrc (&rtp);
seqnum = gst_rtp_buffer_get_seq (&rtp);
payload_type = gst_rtp_buffer_get_payload_type (&rtp);
/* check if we have a retransmission packet (this information comes from SDP) */
GST_OBJECT_LOCK (rtx);
is_rtx =
g_hash_table_lookup_extended (rtx->rtx_pt_map,
GUINT_TO_POINTER (payload_type), NULL, NULL);
if (is_rtx) {
payload = gst_rtp_buffer_get_payload (&rtp);
if (!payload || gst_rtp_buffer_get_payload_len (&rtp) < 2) {
GST_OBJECT_UNLOCK (rtx);
gst_rtp_buffer_unmap (&rtp);
goto invalid_buffer;
}
}
rtx->last_time = GST_BUFFER_PTS (buffer);
if (g_hash_table_size (rtx->seqnum_ssrc1_map) > 0) {
GHashTableIter iter;
gpointer key, value;
g_hash_table_iter_init (&iter, rtx->seqnum_ssrc1_map);
while (g_hash_table_iter_next (&iter, &key, &value)) {
SsrcAssoc *assoc = value;
/* remove association request if it is too old */
if (GST_CLOCK_TIME_IS_VALID (rtx->last_time) &&
GST_CLOCK_TIME_IS_VALID (assoc->time) &&
assoc->time + ASSOC_TIMEOUT < rtx->last_time) {
g_hash_table_iter_remove (&iter);
}
}
}
/* if the current packet is from a retransmission stream */
if (is_rtx) {
/* increase our statistic */
++rtx->num_rtx_packets;
/* read OSN in the rtx payload */
orign_seqnum = GST_READ_UINT16_BE (gst_rtp_buffer_get_payload (&rtp));
origin_payload_type =
GPOINTER_TO_UINT (g_hash_table_lookup (rtx->rtx_pt_map,
GUINT_TO_POINTER (payload_type)));
/* first we check if we already have associated this retransmission stream
* to a master stream */
if (g_hash_table_lookup_extended (rtx->ssrc2_ssrc1_map,
GUINT_TO_POINTER (ssrc), NULL, &ssrc1)) {
GST_DEBUG_OBJECT (rtx,
"packet is from retransmission stream %" G_GUINT32_FORMAT
" already associated to master stream %" G_GUINT32_FORMAT, ssrc,
GPOINTER_TO_UINT (ssrc1));
ssrc2 = ssrc;
} else {
SsrcAssoc *assoc;
/* the current retransmitted packet has its rtx stream not already
* associated to a master stream, so retrieve it from our request
* history */
if (g_hash_table_lookup_extended (rtx->seqnum_ssrc1_map,
GUINT_TO_POINTER (orign_seqnum), NULL, (gpointer *) & assoc)) {
GST_DEBUG_OBJECT (rtx,
"associate retransmitted stream %" G_GUINT32_FORMAT
" to master stream %" G_GUINT32_FORMAT " thanks to packet %"
G_GUINT16_FORMAT "", ssrc, assoc->ssrc, orign_seqnum);
ssrc1 = GUINT_TO_POINTER (assoc->ssrc);
ssrc2 = ssrc;
/* just put a guard */
if (GPOINTER_TO_UINT (ssrc1) == ssrc2)
GST_WARNING_OBJECT (rtx, "RTX receiver ssrc2_ssrc1_map bad state, "
"ssrc %" G_GUINT32_FORMAT " are the same\n", ssrc);
/* free the spot so that this seqnum can be used to do another
* association */
g_hash_table_remove (rtx->seqnum_ssrc1_map,
GUINT_TO_POINTER (orign_seqnum));
/* actually do the association between rtx stream and master stream */
g_hash_table_insert (rtx->ssrc2_ssrc1_map, GUINT_TO_POINTER (ssrc2),
ssrc1);
/* also do the association between master stream and rtx stream
* every ssrc are unique so we can use the same hash table
* for both retrieving the ssrc1 from ssrc2 and also ssrc2 from ssrc1
*/
g_hash_table_insert (rtx->ssrc2_ssrc1_map, ssrc1,
GUINT_TO_POINTER (ssrc2));
} else {
/* we are not able to associate this rtx packet with a master stream */
GST_DEBUG_OBJECT (rtx,
"drop rtx packet because its orign_seqnum %" G_GUINT16_FORMAT
" is not in pending retransmission requests", orign_seqnum);
drop = TRUE;
}
}
}
/* if not dropped the packet was successfully associated */
if (is_rtx && !drop)
++rtx->num_rtx_assoc_packets;
GST_OBJECT_UNLOCK (rtx);
/* just drop the packet if the association could not have been made */
if (drop) {
gst_rtp_buffer_unmap (&rtp);
gst_buffer_unref (buffer);
return GST_FLOW_OK;
}
/* create the retransmission packet */
if (is_rtx)
new_buffer =
_gst_rtp_buffer_new_from_rtx (&rtp, GPOINTER_TO_UINT (ssrc1),
orign_seqnum, origin_payload_type);
gst_rtp_buffer_unmap (&rtp);
/* push the packet */
if (is_rtx) {
gst_buffer_unref (buffer);
GST_LOG_OBJECT (rtx, "push packet seqnum:%" G_GUINT16_FORMAT
" from a restransmission stream ssrc2:%" G_GUINT32_FORMAT " (src %"
G_GUINT32_FORMAT ")", orign_seqnum, ssrc2, GPOINTER_TO_UINT (ssrc1));
ret = gst_pad_push (rtx->srcpad, new_buffer);
} else {
GST_LOG_OBJECT (rtx, "push packet seqnum:%" G_GUINT16_FORMAT
" from a master stream ssrc: %" G_GUINT32_FORMAT, seqnum, ssrc);
ret = gst_pad_push (rtx->srcpad, buffer);
}
return ret;
invalid_buffer:
{
GST_ELEMENT_WARNING (rtx, STREAM, DECODE, (NULL),
("Received invalid RTP payload, dropping"));
gst_buffer_unref (buffer);
return GST_FLOW_OK;
}
}
static void
gst_rtp_rtx_receive_get_property (GObject * object,
guint prop_id, GValue * value, GParamSpec * pspec)
{
GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (object);
switch (prop_id) {
case PROP_PAYLOAD_TYPE_MAP:
GST_OBJECT_LOCK (rtx);
g_value_set_boxed (value, rtx->rtx_pt_map_structure);
GST_OBJECT_UNLOCK (rtx);
break;
case PROP_NUM_RTX_REQUESTS:
GST_OBJECT_LOCK (rtx);
g_value_set_uint (value, rtx->num_rtx_requests);
GST_OBJECT_UNLOCK (rtx);
break;
case PROP_NUM_RTX_PACKETS:
GST_OBJECT_LOCK (rtx);
g_value_set_uint (value, rtx->num_rtx_packets);
GST_OBJECT_UNLOCK (rtx);
break;
case PROP_NUM_RTX_ASSOC_PACKETS:
GST_OBJECT_LOCK (rtx);
g_value_set_uint (value, rtx->num_rtx_assoc_packets);
GST_OBJECT_UNLOCK (rtx);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static gboolean
structure_to_hash_table_inv (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 (value_uint),
GUINT_TO_POINTER (field_uint));
return TRUE;
}
static void
gst_rtp_rtx_receive_set_property (GObject * object,
guint prop_id, const GValue * value, GParamSpec * pspec)
{
GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (object);
switch (prop_id) {
case PROP_PAYLOAD_TYPE_MAP:
GST_OBJECT_LOCK (rtx);
if (rtx->rtx_pt_map_structure)
gst_structure_free (rtx->rtx_pt_map_structure);
rtx->rtx_pt_map_structure = g_value_dup_boxed (value);
g_hash_table_remove_all (rtx->rtx_pt_map);
gst_structure_foreach (rtx->rtx_pt_map_structure,
structure_to_hash_table_inv, rtx->rtx_pt_map);
GST_OBJECT_UNLOCK (rtx);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static GstStateChangeReturn
gst_rtp_rtx_receive_change_state (GstElement * element,
GstStateChange transition)
{
GstStateChangeReturn ret;
GstRtpRtxReceive *rtx;
rtx = GST_RTP_RTX_RECEIVE (element);
switch (transition) {
default:
break;
}
ret =
GST_ELEMENT_CLASS (gst_rtp_rtx_receive_parent_class)->change_state
(element, transition);
switch (transition) {
case GST_STATE_CHANGE_PAUSED_TO_READY:
gst_rtp_rtx_receive_reset (rtx);
break;
default:
break;
}
return ret;
}
gboolean
gst_rtp_rtx_receive_plugin_init (GstPlugin * plugin)
{
GST_DEBUG_CATEGORY_INIT (gst_rtp_rtx_receive_debug, "rtprtxreceive", 0,
"rtp retransmission receiver");
return gst_element_register (plugin, "rtprtxreceive", GST_RANK_NONE,
GST_TYPE_RTP_RTX_RECEIVE);
}