mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-12-21 15:56:42 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/subprojects/gst-plugins-good/gst/rtpmanager/gstrtprtxreceive.c
Matthew Waters 206021e4d4 rtpmanager/rtx: implement initial support for reading/writing rid extensions 
Two RTP Header extensions are very relevant for rtprtxsend/receive.
1. "urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id": will always be removed
2. "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id": will be written
    instead of the "rtp-stream-id" header extension.

Currently it's only a simple replacement of one header extension for
another however a future change would only add the relevant extension
based on some heuristics (like, video frames only on one of the rtp key
frame buffers, or only until the rtx ssrc has been validated by the peer)
in order to reduce the required bandwidth.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/1759>
2022-03-21 03:18:18 +00:00

1081 lines
38 KiB
C
Raw Blame History

/* 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
* @title: 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 overridden here for illustration purposes.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <gst/gst.h>
#include <gst/rtp/rtp.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_SSRC_MAP,
PROP_PAYLOAD_TYPE_MAP,
PROP_NUM_RTX_REQUESTS,
PROP_NUM_RTX_PACKETS,
PROP_NUM_RTX_ASSOC_PACKETS
};
enum
{
SIGNAL_0,
SIGNAL_ADD_EXTENSION,
SIGNAL_CLEAR_EXTENSIONS,
LAST_SIGNAL
};
static guint gst_rtp_rtx_receive_signals[LAST_SIGNAL] = { 0, };
#define RTPHDREXT_STREAM_ID GST_RTP_HDREXT_BASE "sdes:rtp-stream-id"
#define RTPHDREXT_REPAIRED_STREAM_ID GST_RTP_HDREXT_BASE "sdes:repaired-rtp-stream-id"
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_WITH_CODE (GstRtpRtxReceive, gst_rtp_rtx_receive,
GST_TYPE_ELEMENT, GST_DEBUG_CATEGORY_INIT (gst_rtp_rtx_receive_debug,
"rtprtxreceive", 0, "rtp retransmission receiver"));
GST_ELEMENT_REGISTER_DEFINE (rtprtxreceive, "rtprtxreceive", GST_RANK_NONE,
GST_TYPE_RTP_RTX_RECEIVE);
static void
gst_rtp_rtx_receive_add_extension (GstRtpRtxReceive * rtx,
GstRTPHeaderExtension * ext)
{
g_return_if_fail (GST_IS_RTP_HEADER_EXTENSION (ext));
g_return_if_fail (gst_rtp_header_extension_get_id (ext) > 0);
GST_OBJECT_LOCK (rtx);
if (g_strcmp0 (gst_rtp_header_extension_get_uri (ext),
RTPHDREXT_STREAM_ID) == 0) {
gst_clear_object (&rtx->rid_stream);
rtx->rid_stream = gst_object_ref (ext);
} else if (g_strcmp0 (gst_rtp_header_extension_get_uri (ext),
RTPHDREXT_REPAIRED_STREAM_ID) == 0) {
gst_clear_object (&rtx->rid_repaired);
rtx->rid_repaired = gst_object_ref (ext);
} else {
g_warning ("rtprtxsend (%s) doesn't know how to deal with the "
"RTP Header Extension with URI \'%s\'", GST_OBJECT_NAME (rtx),
gst_rtp_header_extension_get_uri (ext));
}
/* XXX: check for other duplicate ids? */
GST_OBJECT_UNLOCK (rtx);
}
static void
gst_rtp_rtx_receive_clear_extensions (GstRtpRtxReceive * rtx)
{
GST_OBJECT_LOCK (rtx);
gst_clear_object (&rtx->rid_stream);
gst_clear_object (&rtx->rid_repaired);
GST_OBJECT_UNLOCK (rtx);
}
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;
/**
* GstRtpRtxReceive:ssrc-map:
*
* Map of SSRCs to their retransmission SSRCs for SSRC-multiplexed mode.
*
* If an application know this information already (WebRTC signals this
* in their SDP), it can allow the rtxreceive element to know a packet
* is a "valid" RTX packet even if it has not been requested.
*
* Since: 1.22
*/
g_object_class_install_property (gobject_class, PROP_SSRC_MAP,
g_param_spec_boxed ("ssrc-map", "SSRC Map",
"Map of SSRCs to their retransmission SSRCs for SSRC-multiplexed mode",
GST_TYPE_STRUCTURE, G_PARAM_WRITABLE | G_PARAM_STATIC_STRINGS));
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));
/**
* rtprtxreceive::add-extension:
*
* Add @ext as an extension for writing part of an RTP header extension onto
* outgoing RTP packets. Currently only supports using the following
* extension URIs. All other RTP header extensions are copied as-is.
* - "urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id": will be removed
* - "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id": will be
* written instead of the "rtp-stream-id" header extension.
*
* Since: 1.22
*/
gst_rtp_rtx_receive_signals[SIGNAL_ADD_EXTENSION] =
g_signal_new_class_handler ("add-extension", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_CALLBACK (gst_rtp_rtx_receive_add_extension), NULL, NULL, NULL,
G_TYPE_NONE, 1, GST_TYPE_RTP_HEADER_EXTENSION);
/**
* rtprtxreceive::clear-extensions:
* @object: the #GstRTPBasePayload
*
* Clear all RTP header extensions used by rtprtxreceive.
*
* Since: 1.22
*/
gst_rtp_rtx_receive_signals[SIGNAL_CLEAR_EXTENSIONS] =
g_signal_new_class_handler ("clear-extensions", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_CALLBACK (gst_rtp_rtx_receive_clear_extensions), NULL, NULL, NULL,
G_TYPE_NONE, 0);
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_CAST (object);
g_hash_table_unref (rtx->ssrc2_ssrc1_map);
if (rtx->external_ssrc_map)
gst_structure_free (rtx->external_ssrc_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);
gst_clear_object (&rtx->rid_stream);
gst_clear_object (&rtx->rid_repaired);
gst_clear_buffer (&rtx->dummy_writable);
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);
rtx->dummy_writable = gst_buffer_new ();
}
static gboolean
gst_rtp_rtx_receive_src_event (GstPad * pad, GstObject * parent,
GstEvent * event)
{
GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE_CAST (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, "got rtx request for seqnum: %u, ssrc: %X",
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_TRACE_OBJECT (rtx, "Retransmitted stream %X already associated "
"to its master, %X", GPOINTER_TO_UINT (ssrc2), ssrc);
} 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) {
/* same seqnum, same 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 translator into a FB NACK
*/
GST_LOG_OBJECT (rtx, "Duplicate request: seqnum: %u, ssrc: %X",
seqnum, ssrc);
} else {
/* same seqnum, different ssrc */
/* 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_INFO_OBJECT (rtx, "rejecting request for seqnum %u"
" of master stream %X; there is already a pending request "
"for the same seqnum on ssrc %X that has not expired",
seqnum, ssrc, assoc->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 %u of master stream %X"
" needs to be retransmitted", seqnum, ssrc);
GST_OBJECT_UNLOCK (rtx);
}
/* Transfer event upstream so that the request can actually 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;
}
static GstMemory *
rewrite_header_extensions (GstRtpRtxReceive * rtx, GstRTPBuffer * rtp)
{
gsize out_size = rtp->size[1] + 32;
guint16 bit_pattern;
guint8 *pdata;
guint wordlen;
GstMemory *mem;
GstMapInfo map;
mem = gst_allocator_alloc (NULL, out_size, NULL);
gst_memory_map (mem, &map, GST_MAP_READWRITE);
if (gst_rtp_buffer_get_extension_data (rtp, &bit_pattern, (gpointer) & pdata,
&wordlen)) {
GstRTPHeaderExtensionFlags ext_flags = 0;
gsize bytelen = wordlen * 4;
guint hdr_unit_bytes;
gsize read_offset = 0, write_offset = 4;
if (bit_pattern == 0xBEDE) {
/* one byte extensions */
hdr_unit_bytes = 1;
ext_flags |= GST_RTP_HEADER_EXTENSION_ONE_BYTE;
} else if (bit_pattern >> 4 == 0x100) {
/* two byte extensions */
hdr_unit_bytes = 2;
ext_flags |= GST_RTP_HEADER_EXTENSION_TWO_BYTE;
} else {
GST_DEBUG_OBJECT (rtx, "unknown extension bit pattern 0x%02x%02x",
bit_pattern >> 8, bit_pattern & 0xff);
goto copy_as_is;
}
GST_WRITE_UINT16_BE (map.data, bit_pattern);
while (TRUE) {
guint8 read_id, read_len;
if (read_offset + hdr_unit_bytes >= bytelen)
/* not enough remaning data */
break;
if (ext_flags & GST_RTP_HEADER_EXTENSION_ONE_BYTE) {
read_id = GST_READ_UINT8 (pdata + read_offset) >> 4;
read_len = (GST_READ_UINT8 (pdata + read_offset) & 0x0F) + 1;
read_offset += 1;
if (read_id == 0)
/* padding */
continue;
if (read_id == 15)
/* special id for possible future expansion */
break;
} else {
read_id = GST_READ_UINT8 (pdata + read_offset);
read_offset += 1;
if (read_id == 0)
/* padding */
continue;
read_len = GST_READ_UINT8 (pdata + read_offset);
read_offset += 1;
}
GST_TRACE_OBJECT (rtx, "found rtp header extension with id %u and "
"length %u", read_id, read_len);
/* Ignore extension headers where the size does not fit */
if (read_offset + read_len > bytelen) {
GST_WARNING_OBJECT (rtx, "Extension length extends past the "
"size of the extension data");
break;
}
/* rewrite the rtp-stream-id into a repaired-stream-id */
if (rtx->rid_stream
&& read_id == gst_rtp_header_extension_get_id (rtx->rid_repaired)) {
if (!gst_rtp_header_extension_read (rtx->rid_repaired, ext_flags,
&pdata[read_offset], read_len, rtx->dummy_writable)) {
GST_WARNING_OBJECT (rtx, "RTP header extension (%s) could "
"not read payloaded data", GST_OBJECT_NAME (rtx->rid_stream));
goto copy_as_is;
}
if (rtx->rid_repaired) {
guint8 write_id = gst_rtp_header_extension_get_id (rtx->rid_stream);
gsize written;
char *rid;
g_object_get (rtx->rid_repaired, "rid", &rid, NULL);
g_object_set (rtx->rid_stream, "rid", rid, NULL);
g_clear_pointer (&rid, g_free);
written =
gst_rtp_header_extension_write (rtx->rid_stream, rtp->buffer,
ext_flags, rtx->dummy_writable,
&map.data[write_offset + hdr_unit_bytes],
map.size - write_offset - hdr_unit_bytes);
GST_TRACE_OBJECT (rtx->rid_repaired, "wrote %" G_GSIZE_FORMAT,
written);
if (written <= 0) {
GST_WARNING_OBJECT (rtx, "Failed to rewrite RID for RTX");
goto copy_as_is;
} else {
if (ext_flags & GST_RTP_HEADER_EXTENSION_ONE_BYTE) {
map.data[write_offset] =
((write_id & 0x0F) << 4) | ((written - 1) & 0x0F);
} else if (ext_flags & GST_RTP_HEADER_EXTENSION_TWO_BYTE) {
map.data[write_offset] = write_id & 0xFF;
map.data[write_offset + 1] = written & 0xFF;
} else {
g_assert_not_reached ();
goto copy_as_is;
}
write_offset += written + hdr_unit_bytes;
}
}
} else {
/* TODO: may need to write mid at different times to the original
* buffer to account for the difference in timing of acknowledgement
* of the rtx ssrc from the original ssrc. This may add extra data to
* the header extension space that needs to be accounted for.
*/
memcpy (&map.data[write_offset],
&map.data[read_offset - hdr_unit_bytes], read_len + hdr_unit_bytes);
write_offset += read_len + hdr_unit_bytes;
}
read_offset += read_len;
}
/* subtract the ext header */
wordlen = write_offset / 4 + ((write_offset % 4) ? 1 : 0);
/* wordlen in the ext data doesn't include the 4-byte header */
GST_WRITE_UINT16_BE (map.data + 2, wordlen - 1);
if (wordlen * 4 > write_offset)
memset (&map.data[write_offset], 0, wordlen * 4 - write_offset);
GST_MEMDUMP_OBJECT (rtx, "generated ext data", map.data, wordlen * 4);
} else {
copy_as_is:
wordlen = rtp->size[1] / 4;
memcpy (map.data, rtp->data[1], rtp->size[1]);
GST_LOG_OBJECT (rtx, "copying data as-is");
}
gst_memory_unmap (mem, &map);
gst_memory_resize (mem, 0, wordlen * 4);
return mem;
}
/* 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 (GstRtpRtxReceive * 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 = rewrite_header_extensions (rtx, rtp);
gst_buffer_append_memory (new_buffer, mem);
}
/* copy payload and remove OSN */
g_assert_cmpint (rtp->size[2], >, 1);
payload_len = rtp->size[2] - 2;
mem = gst_allocator_alloc (NULL, payload_len, NULL);
gst_memory_map (mem, &map, GST_MAP_WRITE);
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_CAST (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;
if (rtx->rtx_pt_map_structure == NULL)
goto no_map;
/* map current rtp packet to parse its header */
if (!gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp))
goto invalid_buffer;
GST_MEMDUMP_OBJECT (rtx, "rtp header", rtp.map[0].data, rtp.map[0].size);
GST_MEMDUMP_OBJECT (rtx, "rtp ext", rtp.map[1].data, rtp.map[1].size);
GST_MEMDUMP_OBJECT (rtx, "rtp payload", rtp.map[2].data, rtp.map[2].size);
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;
/* check if there enough data to read OSN from the paylaod,
we need at least two bytes
*/
if (gst_rtp_buffer_get_payload_len (&rtp) > 1) {
/* 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)));
GST_DEBUG_OBJECT (rtx, "Got rtx packet: rtx seqnum %u, rtx ssrc %X, "
"rtx pt %u, orig seqnum %u, orig pt %u", seqnum, ssrc, payload_type,
orign_seqnum, origin_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_TRACE_OBJECT (rtx,
"packet is from retransmission stream %X already associated to "
"master stream %X", 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_LOG_OBJECT (rtx,
"associating retransmitted stream %X to master stream %X thanks "
"to rtx packet %u (orig seqnum %u)", ssrc, assoc->ssrc, seqnum,
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, "
"master and rtx SSRCs are the same (%X)\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_INFO_OBJECT (rtx,
"dropping rtx packet %u because its orig seqnum (%u) is not in our"
" pending retransmission requests", seqnum, orign_seqnum);
drop = TRUE;
}
}
} else {
/* the rtx packet is empty */
GST_DEBUG_OBJECT (rtx, "drop rtx packet because it is empty");
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 (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, "pushing packet seqnum:%u from restransmission "
"stream ssrc: %X (master ssrc %X)", orign_seqnum, ssrc2,
GPOINTER_TO_UINT (ssrc1));
ret = gst_pad_push (rtx->srcpad, new_buffer);
} else {
GST_TRACE_OBJECT (rtx, "pushing packet seqnum:%u from master stream "
"ssrc: %X", seqnum, ssrc);
ret = gst_pad_push (rtx->srcpad, buffer);
}
return ret;
no_map:
{
GST_DEBUG_OBJECT (pad, "No map set, passthrough");
return gst_pad_push (rtx->srcpad, buffer);
}
invalid_buffer:
{
GST_INFO_OBJECT (pad, "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_CAST (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_CAST (object);
switch (prop_id) {
case PROP_SSRC_MAP:
GST_OBJECT_LOCK (rtx);
if (rtx->external_ssrc_map)
gst_structure_free (rtx->external_ssrc_map);
rtx->external_ssrc_map = g_value_dup_boxed (value);
g_hash_table_remove_all (rtx->ssrc2_ssrc1_map);
gst_structure_foreach (rtx->external_ssrc_map,
structure_to_hash_table_inv, rtx->ssrc2_ssrc1_map);
GST_OBJECT_UNLOCK (rtx);
break;
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_CAST (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;
}
Reference in a new issue View git blame Copy permalink