mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-12-01 22:21:13 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/subprojects/gst-plugins-base/gst-libs/gst/rtp/gstrtpbasedepayload.c
Matt Crane b11169bd32 rtpbasedepayload: Drop redundant reference timestamp buffer meta in RTP depayloaders 
Currently, when rtspsrc property add-reference-timestamp-metadata=true,
a downstream rtph264depay element will attach multiple copies of the
same GstReferenceTimestampMeta to the depayloaded media buffers. This
can have signficant performance impacts further downstream in a pipeline
like the following:

    rtspsrc add-reference-timestamp-metadata=true ! rtph264depay ! h264parse ! ... ! rtph264pay ! ...

For example, if there are 10 packet buffers for a frame of RTP H.264
video, each of those packet buffers will contain the same reference
timestamp meta. The rtph264depay element will then attach all 10
metadata to the depayloaded frame. And then later when we payload the
frame buffer again for proxying, we now have 10 more buffers each with
10 instance of the same metadata. Allocating/deallocating 100+ instances
of metadata @ 30fps for multiple streams has a pretty large performance
impact.

Fixes https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/1578

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/3431>
2022-11-19 07:57:44 +00:00

1720 lines
51 KiB
C
Raw Blame History

/* GStreamer
* Copyright (C) <2005> Philippe Khalaf <burger@speedy.org>
* Copyright (C) <2005> Nokia Corporation <kai.vehmanen@nokia.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
/**
* SECTION:gstrtpbasedepayload
* @title: GstRTPBaseDepayload
* @short_description: Base class for RTP depayloader
*
* Provides a base class for RTP depayloaders
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstrtpbasedepayload.h"
#include "gstrtpmeta.h"
#include "gstrtphdrext.h"
GST_DEBUG_CATEGORY_STATIC (rtpbasedepayload_debug);
#define GST_CAT_DEFAULT (rtpbasedepayload_debug)
static GstStaticCaps ntp_reference_timestamp_caps =
GST_STATIC_CAPS ("timestamp/x-ntp");
struct _GstRTPBaseDepayloadPrivate
{
GstClockTime npt_start;
GstClockTime npt_stop;
gdouble play_speed;
gdouble play_scale;
guint clock_base;
gboolean onvif_mode;
gboolean discont;
GstClockTime pts;
GstClockTime dts;
GstClockTime duration;
GstClockTime ref_ts;
guint32 last_ssrc;
guint32 last_seqnum;
guint32 last_rtptime;
guint32 next_seqnum;
gint max_reorder;
gboolean auto_hdr_ext;
gboolean negotiated;
GstCaps *last_caps;
GstEvent *segment_event;
guint32 segment_seqnum; /* Note: this is a GstEvent seqnum */
gboolean source_info;
GstBuffer *input_buffer;
GstFlowReturn process_flow_ret;
/* array of GstRTPHeaderExtension's * */
GPtrArray *header_exts;
};
/* Filter signals and args */
enum
{
SIGNAL_0,
SIGNAL_REQUEST_EXTENSION,
SIGNAL_ADD_EXTENSION,
SIGNAL_CLEAR_EXTENSIONS,
LAST_SIGNAL
};
static guint gst_rtp_base_depayload_signals[LAST_SIGNAL] = { 0 };
#define DEFAULT_SOURCE_INFO FALSE
#define DEFAULT_MAX_REORDER 100
#define DEFAULT_AUTO_HEADER_EXTENSION TRUE
enum
{
PROP_0,
PROP_STATS,
PROP_SOURCE_INFO,
PROP_MAX_REORDER,
PROP_AUTO_HEADER_EXTENSION,
PROP_LAST
};
static void gst_rtp_base_depayload_finalize (GObject * object);
static void gst_rtp_base_depayload_set_property (GObject * object,
guint prop_id, const GValue * value, GParamSpec * pspec);
static void gst_rtp_base_depayload_get_property (GObject * object,
guint prop_id, GValue * value, GParamSpec * pspec);
static GstFlowReturn gst_rtp_base_depayload_chain (GstPad * pad,
GstObject * parent, GstBuffer * in);
static GstFlowReturn gst_rtp_base_depayload_chain_list (GstPad * pad,
GstObject * parent, GstBufferList * list);
static gboolean gst_rtp_base_depayload_handle_sink_event (GstPad * pad,
GstObject * parent, GstEvent * event);
static GstStateChangeReturn gst_rtp_base_depayload_change_state (GstElement *
element, GstStateChange transition);
static gboolean gst_rtp_base_depayload_packet_lost (GstRTPBaseDepayload *
filter, GstEvent * event);
static gboolean gst_rtp_base_depayload_handle_event (GstRTPBaseDepayload *
filter, GstEvent * event);
static GstElementClass *parent_class = NULL;
static gint private_offset = 0;
static void gst_rtp_base_depayload_class_init (GstRTPBaseDepayloadClass *
klass);
static void gst_rtp_base_depayload_init (GstRTPBaseDepayload * rtpbasepayload,
GstRTPBaseDepayloadClass * klass);
static GstEvent *create_segment_event (GstRTPBaseDepayload * filter,
guint rtptime, GstClockTime position);
static void gst_rtp_base_depayload_add_extension (GstRTPBaseDepayload *
rtpbasepayload, GstRTPHeaderExtension * ext);
static void gst_rtp_base_depayload_clear_extensions (GstRTPBaseDepayload *
rtpbasepayload);
GType
gst_rtp_base_depayload_get_type (void)
{
static GType rtp_base_depayload_type = 0;
if (g_once_init_enter ((gsize *) & rtp_base_depayload_type)) {
static const GTypeInfo rtp_base_depayload_info = {
sizeof (GstRTPBaseDepayloadClass),
NULL,
NULL,
(GClassInitFunc) gst_rtp_base_depayload_class_init,
NULL,
NULL,
sizeof (GstRTPBaseDepayload),
0,
(GInstanceInitFunc) gst_rtp_base_depayload_init,
};
GType _type;
_type = g_type_register_static (GST_TYPE_ELEMENT, "GstRTPBaseDepayload",
&rtp_base_depayload_info, G_TYPE_FLAG_ABSTRACT);
private_offset =
g_type_add_instance_private (_type,
sizeof (GstRTPBaseDepayloadPrivate));
g_once_init_leave ((gsize *) & rtp_base_depayload_type, _type);
}
return rtp_base_depayload_type;
}
static inline GstRTPBaseDepayloadPrivate *
gst_rtp_base_depayload_get_instance_private (GstRTPBaseDepayload * self)
{
return (G_STRUCT_MEMBER_P (self, private_offset));
}
static GstRTPHeaderExtension *
gst_rtp_base_depayload_request_extension_default (GstRTPBaseDepayload *
depayload, guint ext_id, const gchar * uri)
{
GstRTPHeaderExtension *ext = NULL;
if (!depayload->priv->auto_hdr_ext)
return NULL;
ext = gst_rtp_header_extension_create_from_uri (uri);
if (ext) {
GST_DEBUG_OBJECT (depayload,
"Automatically enabled extension %s for uri \'%s\'",
GST_ELEMENT_NAME (ext), uri);
gst_rtp_header_extension_set_id (ext, ext_id);
} else {
GST_DEBUG_OBJECT (depayload,
"Didn't find any extension implementing uri \'%s\'", uri);
}
return ext;
}
static gboolean
extension_accumulator (GSignalInvocationHint * ihint,
GValue * return_accu, const GValue * handler_return, gpointer data)
{
gpointer ext;
/* Call default handler if user callback didn't create the extension */
ext = g_value_get_object (handler_return);
if (!ext)
return TRUE;
g_value_set_object (return_accu, ext);
return FALSE;
}
static void
gst_rtp_base_depayload_class_init (GstRTPBaseDepayloadClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *gstelement_class;
gobject_class = G_OBJECT_CLASS (klass);
gstelement_class = (GstElementClass *) klass;
parent_class = g_type_class_peek_parent (klass);
if (private_offset != 0)
g_type_class_adjust_private_offset (klass, &private_offset);
gobject_class->finalize = gst_rtp_base_depayload_finalize;
gobject_class->set_property = gst_rtp_base_depayload_set_property;
gobject_class->get_property = gst_rtp_base_depayload_get_property;
/**
* GstRTPBaseDepayload:stats:
*
* Various depayloader statistics retrieved atomically (and are therefore
* synchroized with each other). This property return a GstStructure named
* application/x-rtp-depayload-stats containing the following fields relating to
* the last processed buffer and current state of the stream being depayloaded:
*
* * `clock-rate`: #G_TYPE_UINT, clock-rate of the stream
* * `npt-start`: #G_TYPE_UINT64, time of playback start
* * `npt-stop`: #G_TYPE_UINT64, time of playback stop
* * `play-speed`: #G_TYPE_DOUBLE, the playback speed
* * `play-scale`: #G_TYPE_DOUBLE, the playback scale
* * `running-time-dts`: #G_TYPE_UINT64, the last running-time of the
* last DTS
* * `running-time-pts`: #G_TYPE_UINT64, the last running-time of the
* last PTS
* * `seqnum`: #G_TYPE_UINT, the last seen seqnum
* * `timestamp`: #G_TYPE_UINT, the last seen RTP timestamp
**/
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_STATS,
g_param_spec_boxed ("stats", "Statistics", "Various statistics",
GST_TYPE_STRUCTURE, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
/**
* GstRTPBaseDepayload:source-info:
*
* Add RTP source information found in RTP header as meta to output buffer.
*
* Since: 1.16
**/
g_object_class_install_property (gobject_class, PROP_SOURCE_INFO,
g_param_spec_boolean ("source-info", "RTP source information",
"Add RTP source information as buffer meta",
DEFAULT_SOURCE_INFO, G_PARAM_READWRITE));
/**
* GstRTPBaseDepayload:max-reorder:
*
* Max seqnum reorder before the sender is assumed to have restarted.
*
* When max-reorder is set to 0 all reordered/duplicate packets are
* considered coming from a restarted sender.
*
* Since: 1.18
**/
g_object_class_install_property (gobject_class, PROP_MAX_REORDER,
g_param_spec_int ("max-reorder", "Max Reorder",
"Max seqnum reorder before assuming sender has restarted",
0, G_MAXINT, DEFAULT_MAX_REORDER, G_PARAM_READWRITE));
/**
* GstRTPBaseDepayload:auto-header-extension:
*
* If enabled, the depayloader will automatically try to enable all the
* RTP header extensions provided in the sink caps, saving the application
* the need to handle these extensions manually using the
* GstRTPBaseDepayload::request-extension: signal.
*
* Since: 1.20
*/
g_object_class_install_property (G_OBJECT_CLASS (klass),
PROP_AUTO_HEADER_EXTENSION, g_param_spec_boolean ("auto-header-extension",
"Automatic RTP header extension",
"Whether RTP header extensions should be automatically enabled, if an implementation is available",
DEFAULT_AUTO_HEADER_EXTENSION,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
/**
* GstRTPBaseDepayload::request-extension:
* @object: the #GstRTPBaseDepayload
* @ext_id: the extension id being requested
* @ext_uri: (nullable): the extension URI being requested
*
* The returned @ext must be configured with the correct @ext_id and with the
* necessary attributes as required by the extension implementation.
*
* Returns: (transfer full) (nullable): the #GstRTPHeaderExtension for @ext_id, or %NULL
*
* Since: 1.20
*/
gst_rtp_base_depayload_signals[SIGNAL_REQUEST_EXTENSION] =
g_signal_new_class_handler ("request-extension",
G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
G_CALLBACK (gst_rtp_base_depayload_request_extension_default),
extension_accumulator, NULL, NULL,
GST_TYPE_RTP_HEADER_EXTENSION, 2, G_TYPE_UINT, G_TYPE_STRING);
/**
* GstRTPBaseDepayload::add-extension:
* @object: the #GstRTPBaseDepayload
* @ext: (transfer full): the #GstRTPHeaderExtension
*
* Add @ext as an extension for reading part of an RTP header extension from
* incoming RTP packets.
*
* Since: 1.20
*/
gst_rtp_base_depayload_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_base_depayload_add_extension), NULL, NULL, NULL,
G_TYPE_NONE, 1, GST_TYPE_RTP_HEADER_EXTENSION);
/**
* GstRTPBaseDepayload::clear-extensions:
* @object: the #GstRTPBaseDepayload
*
* Clear all RTP header extensions used by this depayloader.
*
* Since: 1.20
*/
gst_rtp_base_depayload_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_base_depayload_clear_extensions), NULL, NULL, NULL,
G_TYPE_NONE, 0);
gstelement_class->change_state = gst_rtp_base_depayload_change_state;
klass->packet_lost = gst_rtp_base_depayload_packet_lost;
klass->handle_event = gst_rtp_base_depayload_handle_event;
GST_DEBUG_CATEGORY_INIT (rtpbasedepayload_debug, "rtpbasedepayload", 0,
"Base class for RTP Depayloaders");
}
static void
gst_rtp_base_depayload_init (GstRTPBaseDepayload * filter,
GstRTPBaseDepayloadClass * klass)
{
GstPadTemplate *pad_template;
GstRTPBaseDepayloadPrivate *priv;
priv = gst_rtp_base_depayload_get_instance_private (filter);
filter->priv = priv;
GST_DEBUG_OBJECT (filter, "init");
pad_template =
gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "sink");
g_return_if_fail (pad_template != NULL);
filter->sinkpad = gst_pad_new_from_template (pad_template, "sink");
gst_pad_set_chain_function (filter->sinkpad, gst_rtp_base_depayload_chain);
gst_pad_set_chain_list_function (filter->sinkpad,
gst_rtp_base_depayload_chain_list);
gst_pad_set_event_function (filter->sinkpad,
gst_rtp_base_depayload_handle_sink_event);
gst_element_add_pad (GST_ELEMENT (filter), filter->sinkpad);
pad_template =
gst_element_class_get_pad_template (GST_ELEMENT_CLASS (klass), "src");
g_return_if_fail (pad_template != NULL);
filter->srcpad = gst_pad_new_from_template (pad_template, "src");
gst_pad_use_fixed_caps (filter->srcpad);
gst_element_add_pad (GST_ELEMENT (filter), filter->srcpad);
priv->npt_start = 0;
priv->npt_stop = -1;
priv->play_speed = 1.0;
priv->play_scale = 1.0;
priv->clock_base = -1;
priv->onvif_mode = FALSE;
priv->dts = -1;
priv->pts = -1;
priv->duration = -1;
priv->ref_ts = -1;
priv->source_info = DEFAULT_SOURCE_INFO;
priv->max_reorder = DEFAULT_MAX_REORDER;
priv->auto_hdr_ext = DEFAULT_AUTO_HEADER_EXTENSION;
gst_segment_init (&filter->segment, GST_FORMAT_UNDEFINED);
priv->header_exts =
g_ptr_array_new_with_free_func ((GDestroyNotify) gst_object_unref);
}
static void
gst_rtp_base_depayload_finalize (GObject * object)
{
GstRTPBaseDepayload *rtpbasedepayload = GST_RTP_BASE_DEPAYLOAD (object);
g_ptr_array_unref (rtpbasedepayload->priv->header_exts);
rtpbasedepayload->priv->header_exts = NULL;
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static void
add_and_ref_item (GstRTPHeaderExtension * ext, GPtrArray * ret)
{
g_ptr_array_add (ret, gst_object_ref (ext));
}
static void
remove_item_from (GstRTPHeaderExtension * ext, GPtrArray * ret)
{
g_ptr_array_remove_fast (ret, ext);
}
static void
add_item_to (GstRTPHeaderExtension * ext, GPtrArray * ret)
{
g_ptr_array_add (ret, ext);
}
static gboolean
gst_rtp_base_depayload_setcaps (GstRTPBaseDepayload * filter, GstCaps * caps)
{
GstRTPBaseDepayloadClass *bclass;
GstRTPBaseDepayloadPrivate *priv;
gboolean res = TRUE;
GstStructure *caps_struct;
const GValue *value;
priv = filter->priv;
bclass = GST_RTP_BASE_DEPAYLOAD_GET_CLASS (filter);
GST_DEBUG_OBJECT (filter, "Set caps %" GST_PTR_FORMAT, caps);
if (priv->last_caps) {
if (gst_caps_is_equal (priv->last_caps, caps)) {
res = TRUE;
goto caps_not_changed;
} else {
gst_caps_unref (priv->last_caps);
priv->last_caps = NULL;
}
}
caps_struct = gst_caps_get_structure (caps, 0);
value = gst_structure_get_value (caps_struct, "onvif-mode");
if (value && G_VALUE_HOLDS_BOOLEAN (value))
priv->onvif_mode = g_value_get_boolean (value);
else
priv->onvif_mode = FALSE;
GST_DEBUG_OBJECT (filter, "Onvif mode: %d", priv->onvif_mode);
if (priv->onvif_mode)
filter->need_newsegment = FALSE;
/* get other values for newsegment */
value = gst_structure_get_value (caps_struct, "npt-start");
if (value && G_VALUE_HOLDS_UINT64 (value))
priv->npt_start = g_value_get_uint64 (value);
else
priv->npt_start = 0;
GST_DEBUG_OBJECT (filter, "NPT start %" G_GUINT64_FORMAT, priv->npt_start);
value = gst_structure_get_value (caps_struct, "npt-stop");
if (value && G_VALUE_HOLDS_UINT64 (value))
priv->npt_stop = g_value_get_uint64 (value);
else
priv->npt_stop = -1;
GST_DEBUG_OBJECT (filter, "NPT stop %" G_GUINT64_FORMAT, priv->npt_stop);
value = gst_structure_get_value (caps_struct, "play-speed");
if (value && G_VALUE_HOLDS_DOUBLE (value))
priv->play_speed = g_value_get_double (value);
else
priv->play_speed = 1.0;
value = gst_structure_get_value (caps_struct, "play-scale");
if (value && G_VALUE_HOLDS_DOUBLE (value))
priv->play_scale = g_value_get_double (value);
else
priv->play_scale = 1.0;
value = gst_structure_get_value (caps_struct, "clock-base");
if (value && G_VALUE_HOLDS_UINT (value))
priv->clock_base = g_value_get_uint (value);
else
priv->clock_base = -1;
{
/* ensure we have header extension implementations for the list in the
* caps */
guint i, j, n_fields = gst_structure_n_fields (caps_struct);
GPtrArray *header_exts = g_ptr_array_new_with_free_func (gst_object_unref);
GPtrArray *to_add = g_ptr_array_new ();
GPtrArray *to_remove = g_ptr_array_new ();
GST_OBJECT_LOCK (filter);
g_ptr_array_foreach (filter->priv->header_exts,
(GFunc) add_and_ref_item, header_exts);
GST_OBJECT_UNLOCK (filter);
for (i = 0; i < n_fields; i++) {
const gchar *field_name = gst_structure_nth_field_name (caps_struct, i);
if (g_str_has_prefix (field_name, "extmap-")) {
const GValue *val;
const gchar *uri = NULL;
gchar *nptr;
guint ext_id;
GstRTPHeaderExtension *ext = NULL;
errno = 0;
ext_id = g_ascii_strtoull (&field_name[strlen ("extmap-")], &nptr, 10);
if (errno != 0 || (ext_id == 0 && field_name == nptr)) {
GST_WARNING_OBJECT (filter, "could not parse id from %s", field_name);
res = FALSE;
goto ext_out;
}
val = gst_structure_get_value (caps_struct, field_name);
if (G_VALUE_HOLDS_STRING (val)) {
uri = g_value_get_string (val);
} else if (GST_VALUE_HOLDS_ARRAY (val)) {
/* the uri is the second value in the array */
const GValue *str = gst_value_array_get_value (val, 1);
if (G_VALUE_HOLDS_STRING (str)) {
uri = g_value_get_string (str);
}
}
if (!uri) {
GST_WARNING_OBJECT (filter, "could not get extmap uri for "
"field %s", field_name);
res = FALSE;
goto ext_out;
}
/* try to find if this extension mapping already exists */
for (j = 0; j < header_exts->len; j++) {
ext = g_ptr_array_index (header_exts, j);
if (gst_rtp_header_extension_get_id (ext) == ext_id) {
if (g_strcmp0 (uri, gst_rtp_header_extension_get_uri (ext)) == 0) {
/* still matching, we're good, set attributes from caps in case
* the caps have changed */
if (!gst_rtp_header_extension_set_attributes_from_caps (ext,
caps)) {
GST_WARNING_OBJECT (filter,
"Failed to configure rtp header " "extension %"
GST_PTR_FORMAT " attributes from caps %" GST_PTR_FORMAT,
ext, caps);
res = FALSE;
goto ext_out;
}
break;
} else {
GST_DEBUG_OBJECT (filter, "extension id %u"
"was replaced with a different extension uri "
"original:\'%s' vs \'%s\'", ext_id,
gst_rtp_header_extension_get_uri (ext), uri);
g_ptr_array_add (to_remove, ext);
ext = NULL;
break;
}
} else {
ext = NULL;
}
}
/* if no extension, attempt to request one */
if (!ext) {
GST_DEBUG_OBJECT (filter, "requesting extension for id %u"
" and uri %s", ext_id, uri);
g_signal_emit (filter,
gst_rtp_base_depayload_signals[SIGNAL_REQUEST_EXTENSION], 0,
ext_id, uri, &ext);
GST_DEBUG_OBJECT (filter, "request returned extension %p \'%s\' "
"for id %u and uri %s", ext,
ext ? GST_OBJECT_NAME (ext) : "", ext_id, uri);
/* We require the caller to set the appropriate extension if it's required */
if (ext && gst_rtp_header_extension_get_id (ext) != ext_id) {
g_warning ("\'request-extension\' signal provided an rtp header "
"extension for uri \'%s\' that does not match the requested "
"extension id %u", uri, ext_id);
gst_clear_object (&ext);
}
if (ext && !gst_rtp_header_extension_set_attributes_from_caps (ext,
caps)) {
GST_WARNING_OBJECT (filter,
"Failed to configure rtp header " "extension %"
GST_PTR_FORMAT " attributes from caps %" GST_PTR_FORMAT,
ext, caps);
res = FALSE;
g_clear_object (&ext);
goto ext_out;
}
if (ext)
g_ptr_array_add (to_add, ext);
}
}
}
/* Note: we intentionally don't remove extensions that are not listed
* in caps */
GST_OBJECT_LOCK (filter);
g_ptr_array_foreach (to_remove, (GFunc) remove_item_from,
filter->priv->header_exts);
g_ptr_array_foreach (to_add, (GFunc) add_item_to,
filter->priv->header_exts);
GST_OBJECT_UNLOCK (filter);
ext_out:
g_ptr_array_unref (to_add);
g_ptr_array_unref (to_remove);
g_ptr_array_unref (header_exts);
if (!res)
return res;
}
if (bclass->set_caps) {
res = bclass->set_caps (filter, caps);
if (!res) {
GST_WARNING_OBJECT (filter, "Subclass rejected caps %" GST_PTR_FORMAT,
caps);
}
} else {
res = TRUE;
}
priv->negotiated = res;
if (priv->negotiated)
priv->last_caps = gst_caps_ref (caps);
return res;
caps_not_changed:
{
GST_DEBUG_OBJECT (filter, "Caps did not change");
return res;
}
}
/* takes ownership of the input buffer */
static GstFlowReturn
gst_rtp_base_depayload_handle_buffer (GstRTPBaseDepayload * filter,
GstRTPBaseDepayloadClass * bclass, GstBuffer * in)
{
GstBuffer *(*process_rtp_packet_func) (GstRTPBaseDepayload * base,
GstRTPBuffer * rtp_buffer);
GstBuffer *(*process_func) (GstRTPBaseDepayload * base, GstBuffer * in);
GstRTPBaseDepayloadPrivate *priv;
GstBuffer *out_buf;
guint32 ssrc;
guint16 seqnum;
guint32 rtptime;
gboolean discont, buf_discont;
gint gap;
GstRTPBuffer rtp = { NULL };
GstReferenceTimestampMeta *meta;
GstCaps *ref_caps;
priv = filter->priv;
priv->process_flow_ret = GST_FLOW_OK;
process_func = bclass->process;
process_rtp_packet_func = bclass->process_rtp_packet;
/* we must have a setcaps first */
if (G_UNLIKELY (!priv->negotiated))
goto not_negotiated;
/* Check for duplicate reference timestamp metadata */
ref_caps = gst_static_caps_get (&ntp_reference_timestamp_caps);
meta = gst_buffer_get_reference_timestamp_meta (in, ref_caps);
gst_caps_unref (ref_caps);
if (meta) {
guint64 ref_ts = meta->timestamp;
if (ref_ts == priv->ref_ts) {
/* Drop the redundant/duplicate reference timstamp metadata */
in = gst_buffer_make_writable (in);
gst_buffer_remove_meta (in, GST_META_CAST (meta));
} else {
priv->ref_ts = ref_ts;
}
}
if (G_UNLIKELY (!gst_rtp_buffer_map (in, GST_MAP_READ, &rtp)))
goto invalid_buffer;
buf_discont = GST_BUFFER_IS_DISCONT (in);
priv->pts = GST_BUFFER_PTS (in);
priv->dts = GST_BUFFER_DTS (in);
priv->duration = GST_BUFFER_DURATION (in);
ssrc = gst_rtp_buffer_get_ssrc (&rtp);
seqnum = gst_rtp_buffer_get_seq (&rtp);
rtptime = gst_rtp_buffer_get_timestamp (&rtp);
priv->last_seqnum = seqnum;
priv->last_rtptime = rtptime;
discont = buf_discont;
GST_LOG_OBJECT (filter, "discont %d, seqnum %u, rtptime %u, pts %"
GST_TIME_FORMAT ", dts %" GST_TIME_FORMAT, buf_discont, seqnum, rtptime,
GST_TIME_ARGS (priv->pts), GST_TIME_ARGS (priv->dts));
/* Check seqnum. This is a very simple check that makes sure that the seqnums
* are strictly increasing, dropping anything that is out of the ordinary. We
* can only do this when the next_seqnum is known. */
if (G_LIKELY (priv->next_seqnum != -1)) {
if (ssrc != priv->last_ssrc) {
GST_LOG_OBJECT (filter,
"New ssrc %u (current ssrc %u), sender restarted",
ssrc, priv->last_ssrc);
discont = TRUE;
} else {
gap = gst_rtp_buffer_compare_seqnum (seqnum, priv->next_seqnum);
/* if we have no gap, all is fine */
if (G_UNLIKELY (gap != 0)) {
GST_LOG_OBJECT (filter, "got packet %u, expected %u, gap %d", seqnum,
priv->next_seqnum, gap);
if (gap < 0) {
/* seqnum > next_seqnum, we are missing some packets, this is always a
* DISCONT. */
GST_LOG_OBJECT (filter, "%d missing packets", gap);
discont = TRUE;
} else {
/* seqnum < next_seqnum, we have seen this packet before, have a
* reordered packet or the sender could be restarted. If the packet
* is not too old, we throw it away as a duplicate. Otherwise we
* mark discont and continue assuming the sender has restarted. See
* also RFC 4737. */
if (gap <= priv->max_reorder) {
GST_WARNING_OBJECT (filter, "got old packet %u, expected %u, "
"gap %d <= max_reorder (%d), dropping!",
seqnum, priv->next_seqnum, gap, priv->max_reorder);
goto dropping;
}
GST_WARNING_OBJECT (filter, "got old packet %u, expected %u, "
"marking discont", seqnum, priv->next_seqnum);
discont = TRUE;
}
}
}
}
priv->next_seqnum = (seqnum + 1) & 0xffff;
priv->last_ssrc = ssrc;
if (G_UNLIKELY (discont)) {
priv->discont = TRUE;
if (!buf_discont) {
gpointer old_inbuf = in;
/* we detected a seqnum discont but the buffer was not flagged with a discont,
* set the discont flag so that the subclass can throw away old data. */
GST_LOG_OBJECT (filter, "mark DISCONT on input buffer");
in = gst_buffer_make_writable (in);
GST_BUFFER_FLAG_SET (in, GST_BUFFER_FLAG_DISCONT);
/* depayloaders will check flag on rtpbuffer->buffer, so if the input
* buffer was not writable already we need to remap to make our
* newly-flagged buffer current on the rtpbuffer */
if (in != old_inbuf) {
gst_rtp_buffer_unmap (&rtp);
if (G_UNLIKELY (!gst_rtp_buffer_map (in, GST_MAP_READ, &rtp)))
goto invalid_buffer;
}
}
}
/* prepare segment event if needed */
if (filter->need_newsegment) {
priv->segment_event = create_segment_event (filter, rtptime,
GST_BUFFER_PTS (in));
filter->need_newsegment = FALSE;
}
priv->input_buffer = in;
if (process_rtp_packet_func != NULL) {
out_buf = process_rtp_packet_func (filter, &rtp);
gst_rtp_buffer_unmap (&rtp);
} else if (process_func != NULL) {
gst_rtp_buffer_unmap (&rtp);
out_buf = process_func (filter, in);
} else {
goto no_process;
}
/* let's send it out to processing */
if (out_buf) {
if (priv->process_flow_ret == GST_FLOW_OK)
priv->process_flow_ret = gst_rtp_base_depayload_push (filter, out_buf);
else
gst_buffer_unref (out_buf);
}
gst_buffer_unref (in);
priv->input_buffer = NULL;
return priv->process_flow_ret;
/* ERRORS */
not_negotiated:
{
/* this is not fatal but should be filtered earlier */
GST_ELEMENT_ERROR (filter, CORE, NEGOTIATION,
("No RTP format was negotiated."),
("Input buffers need to have RTP caps set on them. This is usually "
"achieved by setting the 'caps' property of the upstream source "
"element (often udpsrc or appsrc), or by putting a capsfilter "
"element before the depayloader and setting the 'caps' property "
"on that. Also see http://cgit.freedesktop.org/gstreamer/"
"gst-plugins-good/tree/gst/rtp/README"));
gst_buffer_unref (in);
return GST_FLOW_NOT_NEGOTIATED;
}
invalid_buffer:
{
/* this is not fatal but should be filtered earlier */
GST_ELEMENT_WARNING (filter, STREAM, DECODE, (NULL),
("Received invalid RTP payload, dropping"));
gst_buffer_unref (in);
return GST_FLOW_OK;
}
dropping:
{
gst_rtp_buffer_unmap (&rtp);
gst_buffer_unref (in);
return GST_FLOW_OK;
}
no_process:
{
gst_rtp_buffer_unmap (&rtp);
/* this is not fatal but should be filtered earlier */
GST_ELEMENT_ERROR (filter, STREAM, NOT_IMPLEMENTED, (NULL),
("The subclass does not have a process or process_rtp_packet method"));
gst_buffer_unref (in);
return GST_FLOW_ERROR;
}
}
static GstFlowReturn
gst_rtp_base_depayload_chain (GstPad * pad, GstObject * parent, GstBuffer * in)
{
GstRTPBaseDepayloadClass *bclass;
GstRTPBaseDepayload *basedepay;
GstFlowReturn flow_ret;
basedepay = GST_RTP_BASE_DEPAYLOAD_CAST (parent);
bclass = GST_RTP_BASE_DEPAYLOAD_GET_CLASS (basedepay);
flow_ret = gst_rtp_base_depayload_handle_buffer (basedepay, bclass, in);
return flow_ret;
}
static GstFlowReturn
gst_rtp_base_depayload_chain_list (GstPad * pad, GstObject * parent,
GstBufferList * list)
{
GstRTPBaseDepayloadClass *bclass;
GstRTPBaseDepayload *basedepay;
GstFlowReturn flow_ret;
GstBuffer *buffer;
guint i, len;
basedepay = GST_RTP_BASE_DEPAYLOAD_CAST (parent);
bclass = GST_RTP_BASE_DEPAYLOAD_GET_CLASS (basedepay);
flow_ret = GST_FLOW_OK;
/* chain each buffer in list individually */
len = gst_buffer_list_length (list);
if (len == 0)
goto done;
for (i = 0; i < len; i++) {
buffer = gst_buffer_list_get (list, i);
/* handle_buffer takes ownership of input buffer */
/* FIXME: add a way to steal buffers from list as we will unref it anyway */
gst_buffer_ref (buffer);
/* Should we fix up any missing timestamps for list buffers here
* (e.g. set to first or previous timestamp in list) or just assume
* the's a jitterbuffer that will have done that for us? */
flow_ret = gst_rtp_base_depayload_handle_buffer (basedepay, bclass, buffer);
if (flow_ret != GST_FLOW_OK)
break;
}
done:
gst_buffer_list_unref (list);
return flow_ret;
}
static gboolean
gst_rtp_base_depayload_handle_event (GstRTPBaseDepayload * filter,
GstEvent * event)
{
gboolean res = TRUE;
gboolean forward = TRUE;
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_FLUSH_STOP:
GST_OBJECT_LOCK (filter);
gst_segment_init (&filter->segment, GST_FORMAT_UNDEFINED);
GST_OBJECT_UNLOCK (filter);
filter->need_newsegment = !filter->priv->onvif_mode;
filter->priv->next_seqnum = -1;
filter->priv->ref_ts = -1;
gst_event_replace (&filter->priv->segment_event, NULL);
break;
case GST_EVENT_CAPS:
{
GstCaps *caps;
gst_event_parse_caps (event, &caps);
res = gst_rtp_base_depayload_setcaps (filter, caps);
forward = FALSE;
break;
}
case GST_EVENT_SEGMENT:
{
GstSegment segment;
GST_OBJECT_LOCK (filter);
gst_event_copy_segment (event, &segment);
if (segment.format != GST_FORMAT_TIME) {
GST_ERROR_OBJECT (filter, "Segment with non-TIME format not supported");
res = FALSE;
}
filter->priv->segment_seqnum = gst_event_get_seqnum (event);
filter->segment = segment;
GST_OBJECT_UNLOCK (filter);
/* In ONVIF mode, upstream is expected to send us the correct segment */
if (!filter->priv->onvif_mode) {
/* don't pass the event downstream, we generate our own segment including
* the NTP time and other things we receive in caps */
forward = FALSE;
}
break;
}
case GST_EVENT_CUSTOM_DOWNSTREAM:
{
GstRTPBaseDepayloadClass *bclass;
bclass = GST_RTP_BASE_DEPAYLOAD_GET_CLASS (filter);
if (gst_event_has_name (event, "GstRTPPacketLost")) {
/* we get this event from the jitterbuffer when it considers a packet as
* being lost. We send it to our packet_lost vmethod. The default
* implementation will make time progress by pushing out a GAP event.
* Subclasses can override and do one of the following:
* - Adjust timestamp/duration to something more accurate before
* calling the parent (default) packet_lost method.
* - do some more advanced error concealing on the already received
* (fragmented) packets.
* - ignore the packet lost.
*/
if (bclass->packet_lost)
res = bclass->packet_lost (filter, event);
forward = FALSE;
}
break;
}
default:
break;
}
if (forward)
res = gst_pad_push_event (filter->srcpad, event);
else
gst_event_unref (event);
return res;
}
static gboolean
gst_rtp_base_depayload_handle_sink_event (GstPad * pad, GstObject * parent,
GstEvent * event)
{
gboolean res = FALSE;
GstRTPBaseDepayload *filter;
GstRTPBaseDepayloadClass *bclass;
filter = GST_RTP_BASE_DEPAYLOAD (parent);
bclass = GST_RTP_BASE_DEPAYLOAD_GET_CLASS (filter);
if (bclass->handle_event)
res = bclass->handle_event (filter, event);
else
gst_event_unref (event);
return res;
}
static GstEvent *
create_segment_event (GstRTPBaseDepayload * filter, guint rtptime,
GstClockTime position)
{
GstEvent *event;
GstClockTime start, stop, running_time;
GstRTPBaseDepayloadPrivate *priv;
GstSegment segment;
priv = filter->priv;
/* We don't need the object lock around - the segment
* can't change here while we're holding the STREAM_LOCK
*/
/* determining the start of the segment */
start = filter->segment.start;
if (priv->clock_base != -1 && position != -1) {
GstClockTime exttime, gap;
exttime = priv->clock_base;
gst_rtp_buffer_ext_timestamp (&exttime, rtptime);
gap = gst_util_uint64_scale_int (exttime - priv->clock_base,
filter->clock_rate, GST_SECOND);
/* account for lost packets */
if (position > gap) {
GST_DEBUG_OBJECT (filter,
"Found gap of %" GST_TIME_FORMAT ", adjusting start: %"
GST_TIME_FORMAT " = %" GST_TIME_FORMAT " - %" GST_TIME_FORMAT,
GST_TIME_ARGS (gap), GST_TIME_ARGS (position - gap),
GST_TIME_ARGS (position), GST_TIME_ARGS (gap));
start = position - gap;
}
}
/* determining the stop of the segment */
stop = filter->segment.stop;
if (priv->npt_stop != -1)
stop = start + (priv->npt_stop - priv->npt_start);
if (position == -1)
position = start;
running_time = gst_segment_to_running_time (&filter->segment,
GST_FORMAT_TIME, start);
gst_segment_init (&segment, GST_FORMAT_TIME);
segment.rate = priv->play_speed;
segment.applied_rate = priv->play_scale;
segment.start = start;
segment.stop = stop;
segment.time = priv->npt_start;
segment.position = position;
segment.base = running_time;
GST_DEBUG_OBJECT (filter, "Creating segment event %" GST_SEGMENT_FORMAT,
&segment);
event = gst_event_new_segment (&segment);
if (filter->priv->segment_seqnum != GST_SEQNUM_INVALID)
gst_event_set_seqnum (event, filter->priv->segment_seqnum);
return event;
}
static gboolean
foreach_metadata_drop (GstBuffer * buffer, GstMeta ** meta, gpointer user_data)
{
GType drop_api_type = (GType) user_data;
const GstMetaInfo *info = (*meta)->info;
if (info->api == drop_api_type)
*meta = NULL;
return TRUE;
}
static void
add_rtp_source_meta (GstBuffer * outbuf, GstBuffer * rtpbuf)
{
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
GstRTPSourceMeta *meta;
guint32 ssrc;
GType source_meta_api = gst_rtp_source_meta_api_get_type ();
if (!gst_rtp_buffer_map (rtpbuf, GST_MAP_READ, &rtp))
return;
ssrc = gst_rtp_buffer_get_ssrc (&rtp);
/* remove any pre-existing source-meta */
gst_buffer_foreach_meta (outbuf, foreach_metadata_drop,
(gpointer) source_meta_api);
meta = gst_buffer_add_rtp_source_meta (outbuf, &ssrc, NULL, 0);
if (meta != NULL) {
gint i;
gint csrc_count = gst_rtp_buffer_get_csrc_count (&rtp);
for (i = 0; i < csrc_count; i++) {
guint32 csrc = gst_rtp_buffer_get_csrc (&rtp, i);
gst_rtp_source_meta_append_csrc (meta, &csrc, 1);
}
}
gst_rtp_buffer_unmap (&rtp);
}
static void
gst_rtp_base_depayload_add_extension (GstRTPBaseDepayload * rtpbasepayload,
GstRTPHeaderExtension * ext)
{
g_return_if_fail (GST_IS_RTP_HEADER_EXTENSION (ext));
g_return_if_fail (gst_rtp_header_extension_get_id (ext) > 0);
/* XXX: check for duplicate ids? */
GST_OBJECT_LOCK (rtpbasepayload);
g_ptr_array_add (rtpbasepayload->priv->header_exts, gst_object_ref (ext));
GST_OBJECT_UNLOCK (rtpbasepayload);
}
static void
gst_rtp_base_depayload_clear_extensions (GstRTPBaseDepayload * rtpbasepayload)
{
GST_OBJECT_LOCK (rtpbasepayload);
g_ptr_array_set_size (rtpbasepayload->priv->header_exts, 0);
GST_OBJECT_UNLOCK (rtpbasepayload);
}
static gboolean
read_rtp_header_extensions (GstRTPBaseDepayload * depayload,
GstBuffer * input, GstBuffer * output)
{
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
guint16 bit_pattern;
guint8 *pdata;
guint wordlen;
gboolean needs_src_caps_update = FALSE;
if (!input) {
GST_DEBUG_OBJECT (depayload, "no input buffer");
return needs_src_caps_update;
}
if (!gst_rtp_buffer_map (input, GST_MAP_READ, &rtp)) {
GST_WARNING_OBJECT (depayload, "Failed to map buffer");
return needs_src_caps_update;
}
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 offset = 0;
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 (depayload, "unknown extension bit pattern 0x%02x%02x",
bit_pattern >> 8, bit_pattern & 0xff);
goto out;
}
while (TRUE) {
guint8 read_id, read_len;
GstRTPHeaderExtension *ext = NULL;
guint i;
if (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 + offset) >> 4;
read_len = (GST_READ_UINT8 (pdata + offset) & 0x0F) + 1;
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 + offset);
offset += 1;
if (read_id == 0)
/* padding */
continue;
read_len = GST_READ_UINT8 (pdata + offset);
offset += 1;
}
GST_TRACE_OBJECT (depayload, "found rtp header extension with id %u and "
"length %u", read_id, read_len);
/* Ignore extension headers where the size does not fit */
if (offset + read_len > bytelen) {
GST_WARNING_OBJECT (depayload, "Extension length extends past the "
"size of the extension data");
break;
}
GST_OBJECT_LOCK (depayload);
for (i = 0; i < depayload->priv->header_exts->len; i++) {
ext = g_ptr_array_index (depayload->priv->header_exts, i);
if (read_id == gst_rtp_header_extension_get_id (ext)) {
gst_object_ref (ext);
break;
}
ext = NULL;
}
if (ext) {
if (!gst_rtp_header_extension_read (ext, ext_flags, &pdata[offset],
read_len, output)) {
GST_WARNING_OBJECT (depayload, "RTP header extension (%s) could "
"not read payloaded data", GST_OBJECT_NAME (ext));
gst_object_unref (ext);
goto out;
}
if (gst_rtp_header_extension_wants_update_non_rtp_src_caps (ext)) {
needs_src_caps_update = TRUE;
}
gst_object_unref (ext);
}
GST_OBJECT_UNLOCK (depayload);
offset += read_len;
}
}
out:
gst_rtp_buffer_unmap (&rtp);
return needs_src_caps_update;
}
static gboolean
gst_rtp_base_depayload_set_headers (GstRTPBaseDepayload * depayload,
GstBuffer * buffer)
{
GstRTPBaseDepayloadPrivate *priv = depayload->priv;
GstClockTime pts, dts, duration;
pts = GST_BUFFER_PTS (buffer);
dts = GST_BUFFER_DTS (buffer);
duration = GST_BUFFER_DURATION (buffer);
/* apply last incoming timestamp and duration to outgoing buffer if
* not otherwise set. */
if (!GST_CLOCK_TIME_IS_VALID (pts))
GST_BUFFER_PTS (buffer) = priv->pts;
if (!GST_CLOCK_TIME_IS_VALID (dts))
GST_BUFFER_DTS (buffer) = priv->dts;
if (!GST_CLOCK_TIME_IS_VALID (duration))
GST_BUFFER_DURATION (buffer) = priv->duration;
if (G_UNLIKELY (depayload->priv->discont)) {
GST_LOG_OBJECT (depayload, "Marking DISCONT on output buffer");
GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
depayload->priv->discont = FALSE;
}
/* make sure we only set the timestamp on the first packet */
priv->pts = GST_CLOCK_TIME_NONE;
priv->dts = GST_CLOCK_TIME_NONE;
priv->duration = GST_CLOCK_TIME_NONE;
if (priv->input_buffer) {
if (priv->source_info)
add_rtp_source_meta (buffer, priv->input_buffer);
return read_rtp_header_extensions (depayload, priv->input_buffer, buffer);
}
return FALSE;
}
static GstFlowReturn
gst_rtp_base_depayload_finish_push (GstRTPBaseDepayload * filter,
gboolean is_list, gpointer obj)
{
/* if this is the first buffer send a NEWSEGMENT */
if (G_UNLIKELY (filter->priv->segment_event)) {
gst_pad_push_event (filter->srcpad, filter->priv->segment_event);
filter->priv->segment_event = NULL;
GST_DEBUG_OBJECT (filter, "Pushed newsegment event on this first buffer");
}
if (is_list) {
GstBufferList *blist = obj;
return gst_pad_push_list (filter->srcpad, blist);
} else {
GstBuffer *buf = obj;
return gst_pad_push (filter->srcpad, buf);
}
}
static gboolean
gst_rtp_base_depayload_set_src_caps_from_hdrext (GstRTPBaseDepayload * filter)
{
gboolean update_ok = TRUE;
GstCaps *src_caps = gst_pad_get_current_caps (filter->srcpad);
if (src_caps) {
GstCaps *new_caps;
gint i;
new_caps = gst_caps_copy (src_caps);
for (i = 0; i < filter->priv->header_exts->len; i++) {
GstRTPHeaderExtension *ext;
ext = g_ptr_array_index (filter->priv->header_exts, i);
update_ok =
gst_rtp_header_extension_update_non_rtp_src_caps (ext, new_caps);
if (!update_ok) {
GST_ELEMENT_ERROR (filter, STREAM, DECODE,
("RTP header extension (%s) could not update src caps",
GST_OBJECT_NAME (ext)), (NULL));
break;
}
}
if (G_UNLIKELY (update_ok && !gst_caps_is_equal (src_caps, new_caps))) {
gst_pad_set_caps (filter->srcpad, new_caps);
}
gst_caps_unref (src_caps);
gst_caps_unref (new_caps);
}
return update_ok;
}
static GstFlowReturn
gst_rtp_base_depayload_do_push (GstRTPBaseDepayload * filter, gboolean is_list,
gpointer obj)
{
GstFlowReturn res;
if (is_list) {
GstBufferList *blist = obj;
guint i;
guint first_not_pushed_idx = 0;
for (i = 0; i < gst_buffer_list_length (blist); ++i) {
GstBuffer *buf = gst_buffer_list_get_writable (blist, i);
if (G_UNLIKELY (gst_rtp_base_depayload_set_headers (filter, buf))) {
/* src caps have changed; push the buffers preceding the current one,
* then apply the new caps on the src pad */
guint j;
for (j = first_not_pushed_idx; j < i; ++j) {
res = gst_rtp_base_depayload_finish_push (filter, FALSE,
gst_buffer_ref (gst_buffer_list_get (blist, j)));
if (G_UNLIKELY (res != GST_FLOW_OK)) {
goto error_list;
}
}
first_not_pushed_idx = i;
if (!gst_rtp_base_depayload_set_src_caps_from_hdrext (filter)) {
res = GST_FLOW_ERROR;
goto error_list;
}
}
}
if (G_LIKELY (first_not_pushed_idx == 0)) {
res = gst_rtp_base_depayload_finish_push (filter, TRUE, blist);
blist = NULL;
} else {
for (i = first_not_pushed_idx; i < gst_buffer_list_length (blist); ++i) {
res = gst_rtp_base_depayload_finish_push (filter, FALSE,
gst_buffer_ref (gst_buffer_list_get (blist, i)));
if (G_UNLIKELY (res != GST_FLOW_OK)) {
break;
}
}
}
error_list:
gst_clear_buffer_list (&blist);
} else {
GstBuffer *buf = obj;
if (G_UNLIKELY (gst_rtp_base_depayload_set_headers (filter, buf))) {
if (!gst_rtp_base_depayload_set_src_caps_from_hdrext (filter)) {
res = GST_FLOW_ERROR;
goto error_buffer;
}
}
res = gst_rtp_base_depayload_finish_push (filter, FALSE, buf);
buf = NULL;
error_buffer:
gst_clear_buffer (&buf);
}
return res;
}
/**
* gst_rtp_base_depayload_push:
* @filter: a #GstRTPBaseDepayload
* @out_buf: (transfer full): a #GstBuffer
*
* Push @out_buf to the peer of @filter. This function takes ownership of
* @out_buf.
*
* This function will by default apply the last incoming timestamp on
* the outgoing buffer when it didn't have a timestamp already.
*
* Returns: a #GstFlowReturn.
*/
GstFlowReturn
gst_rtp_base_depayload_push (GstRTPBaseDepayload * filter, GstBuffer * out_buf)
{
GstFlowReturn res;
res = gst_rtp_base_depayload_do_push (filter, FALSE, out_buf);
if (res != GST_FLOW_OK)
filter->priv->process_flow_ret = res;
return res;
}
/**
* gst_rtp_base_depayload_push_list:
* @filter: a #GstRTPBaseDepayload
* @out_list: (transfer full): a #GstBufferList
*
* Push @out_list to the peer of @filter. This function takes ownership of
* @out_list.
*
* Returns: a #GstFlowReturn.
*/
GstFlowReturn
gst_rtp_base_depayload_push_list (GstRTPBaseDepayload * filter,
GstBufferList * out_list)
{
GstFlowReturn res;
res = gst_rtp_base_depayload_do_push (filter, TRUE, out_list);
if (res != GST_FLOW_OK)
filter->priv->process_flow_ret = res;
return res;
}
/* convert the PacketLost event from a jitterbuffer to a GAP event.
* subclasses can override this. */
static gboolean
gst_rtp_base_depayload_packet_lost (GstRTPBaseDepayload * filter,
GstEvent * event)
{
GstClockTime timestamp, duration;
GstEvent *sevent;
const GstStructure *s;
gboolean might_have_been_fec;
gboolean res = TRUE;
s = gst_event_get_structure (event);
/* first start by parsing the timestamp and duration */
timestamp = -1;
duration = -1;
if (!gst_structure_get_clock_time (s, "timestamp", &timestamp) ||
!gst_structure_get_clock_time (s, "duration", &duration)) {
GST_ERROR_OBJECT (filter,
"Packet loss event without timestamp or duration");
return FALSE;
}
sevent = gst_pad_get_sticky_event (filter->srcpad, GST_EVENT_SEGMENT, 0);
if (G_UNLIKELY (!sevent)) {
/* Typically happens if lost event arrives before first buffer */
GST_DEBUG_OBJECT (filter,
"Ignore packet loss because segment event missing");
return FALSE;
}
gst_event_unref (sevent);
if (!gst_structure_get_boolean (s, "might-have-been-fec",
&might_have_been_fec) || !might_have_been_fec) {
/* send GAP event */
sevent = gst_event_new_gap (timestamp, duration);
gst_event_set_gap_flags (sevent, GST_GAP_FLAG_MISSING_DATA);
res = gst_pad_push_event (filter->srcpad, sevent);
}
return res;
}
static GstStateChangeReturn
gst_rtp_base_depayload_change_state (GstElement * element,
GstStateChange transition)
{
GstRTPBaseDepayload *filter;
GstRTPBaseDepayloadPrivate *priv;
GstStateChangeReturn ret;
filter = GST_RTP_BASE_DEPAYLOAD (element);
priv = filter->priv;
switch (transition) {
case GST_STATE_CHANGE_NULL_TO_READY:
break;
case GST_STATE_CHANGE_READY_TO_PAUSED:
filter->need_newsegment = TRUE;
priv->npt_start = 0;
priv->npt_stop = -1;
priv->play_speed = 1.0;
priv->play_scale = 1.0;
priv->clock_base = -1;
priv->ref_ts = -1;
priv->onvif_mode = FALSE;
priv->next_seqnum = -1;
priv->negotiated = FALSE;
priv->discont = FALSE;
priv->segment_seqnum = GST_SEQNUM_INVALID;
break;
case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
break;
default:
break;
}
ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
switch (transition) {
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
break;
case GST_STATE_CHANGE_PAUSED_TO_READY:
gst_caps_replace (&priv->last_caps, NULL);
gst_event_replace (&priv->segment_event, NULL);
break;
case GST_STATE_CHANGE_READY_TO_NULL:
break;
default:
break;
}
return ret;
}
static GstStructure *
gst_rtp_base_depayload_create_stats (GstRTPBaseDepayload * depayload)
{
GstRTPBaseDepayloadPrivate *priv;
GstStructure *s;
GstClockTime pts = GST_CLOCK_TIME_NONE, dts = GST_CLOCK_TIME_NONE;
priv = depayload->priv;
GST_OBJECT_LOCK (depayload);
if (depayload->segment.format != GST_FORMAT_UNDEFINED) {
pts = gst_segment_to_running_time (&depayload->segment, GST_FORMAT_TIME,
priv->pts);
dts = gst_segment_to_running_time (&depayload->segment, GST_FORMAT_TIME,
priv->dts);
}
GST_OBJECT_UNLOCK (depayload);
s = gst_structure_new ("application/x-rtp-depayload-stats",
"clock_rate", G_TYPE_UINT, depayload->clock_rate,
"npt-start", G_TYPE_UINT64, priv->npt_start,
"npt-stop", G_TYPE_UINT64, priv->npt_stop,
"play-speed", G_TYPE_DOUBLE, priv->play_speed,
"play-scale", G_TYPE_DOUBLE, priv->play_scale,
"running-time-dts", G_TYPE_UINT64, dts,
"running-time-pts", G_TYPE_UINT64, pts,
"seqnum", G_TYPE_UINT, (guint) priv->last_seqnum,
"timestamp", G_TYPE_UINT, (guint) priv->last_rtptime, NULL);
return s;
}
static void
gst_rtp_base_depayload_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec)
{
GstRTPBaseDepayload *depayload;
GstRTPBaseDepayloadPrivate *priv;
depayload = GST_RTP_BASE_DEPAYLOAD (object);
priv = depayload->priv;
switch (prop_id) {
case PROP_SOURCE_INFO:
gst_rtp_base_depayload_set_source_info_enabled (depayload,
g_value_get_boolean (value));
break;
case PROP_MAX_REORDER:
priv->max_reorder = g_value_get_int (value);
break;
case PROP_AUTO_HEADER_EXTENSION:
priv->auto_hdr_ext = g_value_get_boolean (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static void
gst_rtp_base_depayload_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec)
{
GstRTPBaseDepayload *depayload;
GstRTPBaseDepayloadPrivate *priv;
depayload = GST_RTP_BASE_DEPAYLOAD (object);
priv = depayload->priv;
switch (prop_id) {
case PROP_STATS:
g_value_take_boxed (value,
gst_rtp_base_depayload_create_stats (depayload));
break;
case PROP_SOURCE_INFO:
g_value_set_boolean (value,
gst_rtp_base_depayload_is_source_info_enabled (depayload));
break;
case PROP_MAX_REORDER:
g_value_set_int (value, priv->max_reorder);
break;
case PROP_AUTO_HEADER_EXTENSION:
g_value_set_boolean (value, priv->auto_hdr_ext);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
/**
* gst_rtp_base_depayload_set_source_info_enabled:
* @depayload: a #GstRTPBaseDepayload
* @enable: whether to add meta about RTP sources to buffer
*
* Enable or disable adding #GstRTPSourceMeta to depayloaded buffers.
*
* Since: 1.16
**/
void
gst_rtp_base_depayload_set_source_info_enabled (GstRTPBaseDepayload * depayload,
gboolean enable)
{
depayload->priv->source_info = enable;
}
/**
* gst_rtp_base_depayload_is_source_info_enabled:
* @depayload: a #GstRTPBaseDepayload
*
* Queries whether #GstRTPSourceMeta will be added to depayloaded buffers.
*
* Returns: %TRUE if source-info is enabled.
*
* Since: 1.16
**/
gboolean
gst_rtp_base_depayload_is_source_info_enabled (GstRTPBaseDepayload * depayload)
{
return depayload->priv->source_info;
}
Reference in a new issue View git blame Copy permalink