gstreamer/gst/onvif/gstrtponviftimestamp.c
Tim-Philipp Müller 4d76070f4e rtponviftimestamp: fix state change function init/reset
When starting up we need to initialise things *before*
streaming starts, so before we chain up to the parent
class in the state change function. And when we shut
down the element, we need to reset things after streaming
has stopped, so after we chain up to the parent class
in the state change function.

Possibly related to memory leak in:
https://bugzilla.gnome.org/show_bug.cgi?id=794353
2018-03-27 16:51:30 +01:00

633 lines
18 KiB
C

/*
* gstrtponviftimestamp.h
*
* Copyright (C) 2014 Axis Communications AB
* Author: Guillaume Desmottes <guillaume.desmottes@collabora.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, see <http://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <gst/rtp/gstrtpbuffer.h>
#include "gstrtponviftimestamp.h"
#define GST_NTP_OFFSET_EVENT_NAME "GstNtpOffset"
#define DEFAULT_NTP_OFFSET GST_CLOCK_TIME_NONE
#define DEFAULT_CSEQ 0
#define DEFAULT_SET_E_BIT FALSE
GST_DEBUG_CATEGORY_STATIC (rtponviftimestamp_debug);
#define GST_CAT_DEFAULT (rtponviftimestamp_debug)
static GstFlowReturn gst_rtp_onvif_timestamp_chain (GstPad * pad,
GstObject * parent, GstBuffer * buf);
static GstFlowReturn gst_rtp_onvif_timestamp_chain_list (GstPad * pad,
GstObject * parent, GstBufferList * list);
static GstFlowReturn handle_and_push_buffer (GstRtpOnvifTimestamp * self,
GstBuffer * buf);
static GstFlowReturn handle_and_push_buffer_list (GstRtpOnvifTimestamp * self,
GstBufferList * list);
static GstStaticPadTemplate sink_template_factory =
GST_STATIC_PAD_TEMPLATE ("sink",
GST_PAD_SINK,
GST_PAD_ALWAYS,
GST_STATIC_CAPS ("application/x-rtp")
);
static GstStaticPadTemplate src_template_factory =
GST_STATIC_PAD_TEMPLATE ("src",
GST_PAD_SRC,
GST_PAD_ALWAYS,
GST_STATIC_CAPS ("application/x-rtp")
);
enum
{
PROP_0,
PROP_NTP_OFFSET,
PROP_CSEQ,
PROP_SET_E_BIT,
};
/*static guint gst_rtp_onvif_timestamp_signals[LAST_SIGNAL] = { 0 }; */
G_DEFINE_TYPE (GstRtpOnvifTimestamp, gst_rtp_onvif_timestamp, GST_TYPE_ELEMENT);
static void
gst_rtp_onvif_timestamp_get_property (GObject * object,
guint prop_id, GValue * value, GParamSpec * pspec)
{
GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (object);
switch (prop_id) {
case PROP_NTP_OFFSET:
g_value_set_uint64 (value, self->prop_ntp_offset);
break;
case PROP_CSEQ:
g_value_set_uint (value, self->prop_cseq);
break;
case PROP_SET_E_BIT:
g_value_set_boolean (value, self->prop_set_e_bit);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static void
gst_rtp_onvif_timestamp_set_property (GObject * object,
guint prop_id, const GValue * value, GParamSpec * pspec)
{
GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (object);
switch (prop_id) {
case PROP_NTP_OFFSET:
self->prop_ntp_offset = g_value_get_uint64 (value);
break;
case PROP_CSEQ:
self->prop_cseq = g_value_get_uint (value);
break;
case PROP_SET_E_BIT:
self->prop_set_e_bit = g_value_get_boolean (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
/* send cached buffer or list, and events, if present */
static GstFlowReturn
send_cached_buffer_and_events (GstRtpOnvifTimestamp * self)
{
GstFlowReturn ret = GST_FLOW_OK;
g_assert (!(self->buffer && self->list));
if (self->buffer) {
GST_DEBUG_OBJECT (self, "pushing %" GST_PTR_FORMAT, self->buffer);
ret = handle_and_push_buffer (self, self->buffer);
self->buffer = NULL;
}
if (self->list) {
GST_DEBUG_OBJECT (self, "pushing %" GST_PTR_FORMAT, self->list);
ret = handle_and_push_buffer_list (self, self->list);
self->list = NULL;
}
if (ret != GST_FLOW_OK)
goto out;
while (!g_queue_is_empty (self->event_queue)) {
GstEvent *event;
event = GST_EVENT_CAST (g_queue_pop_head (self->event_queue));
GST_LOG_OBJECT (self->sinkpad, "sending %" GST_PTR_FORMAT, event);
(void) gst_pad_send_event (self->sinkpad, event);
}
out:
return ret;
}
static void
purge_cached_buffer_and_events (GstRtpOnvifTimestamp * self)
{
g_assert (!(self->buffer && self->list));
if (self->buffer) {
GST_DEBUG_OBJECT (self, "purging %" GST_PTR_FORMAT, self->buffer);
gst_buffer_unref (self->buffer);
self->buffer = NULL;
}
if (self->list) {
GST_DEBUG_OBJECT (self, "purging %" GST_PTR_FORMAT, self->list);
gst_buffer_list_unref (self->list);
self->list = NULL;
}
while (!g_queue_is_empty (self->event_queue)) {
GstEvent *event;
event = GST_EVENT_CAST (g_queue_pop_head (self->event_queue));
gst_event_unref (event);
}
}
static GstStateChangeReturn
gst_rtp_onvif_timestamp_change_state (GstElement * element,
GstStateChange transition)
{
GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (element);
GstStateChangeReturn ret;
switch (transition) {
case GST_STATE_CHANGE_READY_TO_PAUSED:
self->ntp_offset = self->prop_ntp_offset;
GST_DEBUG_OBJECT (self, "ntp-offset: %" GST_TIME_FORMAT,
GST_TIME_ARGS (self->ntp_offset));
self->set_d_bit = TRUE;
self->set_e_bit = FALSE;
break;
default:
break;
}
ret = GST_ELEMENT_CLASS (gst_rtp_onvif_timestamp_parent_class)->change_state
(element, transition);
if (ret == GST_STATE_CHANGE_FAILURE)
return ret;
switch (transition) {
case GST_STATE_CHANGE_PAUSED_TO_READY:
purge_cached_buffer_and_events (self);
gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
break;
default:
break;
}
return ret;
}
static void
gst_rtp_onvif_timestamp_finalize (GObject * object)
{
GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (object);
g_queue_free (self->event_queue);
G_OBJECT_CLASS (gst_rtp_onvif_timestamp_parent_class)->finalize (object);
}
static void
gst_rtp_onvif_timestamp_class_init (GstRtpOnvifTimestampClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *gstelement_class;
gobject_class = G_OBJECT_CLASS (klass);
gstelement_class = GST_ELEMENT_CLASS (klass);
gobject_class->get_property = gst_rtp_onvif_timestamp_get_property;
gobject_class->set_property = gst_rtp_onvif_timestamp_set_property;
gobject_class->finalize = gst_rtp_onvif_timestamp_finalize;
g_object_class_install_property (gobject_class, PROP_NTP_OFFSET,
g_param_spec_uint64 ("ntp-offset", "NTP offset",
"Offset between the pipeline running time and the absolute UTC time, "
"in nano-seconds since 1900 (-1 for automatic computation)",
0, G_MAXUINT64,
DEFAULT_NTP_OFFSET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_CSEQ,
g_param_spec_uint ("cseq", "CSeq",
"The RTSP CSeq which initiated the playback",
0, G_MAXUINT32,
DEFAULT_CSEQ, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_SET_E_BIT,
g_param_spec_boolean ("set-e-bit", "Set 'E' bit",
"If the element should set the 'E' bit as defined in the ONVIF RTP "
"extension. This increases latency by one packet",
DEFAULT_SET_E_BIT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
/* register pads */
gst_element_class_add_static_pad_template (gstelement_class,
&sink_template_factory);
gst_element_class_add_static_pad_template (gstelement_class,
&src_template_factory);
gst_element_class_set_static_metadata (gstelement_class,
"ONVIF NTP timestamps RTP extension", "Effect/RTP",
"Add absolute timestamps and flags of recorded data in a playback "
"session", "Guillaume Desmottes <guillaume.desmottes@collabora.com>");
gstelement_class->change_state =
GST_DEBUG_FUNCPTR (gst_rtp_onvif_timestamp_change_state);
GST_DEBUG_CATEGORY_INIT (rtponviftimestamp_debug, "rtponviftimestamp",
0, "ONVIF NTP timestamps RTP extension");
}
static gboolean
parse_event_ntp_offset (GstRtpOnvifTimestamp * self, GstEvent * event,
GstClockTime * offset, gboolean * discont)
{
const GstStructure *structure = gst_event_get_structure (event);
GstClockTime event_offset;
gboolean event_discont;
if (!gst_structure_get_clock_time (structure, "ntp-offset", &event_offset)) {
GST_ERROR_OBJECT (self, "no ntp-offset in %" GST_PTR_FORMAT, event);
return FALSE;
}
if (!gst_structure_get_boolean (structure, "discont", &event_discont)) {
GST_ERROR_OBJECT (self, "no discontinue in %" GST_PTR_FORMAT, event);
return FALSE;
}
if (offset)
*offset = event_offset;
if (discont)
*discont = event_discont;
return TRUE;
}
static gboolean
gst_rtp_onvif_timestamp_sink_event (GstPad * pad, GstObject * parent,
GstEvent * event)
{
GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (parent);
gboolean drop = FALSE;
gboolean ret = TRUE;
GST_DEBUG_OBJECT (pad, "handling event %s", GST_EVENT_TYPE_NAME (event));
/* handle serialized events, which, should not be enqueued */
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_CUSTOM_DOWNSTREAM:
/* if the "set-e-bit" property is set, an offset event might mark the
* stream as discontinued. We need to check if the currently cached buffer
* needs the e-bit before it's pushed */
if (self->buffer != NULL && self->prop_set_e_bit &&
gst_event_has_name (event, GST_NTP_OFFSET_EVENT_NAME)) {
gboolean discont;
if (parse_event_ntp_offset (self, event, NULL, &discont)) {
GST_DEBUG_OBJECT (self, "stream %s discontinued",
(discont ? "is" : "is not"));
self->set_e_bit = discont;
} else {
drop = TRUE;
ret = FALSE;
goto out;
}
}
break;
case GST_EVENT_EOS:
{
GstFlowReturn res;
/* Push pending buffers, if any */
self->set_e_bit = TRUE;
res = send_cached_buffer_and_events (self);
if (res != GST_FLOW_OK) {
drop = TRUE;
ret = FALSE;
goto out;
}
break;
}
case GST_EVENT_FLUSH_STOP:
purge_cached_buffer_and_events (self);
self->set_d_bit = TRUE;
self->set_e_bit = FALSE;
gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
break;
default:
break;
}
/* enqueue serialized events if there is a cached buffer */
if (GST_EVENT_IS_SERIALIZED (event) && (self->buffer || self->list)) {
GST_DEBUG ("enqueueing serialized event");
g_queue_push_tail (self->event_queue, event);
event = NULL;
goto out;
}
/* handle rest of the events */
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_CUSTOM_DOWNSTREAM:
/* update the ntp-offset after any cached buffer/buffer list has been
* pushed. the d-bit of the next buffer/buffer list should be set if
* the stream is discontinued */
if (gst_event_has_name (event, GST_NTP_OFFSET_EVENT_NAME)) {
GstClockTime offset;
gboolean discont;
if (parse_event_ntp_offset (self, event, &offset, &discont)) {
GST_DEBUG_OBJECT (self, "new ntp-offset: %" GST_TIME_FORMAT
", stream %s discontinued", GST_TIME_ARGS (offset),
(discont ? "is" : "is not"));
self->ntp_offset = offset;
self->set_d_bit = discont;
} else {
ret = FALSE;
}
drop = TRUE;
}
break;
case GST_EVENT_SEGMENT:
gst_event_copy_segment (event, &self->segment);
break;
default:
break;
}
out:
if (drop)
gst_event_unref (event);
else if (event)
ret = gst_pad_event_default (pad, parent, event);
return ret;
}
static void
gst_rtp_onvif_timestamp_init (GstRtpOnvifTimestamp * self)
{
self->sinkpad =
gst_pad_new_from_static_template (&sink_template_factory, "sink");
gst_pad_set_chain_function (self->sinkpad, gst_rtp_onvif_timestamp_chain);
gst_pad_set_chain_list_function (self->sinkpad,
gst_rtp_onvif_timestamp_chain_list);
gst_pad_set_event_function (self->sinkpad,
gst_rtp_onvif_timestamp_sink_event);
gst_element_add_pad (GST_ELEMENT (self), self->sinkpad);
GST_PAD_SET_PROXY_CAPS (self->sinkpad);
GST_PAD_SET_PROXY_ALLOCATION (self->sinkpad);
self->srcpad =
gst_pad_new_from_static_template (&src_template_factory, "src");
gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
self->prop_ntp_offset = DEFAULT_NTP_OFFSET;
self->prop_set_e_bit = DEFAULT_SET_E_BIT;
gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
self->event_queue = g_queue_new ();
self->buffer = NULL;
self->list = NULL;
}
#define EXTENSION_ID 0xABAC
#define EXTENSION_SIZE 3
static gboolean
handle_buffer (GstRtpOnvifTimestamp * self, GstBuffer * buf)
{
GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
guint8 *data;
guint16 bits;
guint wordlen;
guint64 time;
guint8 field = 0;
if (!GST_CLOCK_TIME_IS_VALID (self->ntp_offset)) {
GstClock *clock = gst_element_get_clock (GST_ELEMENT (self));
if (clock) {
GstClockTime clock_time = gst_clock_get_time (clock);
guint64 real_time = g_get_real_time ();
GstClockTime running_time = clock_time -
gst_element_get_base_time (GST_ELEMENT (self));
/* convert microseconds to nanoseconds */
real_time *= 1000;
/* add constant to convert from 1970 based time to 1900 based time */
real_time += (G_GUINT64_CONSTANT (2208988800) * GST_SECOND);
self->ntp_offset = real_time - running_time;
GST_DEBUG_OBJECT (self, "new ntp-offset: %" GST_TIME_FORMAT,
GST_TIME_ARGS (self->ntp_offset));
gst_object_unref (clock);
} else {
GST_ELEMENT_ERROR (self, STREAM, FAILED, ("No ntp-offset present"),
("Can not guess ntp-offset with no clock."));
/* Received a buffer in PAUSED, so we can't guess the match
* between the running time and the NTP clock yet.
*/
return FALSE;
}
}
if (self->segment.format != GST_FORMAT_TIME) {
GST_ELEMENT_ERROR (self, STREAM, FAILED,
("did not receive a time segment yet"), (NULL));
return FALSE;
}
if (!gst_rtp_buffer_map (buf, GST_MAP_READWRITE, &rtp)) {
GST_ELEMENT_ERROR (self, STREAM, FAILED,
("Failed to map RTP buffer"), (NULL));
return FALSE;
}
if (!gst_rtp_buffer_set_extension_data (&rtp, EXTENSION_ID, EXTENSION_SIZE)) {
GST_ELEMENT_ERROR (self, STREAM, FAILED, ("Failed to set extension data"),
(NULL));
gst_rtp_buffer_unmap (&rtp);
return FALSE;
}
if (!gst_rtp_buffer_get_extension_data (&rtp, &bits, (gpointer) & data,
&wordlen)) {
GST_ELEMENT_ERROR (self, STREAM, FAILED, ("Failed to get extension data"),
(NULL));
gst_rtp_buffer_unmap (&rtp);
return FALSE;
}
/* NTP timestamp */
if (GST_BUFFER_DTS_IS_VALID (buf)) {
time = gst_segment_to_stream_time (&self->segment, GST_FORMAT_TIME,
GST_BUFFER_DTS (buf));
} else if (GST_BUFFER_PTS_IS_VALID (buf)) {
time = gst_segment_to_stream_time (&self->segment, GST_FORMAT_TIME,
GST_BUFFER_PTS (buf));
} else {
GST_ERROR_OBJECT (self,
"Buffer doesn't contain any valid DTS or PTS timestamp");
goto done;
}
if (time == GST_CLOCK_TIME_NONE) {
GST_ERROR_OBJECT (self, "Failed to get stream time");
goto done;
}
/* add the offset (in seconds) */
time += self->ntp_offset;
/* convert to NTP time. upper 32 bits should contain the seconds
* and the lower 32 bits, the fractions of a second. */
time = gst_util_uint64_scale (time, (G_GINT64_CONSTANT (1) << 32),
GST_SECOND);
GST_DEBUG_OBJECT (self, "timestamp: %" G_GUINT64_FORMAT, time);
GST_WRITE_UINT64_BE (data, time);
/* The next byte is composed of: C E D mbz (5 bits) */
/* Set C if the buffer does *not* have the DELTA_UNIT flag as it means
* that's a key frame (or 'clean point'). */
if (!GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT)) {
GST_DEBUG_OBJECT (self, "set C flag");
field |= (1 << 7);
}
/* Set E if the next buffer has DISCONT */
if (self->set_e_bit) {
GST_DEBUG_OBJECT (self, "set E flag");
field |= (1 << 6);
self->set_e_bit = FALSE;
}
/* Set D if the buffer has the DISCONT flag */
if (self->set_d_bit) {
GST_DEBUG_OBJECT (self, "set D flag");
field |= (1 << 5);
self->set_d_bit = FALSE;
}
GST_WRITE_UINT8 (data + 8, field);
/* CSeq (low-order byte) */
GST_WRITE_UINT8 (data + 9, (guchar) self->prop_cseq);
memset (data + 10, 0, 3);
done:
gst_rtp_buffer_unmap (&rtp);
return TRUE;
}
/* @buf: (transfer full) */
static GstFlowReturn
handle_and_push_buffer (GstRtpOnvifTimestamp * self, GstBuffer * buf)
{
if (!handle_buffer (self, buf)) {
gst_buffer_unref (buf);
return GST_FLOW_ERROR;
}
return gst_pad_push (self->srcpad, buf);
}
static GstFlowReturn
gst_rtp_onvif_timestamp_chain (GstPad * pad, GstObject * parent,
GstBuffer * buf)
{
GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (parent);
GstFlowReturn result = GST_FLOW_OK;
if (!self->prop_set_e_bit) {
/* Modify and push this buffer right away */
return handle_and_push_buffer (self, buf);
}
/* send any previously cached item(s), this leaves an empty queue */
result = send_cached_buffer_and_events (self);
/* enqueue the new item, as the only item in the queue */
self->buffer = buf;
return result;
}
/* @buf: (transfer full) */
static GstFlowReturn
handle_and_push_buffer_list (GstRtpOnvifTimestamp * self, GstBufferList * list)
{
GstBuffer *buf;
/* Set the extension on the *first* buffer */
buf = gst_buffer_list_get (list, 0);
if (!handle_buffer (self, buf)) {
gst_buffer_list_unref (list);
return GST_FLOW_ERROR;
}
return gst_pad_push_list (self->srcpad, list);
}
/* gst_pad_chain_list_default() refs the buffer when passing it to the chain
* function, making it not writable. We implement our own chain_list function
* to avoid having to copy each buffer. */
static GstFlowReturn
gst_rtp_onvif_timestamp_chain_list (GstPad * pad, GstObject * parent,
GstBufferList * list)
{
GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (parent);
GstFlowReturn result = GST_FLOW_OK;
if (!self->prop_set_e_bit) {
return handle_and_push_buffer_list (self, list);
}
/* send any previously cached item(s), this leaves an empty queue */
result = send_cached_buffer_and_events (self);
/* enqueue the new item, as the only item in the queue */
self->list = list;
return result;
}