/* GStreamer
* Copyright (C) 2008 Wim Taymans <wim.taymans at gmail.com>
+ * Copyright (C) 2015 Centricular Ltd
+ * Author: Sebastian Dröge <sebastian@centricular.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
#include <gst/app/gstappsrc.h>
#include <gst/app/gstappsink.h>
+#include <gst/rtp/gstrtpbuffer.h>
+
#include "rtsp-stream.h"
#define GST_RTSP_STREAM_GET_PRIVATE(obj) \
(G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTSP_STREAM, GstRTSPStreamPrivate))
+typedef struct
+{
+ GstRTSPStreamTransport *transport;
+
+ /* RTP and RTCP source */
+ GstElement *udpsrc[2];
+ GstPad *selpad[2];
+} GstRTSPMulticastTransportSource;
+
struct _GstRTSPStreamPrivate
{
GMutex lock;
guint idx;
- GstPad *srcpad;
+ /* Only one pad is ever set */
+ GstPad *srcpad, *sinkpad;
GstElement *payloader;
guint buffer_size;
gboolean is_joined;
/* pads on the rtpbin */
GstPad *send_rtp_sink;
+ GstPad *recv_rtp_src;
GstPad *recv_sink[2];
GstPad *send_src[2];
* sockets */
GstElement *udpsrc_v6[2];
+ GstElement *udpqueue[2];
GstElement *udpsink[2];
/* for TCP transport */
GstElement *appsrc[2];
+ GstClockTime appsrc_base_time[2];
GstElement *appqueue[2];
GstElement *appsink[2];
GstElement *tee[2];
GstElement *funnel[2];
+ /* retransmission */
+ GstElement *rtxsend;
+ guint rtx_pt;
+ GstClockTime rtx_time;
+
/* server ports for sending/receiving over ipv4 */
GstRTSPRange server_port_v4;
GstRTSPAddress *server_addr_v4;
guint n_active;
GList *transports;
guint transports_cookie;
- GList *tr_cache;
- guint tr_cache_cookie;
+ GList *tr_cache_rtp;
+ GList *tr_cache_rtcp;
+ guint tr_cache_cookie_rtp;
+ guint tr_cache_cookie_rtcp;
+
+
+ /* UDP sources for UDP multicast transports */
+ GList *transport_sources;
gint dscp_qos;
/* stream blocking */
gulong blocked_id;
gboolean blocking;
+
+ /* pt->caps map for RECORD streams */
+ GHashTable *ptmap;
};
#define DEFAULT_CONTROL NULL
priv->keys = g_hash_table_new_full (g_direct_hash, g_direct_equal,
NULL, (GDestroyNotify) gst_caps_unref);
+ priv->ptmap = g_hash_table_new_full (NULL, NULL, NULL,
+ (GDestroyNotify) gst_caps_unref);
}
static void
gst_rtsp_address_free (priv->server_addr_v6);
if (priv->pool)
g_object_unref (priv->pool);
+ if (priv->rtxsend)
+ g_object_unref (priv->rtxsend);
+
gst_object_unref (priv->payloader);
- gst_object_unref (priv->srcpad);
+ if (priv->srcpad)
+ gst_object_unref (priv->srcpad);
+ if (priv->sinkpad)
+ gst_object_unref (priv->sinkpad);
g_free (priv->control);
g_mutex_clear (&priv->lock);
g_hash_table_unref (priv->keys);
+ g_hash_table_destroy (priv->ptmap);
G_OBJECT_CLASS (gst_rtsp_stream_parent_class)->finalize (obj);
}
/**
* gst_rtsp_stream_new:
* @idx: an index
- * @srcpad: a #GstPad
+ * @pad: a #GstPad
* @payloader: a #GstElement
*
* Create a new media stream with index @idx that handles RTP data on
- * @srcpad and has a payloader element @payloader.
+ * @pad and has a payloader element @payloader if @pad is a source pad
+ * or a depayloader element @payloader if @pad is a sink pad.
*
* Returns: (transfer full): a new #GstRTSPStream
*/
GstRTSPStream *
-gst_rtsp_stream_new (guint idx, GstElement * payloader, GstPad * srcpad)
+gst_rtsp_stream_new (guint idx, GstElement * payloader, GstPad * pad)
{
GstRTSPStreamPrivate *priv;
GstRTSPStream *stream;
g_return_val_if_fail (GST_IS_ELEMENT (payloader), NULL);
- g_return_val_if_fail (GST_IS_PAD (srcpad), NULL);
- g_return_val_if_fail (GST_PAD_IS_SRC (srcpad), NULL);
+ g_return_val_if_fail (GST_IS_PAD (pad), NULL);
stream = g_object_new (GST_TYPE_RTSP_STREAM, NULL);
priv = stream->priv;
priv->idx = idx;
priv->payloader = gst_object_ref (payloader);
- priv->srcpad = gst_object_ref (srcpad);
+ if (GST_PAD_IS_SRC (pad))
+ priv->srcpad = gst_object_ref (pad);
+ else
+ priv->sinkpad = gst_object_ref (pad);
return stream;
}
{
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
+ if (!stream->priv->srcpad)
+ return NULL;
+
return gst_object_ref (stream->priv->srcpad);
}
/**
+ * gst_rtsp_stream_get_sinkpad:
+ * @stream: a #GstRTSPStream
+ *
+ * Get the sinkpad associated with @stream.
+ *
+ * Returns: (transfer full): the sinkpad. Unref after usage.
+ */
+GstPad *
+gst_rtsp_stream_get_sinkpad (GstRTSPStream * stream)
+{
+ g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
+
+ if (!stream->priv->sinkpad)
+ return NULL;
+
+ return gst_object_ref (stream->priv->sinkpad);
+}
+
+/**
* gst_rtsp_stream_get_control:
* @stream: a #GstRTSPStream
*
}
static gboolean
-alloc_ports_one_family (GstRTSPAddressPool * pool, gint buffer_size,
- GSocketFamily family, GstElement * udpsrc_out[2],
+alloc_ports_one_family (GstRTSPStream * stream, GstRTSPAddressPool * pool,
+ gint buffer_size, GSocketFamily family, GstElement * udpsrc_out[2],
GstElement * udpsink_out[2], GstRTSPRange * server_port_out,
GstRTSPAddress ** server_addr_out)
{
+ GstRTSPStreamPrivate *priv = stream->priv;
GstStateChangeReturn ret;
GstElement *udpsrc0, *udpsrc1;
GstElement *udpsink0, *udpsink1;
g_object_set (G_OBJECT (udpsink1), "close-socket", FALSE, NULL);
g_object_set (G_OBJECT (udpsink1), multisink_socket, rtcp_socket, NULL);
g_object_set (G_OBJECT (udpsink1), "sync", FALSE, NULL);
+ /* Needs to be async for RECORD streams, otherwise we will never go to
+ * PLAYING because the sinks will wait for data while the udpsrc can't
+ * provide data with timestamps in PAUSED. */
+ if (priv->sinkpad)
+ g_object_set (G_OBJECT (udpsink0), "async", FALSE, NULL);
g_object_set (G_OBJECT (udpsink1), "async", FALSE, NULL);
g_object_set (G_OBJECT (udpsink0), "auto-multicast", FALSE, NULL);
g_object_set (G_OBJECT (udpsink0), "loop", FALSE, NULL);
udpsrc_out[1] = udpsrc1;
udpsink_out[0] = udpsink0;
udpsink_out[1] = udpsink1;
+
server_port_out->min = rtpport;
server_port_out->max = rtcpport;
{
GstRTSPStreamPrivate *priv = stream->priv;
- priv->have_ipv4 = alloc_ports_one_family (priv->pool, priv->buffer_size,
+ priv->have_ipv4 =
+ alloc_ports_one_family (stream, priv->pool, priv->buffer_size,
G_SOCKET_FAMILY_IPV4, priv->udpsrc_v4, priv->udpsink,
&priv->server_port_v4, &priv->server_addr_v4);
- priv->have_ipv6 = alloc_ports_one_family (priv->pool, priv->buffer_size,
+ priv->have_ipv6 =
+ alloc_ports_one_family (stream, priv->pool, priv->buffer_size,
G_SOCKET_FAMILY_IPV6, priv->udpsrc_v6, priv->udpsink,
&priv->server_port_v6, &priv->server_addr_v6);
g_mutex_unlock (&priv->lock);
}
+/**
+ * gst_rtsp_stream_set_retransmission_time:
+ * @stream: a #GstRTSPStream
+ * @time: a #GstClockTime
+ *
+ * Set the amount of time to store retransmission packets.
+ */
+void
+gst_rtsp_stream_set_retransmission_time (GstRTSPStream * stream,
+ GstClockTime time)
+{
+ GST_DEBUG_OBJECT (stream, "set retransmission time %" G_GUINT64_FORMAT, time);
+
+ g_mutex_lock (&stream->priv->lock);
+ stream->priv->rtx_time = time;
+ if (stream->priv->rtxsend)
+ g_object_set (stream->priv->rtxsend, "max-size-time",
+ GST_TIME_AS_MSECONDS (time), NULL);
+ g_mutex_unlock (&stream->priv->lock);
+}
+
+/**
+ * gst_rtsp_stream_get_retransmission_time:
+ * @stream: a #GstRTSPStream
+ *
+ * Get the amount of time to store retransmission data.
+ *
+ * Returns: the amount of time to store retransmission data.
+ */
+GstClockTime
+gst_rtsp_stream_get_retransmission_time (GstRTSPStream * stream)
+{
+ GstClockTime ret;
+
+ g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), 0);
+
+ g_mutex_lock (&stream->priv->lock);
+ ret = stream->priv->rtx_time;
+ g_mutex_unlock (&stream->priv->lock);
+
+ return ret;
+}
+
+/**
+ * gst_rtsp_stream_set_retransmission_pt:
+ * @stream: a #GstRTSPStream
+ * @rtx_pt: a #guint
+ *
+ * Set the payload type (pt) for retransmission of this stream.
+ */
+void
+gst_rtsp_stream_set_retransmission_pt (GstRTSPStream * stream, guint rtx_pt)
+{
+ g_return_if_fail (GST_IS_RTSP_STREAM (stream));
+
+ GST_DEBUG_OBJECT (stream, "set retransmission pt %u", rtx_pt);
+
+ g_mutex_lock (&stream->priv->lock);
+ stream->priv->rtx_pt = rtx_pt;
+ if (stream->priv->rtxsend) {
+ guint pt = gst_rtsp_stream_get_pt (stream);
+ gchar *pt_s = g_strdup_printf ("%d", pt);
+ GstStructure *rtx_pt_map = gst_structure_new ("application/x-rtp-pt-map",
+ pt_s, G_TYPE_UINT, rtx_pt, NULL);
+ g_object_set (stream->priv->rtxsend, "payload-type-map", rtx_pt_map, NULL);
+ g_free (pt_s);
+ gst_structure_free (rtx_pt_map);
+ }
+ g_mutex_unlock (&stream->priv->lock);
+}
+
+/**
+ * gst_rtsp_stream_get_retransmission_pt:
+ * @stream: a #GstRTSPStream
+ *
+ * Get the payload-type used for retransmission of this stream
+ *
+ * Returns: The retransmission PT.
+ */
+guint
+gst_rtsp_stream_get_retransmission_pt (GstRTSPStream * stream)
+{
+ guint rtx_pt;
+
+ g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), 0);
+
+ g_mutex_lock (&stream->priv->lock);
+ rtx_pt = stream->priv->rtx_pt;
+ g_mutex_unlock (&stream->priv->lock);
+
+ return rtx_pt;
+}
+
+/**
+ * gst_rtsp_stream_set_buffer_size:
+ * @stream: a #GstRTSPStream
+ * @size: the buffer size
+ *
+ * Set the size of the UDP transmission buffer (in bytes)
+ * Needs to be set before the stream is joined to a bin.
+ *
+ * Since: 1.6
+ */
+void
+gst_rtsp_stream_set_buffer_size (GstRTSPStream * stream, guint size)
+{
+ g_mutex_lock (&stream->priv->lock);
+ stream->priv->buffer_size = size;
+ g_mutex_unlock (&stream->priv->lock);
+}
+
+/**
+ * gst_rtsp_stream_get_buffer_size:
+ * @stream: a #GstRTSPStream
+ *
+ * Get the size of the UDP transmission buffer (in bytes)
+ *
+ * Returns: the size of the UDP TX buffer
+ *
+ * Since: 1.6
+ */
+guint
+gst_rtsp_stream_get_buffer_size (GstRTSPStream * stream)
+{
+ guint buffer_size;
+
+ g_mutex_lock (&stream->priv->lock);
+ buffer_size = stream->priv->buffer_size;
+ g_mutex_unlock (&stream->priv->lock);
+
+ return buffer_size;
+}
+
/* executed from streaming thread */
static void
caps_notify (GstPad * pad, GParamSpec * unused, GstRTSPStream * stream)
}
static void
-clear_tr_cache (GstRTSPStreamPrivate * priv)
+on_new_sender_ssrc (GObject * session, GObject * source, GstRTSPStream * stream)
+{
+ GST_INFO ("%p: new sender source %p", stream, source);
+#ifndef DUMP_STATS
+ {
+ GstStructure *stats;
+ g_object_get (source, "stats", &stats, NULL);
+ if (stats) {
+ dump_structure (stats);
+ gst_structure_free (stats);
+ }
+ }
+#endif
+}
+
+static void
+on_sender_ssrc_active (GObject * session, GObject * source,
+ GstRTSPStream * stream)
{
- g_list_foreach (priv->tr_cache, (GFunc) g_object_unref, NULL);
- g_list_free (priv->tr_cache);
- priv->tr_cache = NULL;
+#ifndef DUMP_STATS
+ {
+ GstStructure *stats;
+ g_object_get (source, "stats", &stats, NULL);
+ if (stats) {
+ dump_structure (stats);
+ gst_structure_free (stats);
+ }
+ }
+#endif
+}
+
+static void
+clear_tr_cache (GstRTSPStreamPrivate * priv, gboolean is_rtp)
+{
+ if (is_rtp) {
+ g_list_foreach (priv->tr_cache_rtp, (GFunc) g_object_unref, NULL);
+ g_list_free (priv->tr_cache_rtp);
+ priv->tr_cache_rtp = NULL;
+ } else {
+ g_list_foreach (priv->tr_cache_rtcp, (GFunc) g_object_unref, NULL);
+ g_list_free (priv->tr_cache_rtcp);
+ priv->tr_cache_rtcp = NULL;
+ }
}
static GstFlowReturn
is_rtp = GST_ELEMENT_CAST (sink) == priv->appsink[0];
g_mutex_lock (&priv->lock);
- if (priv->tr_cache_cookie != priv->transports_cookie) {
- clear_tr_cache (priv);
- for (walk = priv->transports; walk; walk = g_list_next (walk)) {
- GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
- priv->tr_cache = g_list_prepend (priv->tr_cache, g_object_ref (tr));
+ if (is_rtp) {
+ if (priv->tr_cache_cookie_rtp != priv->transports_cookie) {
+ clear_tr_cache (priv, is_rtp);
+ for (walk = priv->transports; walk; walk = g_list_next (walk)) {
+ GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
+ priv->tr_cache_rtp =
+ g_list_prepend (priv->tr_cache_rtp, g_object_ref (tr));
+ }
+ priv->tr_cache_cookie_rtp = priv->transports_cookie;
+ }
+ } else {
+ if (priv->tr_cache_cookie_rtcp != priv->transports_cookie) {
+ clear_tr_cache (priv, is_rtp);
+ for (walk = priv->transports; walk; walk = g_list_next (walk)) {
+ GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
+ priv->tr_cache_rtcp =
+ g_list_prepend (priv->tr_cache_rtcp, g_object_ref (tr));
+ }
+ priv->tr_cache_cookie_rtcp = priv->transports_cookie;
}
- priv->tr_cache_cookie = priv->transports_cookie;
}
g_mutex_unlock (&priv->lock);
- for (walk = priv->tr_cache; walk; walk = g_list_next (walk)) {
- GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
-
- if (is_rtp) {
+ if (is_rtp) {
+ for (walk = priv->tr_cache_rtp; walk; walk = g_list_next (walk)) {
+ GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
gst_rtsp_stream_transport_send_rtp (tr, buffer);
- } else {
+ }
+ } else {
+ for (walk = priv->tr_cache_rtcp; walk; walk = g_list_next (walk)) {
+ GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
gst_rtsp_stream_transport_send_rtcp (tr, buffer);
}
}
}
static GstElement *
-request_rtcp_decoder (GstElement * rtpbin, guint session,
+request_rtp_rtcp_decoder (GstElement * rtpbin, guint session,
GstRTSPStream * stream)
{
GstRTSPStreamPrivate *priv = stream->priv;
}
/**
+ * gst_rtsp_stream_request_aux_sender:
+ * @stream: a #GstRTSPStream
+ * @sessid: the session id
+ *
+ * Creating a rtxsend bin
+ *
+ * Returns: (transfer full): a #GstElement.
+ *
+ * Since: 1.6
+ */
+GstElement *
+gst_rtsp_stream_request_aux_sender (GstRTSPStream * stream, guint sessid)
+{
+ GstElement *bin;
+ GstPad *pad;
+ GstStructure *pt_map;
+ gchar *name;
+ guint pt, rtx_pt;
+ gchar *pt_s;
+
+ g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
+
+ pt = gst_rtsp_stream_get_pt (stream);
+ pt_s = g_strdup_printf ("%u", pt);
+ rtx_pt = stream->priv->rtx_pt;
+
+ GST_INFO ("creating rtxsend with pt %u to %u", pt, rtx_pt);
+
+ bin = gst_bin_new (NULL);
+ stream->priv->rtxsend = gst_element_factory_make ("rtprtxsend", NULL);
+ pt_map = gst_structure_new ("application/x-rtp-pt-map",
+ pt_s, G_TYPE_UINT, rtx_pt, NULL);
+ g_object_set (stream->priv->rtxsend, "payload-type-map", pt_map,
+ "max-size-time", GST_TIME_AS_MSECONDS (stream->priv->rtx_time), NULL);
+ g_free (pt_s);
+ gst_structure_free (pt_map);
+ gst_bin_add (GST_BIN (bin), gst_object_ref (stream->priv->rtxsend));
+
+ pad = gst_element_get_static_pad (stream->priv->rtxsend, "src");
+ name = g_strdup_printf ("src_%u", sessid);
+ gst_element_add_pad (bin, gst_ghost_pad_new (name, pad));
+ g_free (name);
+ gst_object_unref (pad);
+
+ pad = gst_element_get_static_pad (stream->priv->rtxsend, "sink");
+ name = g_strdup_printf ("sink_%u", sessid);
+ gst_element_add_pad (bin, gst_ghost_pad_new (name, pad));
+ g_free (name);
+ gst_object_unref (pad);
+
+ return bin;
+}
+
+/**
+ * gst_rtsp_stream_set_pt_map:
+ * @stream: a #GstRTSPStream
+ * @pt: the pt
+ * @caps: a #GstCaps
+ *
+ * Configure a pt map between @pt and @caps.
+ */
+void
+gst_rtsp_stream_set_pt_map (GstRTSPStream * stream, guint pt, GstCaps * caps)
+{
+ GstRTSPStreamPrivate *priv = stream->priv;
+
+ g_mutex_lock (&priv->lock);
+ g_hash_table_insert (priv->ptmap, GINT_TO_POINTER (pt), gst_caps_ref (caps));
+ g_mutex_unlock (&priv->lock);
+}
+
+static GstCaps *
+request_pt_map (GstElement * rtpbin, guint session, guint pt,
+ GstRTSPStream * stream)
+{
+ GstRTSPStreamPrivate *priv = stream->priv;
+ GstCaps *caps = NULL;
+
+ g_mutex_lock (&priv->lock);
+
+ if (priv->idx == session) {
+ caps = g_hash_table_lookup (priv->ptmap, GINT_TO_POINTER (pt));
+ if (caps) {
+ GST_DEBUG ("Stream %p, pt %u: caps %" GST_PTR_FORMAT, stream, pt, caps);
+ gst_caps_ref (caps);
+ } else {
+ GST_DEBUG ("Stream %p, pt %u: no caps", stream, pt);
+ }
+ }
+
+ g_mutex_unlock (&priv->lock);
+
+ return caps;
+}
+
+static void
+pad_added (GstElement * rtpbin, GstPad * pad, GstRTSPStream * stream)
+{
+ GstRTSPStreamPrivate *priv = stream->priv;
+ gchar *name;
+ GstPadLinkReturn ret;
+ guint sessid;
+
+ GST_DEBUG ("Stream %p added pad %s:%s for pad %s:%s", stream,
+ GST_DEBUG_PAD_NAME (pad), GST_DEBUG_PAD_NAME (priv->sinkpad));
+
+ name = gst_pad_get_name (pad);
+ if (sscanf (name, "recv_rtp_src_%u", &sessid) != 1) {
+ g_free (name);
+ return;
+ }
+ g_free (name);
+
+ if (priv->idx != sessid)
+ return;
+
+ if (gst_pad_is_linked (priv->sinkpad)) {
+ GST_WARNING ("Stream %p: Pad %s:%s is linked already", stream,
+ GST_DEBUG_PAD_NAME (priv->sinkpad));
+ return;
+ }
+
+ /* link the RTP pad to the session manager, it should not really fail unless
+ * this is not really an RTP pad */
+ ret = gst_pad_link (pad, priv->sinkpad);
+ if (ret != GST_PAD_LINK_OK)
+ goto link_failed;
+ priv->recv_rtp_src = gst_object_ref (pad);
+
+ return;
+
+/* ERRORS */
+link_failed:
+ {
+ GST_ERROR ("Stream %p: Failed to link pads %s:%s and %s:%s", stream,
+ GST_DEBUG_PAD_NAME (pad), GST_DEBUG_PAD_NAME (priv->sinkpad));
+ }
+}
+
+static void
+on_npt_stop (GstElement * rtpbin, guint session, guint ssrc,
+ GstRTSPStream * stream)
+{
+ /* TODO: What to do here other than this? */
+ GST_DEBUG ("Stream %p: Got EOS", stream);
+ gst_pad_send_event (stream->priv->sinkpad, gst_event_new_eos ());
+}
+
+/**
* gst_rtsp_stream_join_bin:
* @stream: a #GstRTSPStream
* @bin: (transfer none): a #GstBin to join
gint i;
guint idx;
gchar *name;
- GstPad *pad, *sinkpad, *selpad;
+ GstPad *pad, *sinkpad = NULL, *selpad;
GstPadLinkReturn ret;
+ gboolean is_tcp = FALSE, is_udp = FALSE;
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
GST_INFO ("stream %p joining bin as session %u", stream, idx);
- if (!alloc_ports (stream))
+ is_tcp = priv->protocols & GST_RTSP_LOWER_TRANS_TCP;
+
+ is_udp = ((priv->protocols & GST_RTSP_LOWER_TRANS_UDP) ||
+ (priv->protocols & GST_RTSP_LOWER_TRANS_UDP_MCAST));
+
+ if (is_udp && !alloc_ports (stream))
goto no_ports;
/* update the dscp qos field in the sinks */
(GCallback) request_rtp_encoder, stream);
g_signal_connect (rtpbin, "request-rtcp-encoder",
(GCallback) request_rtcp_encoder, stream);
+ g_signal_connect (rtpbin, "request-rtp-decoder",
+ (GCallback) request_rtp_rtcp_decoder, stream);
g_signal_connect (rtpbin, "request-rtcp-decoder",
- (GCallback) request_rtcp_decoder, stream);
+ (GCallback) request_rtp_rtcp_decoder, stream);
}
- /* get a pad for sending RTP */
- name = g_strdup_printf ("send_rtp_sink_%u", idx);
- priv->send_rtp_sink = gst_element_get_request_pad (rtpbin, name);
- g_free (name);
- /* link the RTP pad to the session manager, it should not really fail unless
- * this is not really an RTP pad */
- ret = gst_pad_link (priv->srcpad, priv->send_rtp_sink);
- if (ret != GST_PAD_LINK_OK)
- goto link_failed;
+ if (priv->sinkpad) {
+ g_signal_connect (rtpbin, "request-pt-map",
+ (GCallback) request_pt_map, stream);
+ }
/* get pads from the RTP session element for sending and receiving
* RTP/RTCP*/
- name = g_strdup_printf ("send_rtp_src_%u", idx);
- priv->send_src[0] = gst_element_get_static_pad (rtpbin, name);
- g_free (name);
+ if (priv->srcpad) {
+ /* get a pad for sending RTP */
+ name = g_strdup_printf ("send_rtp_sink_%u", idx);
+ priv->send_rtp_sink = gst_element_get_request_pad (rtpbin, name);
+ g_free (name);
+
+ /* link the RTP pad to the session manager, it should not really fail unless
+ * this is not really an RTP pad */
+ ret = gst_pad_link (priv->srcpad, priv->send_rtp_sink);
+ if (ret != GST_PAD_LINK_OK)
+ goto link_failed;
+
+ name = g_strdup_printf ("send_rtp_src_%u", idx);
+ priv->send_src[0] = gst_element_get_static_pad (rtpbin, name);
+ g_free (name);
+ } else {
+ /* Need to connect our sinkpad from here */
+ g_signal_connect (rtpbin, "pad-added", (GCallback) pad_added, stream);
+ /* EOS */
+ g_signal_connect (rtpbin, "on-npt-stop", (GCallback) on_npt_stop, stream);
+
+ name = g_strdup_printf ("recv_rtp_sink_%u", idx);
+ priv->recv_sink[0] = gst_element_get_request_pad (rtpbin, name);
+ g_free (name);
+ }
+
name = g_strdup_printf ("send_rtcp_src_%u", idx);
priv->send_src[1] = gst_element_get_request_pad (rtpbin, name);
g_free (name);
- name = g_strdup_printf ("recv_rtp_sink_%u", idx);
- priv->recv_sink[0] = gst_element_get_request_pad (rtpbin, name);
- g_free (name);
name = g_strdup_printf ("recv_rtcp_sink_%u", idx);
priv->recv_sink[1] = gst_element_get_request_pad (rtpbin, name);
g_free (name);
g_signal_connect (priv->session, "on-timeout", (GCallback) on_timeout,
stream);
+ /* signal for sender ssrc */
+ g_signal_connect (priv->session, "on-new-sender-ssrc",
+ (GCallback) on_new_sender_ssrc, stream);
+ g_signal_connect (priv->session, "on-sender-ssrc-active",
+ (GCallback) on_sender_ssrc_active, stream);
+
for (i = 0; i < 2; i++) {
GstPad *teepad, *queuepad;
/* For the sender we create this bit of pipeline for both
* RTP and RTCP. Sync and preroll are enabled on udpsink so
- * we need to add a queue before appsink to make the pipeline
- * not block. For the TCP case, we want to pump data to the
- * client as fast as possible anyway.
+ * we need to add a queue before appsink and udpsink to make
+ * the pipeline not block. For the TCP case, we want to pump
+ * client as fast as possible anyway. This pipeline is used
+ * when both TCP and UDP are present.
*
- * .--------. .-----. .---------.
- * | rtpbin | | tee | | udpsink |
- * | send->sink src->sink |
- * '--------' | | '---------'
+ * .--------. .-----. .---------. .---------.
+ * | rtpbin | | tee | | queue | | udpsink |
+ * | send->sink src->sink src->sink |
+ * '--------' | | '---------' '---------'
* | | .---------. .---------.
* | | | queue | | appsink |
* | src->sink src->sink |
* '-----' '---------' '---------'
*
- * When only UDP is allowed, we skip the tee, queue and appsink and link the
- * udpsink directly to the session.
+ * When only UDP or only TCP is allowed, we skip the tee and queue
+ * and link the udpsink (for UDP) or appsink (for TCP) directly to
+ * the session.
*/
- /* add udpsink */
- gst_bin_add (bin, priv->udpsink[i]);
- sinkpad = gst_element_get_static_pad (priv->udpsink[i], "sink");
-
- if (priv->protocols & GST_RTSP_LOWER_TRANS_TCP) {
- /* make tee for RTP/RTCP */
- priv->tee[i] = gst_element_factory_make ("tee", NULL);
- gst_bin_add (bin, priv->tee[i]);
-
- /* and link to rtpbin send pad */
- pad = gst_element_get_static_pad (priv->tee[i], "sink");
- gst_pad_link (priv->send_src[i], pad);
- gst_object_unref (pad);
-
- /* link tee to udpsink */
- teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
- gst_pad_link (teepad, sinkpad);
- gst_object_unref (teepad);
-
- /* make queue */
- priv->appqueue[i] = gst_element_factory_make ("queue", NULL);
- gst_bin_add (bin, priv->appqueue[i]);
- /* and link to tee */
- teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
- pad = gst_element_get_static_pad (priv->appqueue[i], "sink");
- gst_pad_link (teepad, pad);
- gst_object_unref (pad);
- gst_object_unref (teepad);
-
- /* make appsink */
- priv->appsink[i] = gst_element_factory_make ("appsink", NULL);
- g_object_set (priv->appsink[i], "async", FALSE, "sync", FALSE, NULL);
- g_object_set (priv->appsink[i], "emit-signals", FALSE, NULL);
- gst_bin_add (bin, priv->appsink[i]);
- gst_app_sink_set_callbacks (GST_APP_SINK_CAST (priv->appsink[i]),
- &sink_cb, stream, NULL);
- /* and link to queue */
- queuepad = gst_element_get_static_pad (priv->appqueue[i], "src");
- pad = gst_element_get_static_pad (priv->appsink[i], "sink");
- gst_pad_link (queuepad, pad);
- gst_object_unref (pad);
- gst_object_unref (queuepad);
- } else {
- /* else only udpsink needed, link it to the session */
- gst_pad_link (priv->send_src[i], sinkpad);
- }
- gst_object_unref (sinkpad);
- /* For the receiver we create this bit of pipeline for both
- * RTP and RTCP. We receive RTP/RTCP on appsrc and udpsrc
- * and it is all funneled into the rtpbin receive pad.
- *
- * .--------. .--------. .--------.
- * | udpsrc | | funnel | | rtpbin |
- * | src->sink src->sink |
- * '--------' | | '--------'
- * .--------. | |
- * | appsrc | | |
- * | src->sink |
- * '--------' '--------'
- */
- /* make funnel for the RTP/RTCP receivers */
- priv->funnel[i] = gst_element_factory_make ("funnel", NULL);
- gst_bin_add (bin, priv->funnel[i]);
+ /* Only link the RTP send src if we're going to send RTP, link
+ * the RTCP send src always */
+ if (priv->srcpad || i == 1) {
+ if (is_udp) {
+ /* add udpsink */
+ gst_bin_add (bin, priv->udpsink[i]);
+ sinkpad = gst_element_get_static_pad (priv->udpsink[i], "sink");
+ }
- pad = gst_element_get_static_pad (priv->funnel[i], "src");
- gst_pad_link (pad, priv->recv_sink[i]);
- gst_object_unref (pad);
+ if (is_tcp) {
+ /* make appsink */
+ priv->appsink[i] = gst_element_factory_make ("appsink", NULL);
+ g_object_set (priv->appsink[i], "emit-signals", FALSE, NULL);
+ gst_bin_add (bin, priv->appsink[i]);
+ gst_app_sink_set_callbacks (GST_APP_SINK_CAST (priv->appsink[i]),
+ &sink_cb, stream, NULL);
+ }
- if (priv->udpsrc_v4[i]) {
- /* we set and keep these to playing so that they don't cause NO_PREROLL return
- * values */
- gst_element_set_state (priv->udpsrc_v4[i], GST_STATE_PLAYING);
- gst_element_set_locked_state (priv->udpsrc_v4[i], TRUE);
- /* add udpsrc */
- gst_bin_add (bin, priv->udpsrc_v4[i]);
-
- /* and link to the funnel v4 */
- selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
- pad = gst_element_get_static_pad (priv->udpsrc_v4[i], "src");
- gst_pad_link (pad, selpad);
- gst_object_unref (pad);
- gst_object_unref (selpad);
+ if (is_udp && is_tcp) {
+ g_object_set (priv->appsink[i], "async", FALSE, "sync", FALSE, NULL);
+
+ /* make tee for RTP/RTCP */
+ priv->tee[i] = gst_element_factory_make ("tee", NULL);
+ gst_bin_add (bin, priv->tee[i]);
+
+ /* and link to rtpbin send pad */
+ pad = gst_element_get_static_pad (priv->tee[i], "sink");
+ gst_pad_link (priv->send_src[i], pad);
+ gst_object_unref (pad);
+
+ priv->udpqueue[i] = gst_element_factory_make ("queue", NULL);
+ g_object_set (priv->udpqueue[i], "max-size-buffers",
+ 1, "max-size-bytes", 0, "max-size-time", G_GINT64_CONSTANT (0),
+ NULL);
+ gst_bin_add (bin, priv->udpqueue[i]);
+ /* link tee to udpqueue */
+ teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
+ pad = gst_element_get_static_pad (priv->udpqueue[i], "sink");
+ gst_pad_link (teepad, pad);
+ gst_object_unref (pad);
+ gst_object_unref (teepad);
+
+ /* link udpqueue to udpsink */
+ queuepad = gst_element_get_static_pad (priv->udpqueue[i], "src");
+ gst_pad_link (queuepad, sinkpad);
+ gst_object_unref (queuepad);
+ gst_object_unref (sinkpad);
+
+ /* make appqueue */
+ priv->appqueue[i] = gst_element_factory_make ("queue", NULL);
+ g_object_set (priv->appqueue[i], "max-size-buffers",
+ 1, "max-size-bytes", 0, "max-size-time", G_GINT64_CONSTANT (0),
+ NULL);
+ gst_bin_add (bin, priv->appqueue[i]);
+ /* and link tee to appqueue */
+ teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
+ pad = gst_element_get_static_pad (priv->appqueue[i], "sink");
+ gst_pad_link (teepad, pad);
+ gst_object_unref (pad);
+ gst_object_unref (teepad);
+
+ /* and link appqueue to appsink */
+ queuepad = gst_element_get_static_pad (priv->appqueue[i], "src");
+ pad = gst_element_get_static_pad (priv->appsink[i], "sink");
+ gst_pad_link (queuepad, pad);
+ gst_object_unref (pad);
+ gst_object_unref (queuepad);
+ } else if (is_tcp) {
+ /* only appsink needed, link it to the session */
+ pad = gst_element_get_static_pad (priv->appsink[i], "sink");
+ gst_pad_link (priv->send_src[i], pad);
+ gst_object_unref (pad);
+
+ /* when its only TCP, we need to set sync and preroll to FALSE
+ * for the sink to avoid deadlock. And this is only needed for
+ * sink used for RTCP data, not the RTP data. */
+ if (i == 1)
+ g_object_set (priv->appsink[i], "async", FALSE, "sync", FALSE, NULL);
+ } else {
+ /* else only udpsink needed, link it to the session */
+ gst_pad_link (priv->send_src[i], sinkpad);
+ gst_object_unref (sinkpad);
+ }
}
- if (priv->udpsrc_v6[i]) {
- gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_PLAYING);
- gst_element_set_locked_state (priv->udpsrc_v6[i], TRUE);
- gst_bin_add (bin, priv->udpsrc_v6[i]);
-
- /* and link to the funnel v6 */
- selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
- pad = gst_element_get_static_pad (priv->udpsrc_v6[i], "src");
- gst_pad_link (pad, selpad);
+ /* Only connect recv RTP sink if we expect to receive RTP. Connect recv
+ * RTCP sink always */
+ if (priv->sinkpad || i == 1) {
+ /* For the receiver we create this bit of pipeline for both
+ * RTP and RTCP. We receive RTP/RTCP on appsrc and udpsrc
+ * and it is all funneled into the rtpbin receive pad.
+ *
+ * .--------. .--------. .--------.
+ * | udpsrc | | funnel | | rtpbin |
+ * | src->sink src->sink |
+ * '--------' | | '--------'
+ * .--------. | |
+ * | appsrc | | |
+ * | src->sink |
+ * '--------' '--------'
+ */
+ /* make funnel for the RTP/RTCP receivers */
+ priv->funnel[i] = gst_element_factory_make ("funnel", NULL);
+ gst_bin_add (bin, priv->funnel[i]);
+
+ pad = gst_element_get_static_pad (priv->funnel[i], "src");
+ gst_pad_link (pad, priv->recv_sink[i]);
gst_object_unref (pad);
- gst_object_unref (selpad);
- }
- if (priv->protocols & GST_RTSP_LOWER_TRANS_TCP) {
- /* make and add appsrc */
- priv->appsrc[i] = gst_element_factory_make ("appsrc", NULL);
- gst_bin_add (bin, priv->appsrc[i]);
- /* and link to the funnel */
- selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
- pad = gst_element_get_static_pad (priv->appsrc[i], "src");
- gst_pad_link (pad, selpad);
- gst_object_unref (pad);
- gst_object_unref (selpad);
+ if (priv->udpsrc_v4[i]) {
+ if (priv->srcpad) {
+ /* we set and keep these to playing so that they don't cause NO_PREROLL return
+ * values. This is only relevant for PLAY pipelines */
+ gst_element_set_state (priv->udpsrc_v4[i], GST_STATE_PLAYING);
+ gst_element_set_locked_state (priv->udpsrc_v4[i], TRUE);
+ }
+ /* add udpsrc */
+ gst_bin_add (bin, priv->udpsrc_v4[i]);
+
+ /* and link to the funnel v4 */
+ selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
+ pad = gst_element_get_static_pad (priv->udpsrc_v4[i], "src");
+ gst_pad_link (pad, selpad);
+ gst_object_unref (pad);
+ gst_object_unref (selpad);
+ }
+
+ if (priv->udpsrc_v6[i]) {
+ if (priv->srcpad) {
+ gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_PLAYING);
+ gst_element_set_locked_state (priv->udpsrc_v6[i], TRUE);
+ }
+ gst_bin_add (bin, priv->udpsrc_v6[i]);
+
+ /* and link to the funnel v6 */
+ selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
+ pad = gst_element_get_static_pad (priv->udpsrc_v6[i], "src");
+ gst_pad_link (pad, selpad);
+ gst_object_unref (pad);
+ gst_object_unref (selpad);
+ }
+
+ if (is_tcp) {
+ /* make and add appsrc */
+ priv->appsrc[i] = gst_element_factory_make ("appsrc", NULL);
+ priv->appsrc_base_time[i] = -1;
+ g_object_set (priv->appsrc[i], "format", GST_FORMAT_TIME, NULL);
+ gst_bin_add (bin, priv->appsrc[i]);
+ /* and link to the funnel */
+ selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
+ pad = gst_element_get_static_pad (priv->appsrc[i], "src");
+ gst_pad_link (pad, selpad);
+ gst_object_unref (pad);
+ gst_object_unref (selpad);
+ }
}
/* check if we need to set to a special state */
if (state != GST_STATE_NULL) {
- if (priv->udpsink[i])
+ if (priv->udpsink[i] && (priv->srcpad || i == 1))
gst_element_set_state (priv->udpsink[i], state);
- if (priv->appsink[i])
+ if (priv->appsink[i] && (priv->srcpad || i == 1))
gst_element_set_state (priv->appsink[i], state);
- if (priv->appqueue[i])
+ if (priv->appqueue[i] && (priv->srcpad || i == 1))
gst_element_set_state (priv->appqueue[i], state);
- if (priv->tee[i])
+ if (priv->udpqueue[i] && (priv->srcpad || i == 1))
+ gst_element_set_state (priv->udpqueue[i], state);
+ if (priv->tee[i] && (priv->srcpad || i == 1))
gst_element_set_state (priv->tee[i], state);
- if (priv->funnel[i])
+ if (priv->funnel[i] && (priv->sinkpad || i == 1))
gst_element_set_state (priv->funnel[i], state);
- if (priv->appsrc[i])
+ if (priv->appsrc[i] && (priv->sinkpad || i == 1))
gst_element_set_state (priv->appsrc[i], state);
}
}
- /* be notified of caps changes */
- priv->caps_sig = g_signal_connect (priv->send_src[0], "notify::caps",
- (GCallback) caps_notify, stream);
+ if (priv->srcpad) {
+ /* be notified of caps changes */
+ priv->caps_sig = g_signal_connect (priv->send_src[0], "notify::caps",
+ (GCallback) caps_notify, stream);
+ }
priv->is_joined = TRUE;
g_mutex_unlock (&priv->lock);
{
GstRTSPStreamPrivate *priv;
gint i;
+ GList *l;
+ gboolean is_tcp, is_udp;
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
goto was_not_joined;
/* all transports must be removed by now */
- g_return_val_if_fail (priv->transports == NULL, FALSE);
+ if (priv->transports != NULL)
+ goto transports_not_removed;
- clear_tr_cache (priv);
+ clear_tr_cache (priv, TRUE);
+ clear_tr_cache (priv, FALSE);
GST_INFO ("stream %p leaving bin", stream);
- gst_pad_unlink (priv->srcpad, priv->send_rtp_sink);
- g_signal_handler_disconnect (priv->send_src[0], priv->caps_sig);
- gst_element_release_request_pad (rtpbin, priv->send_rtp_sink);
- gst_object_unref (priv->send_rtp_sink);
- priv->send_rtp_sink = NULL;
+ if (priv->srcpad) {
+ gst_pad_unlink (priv->srcpad, priv->send_rtp_sink);
+
+ g_signal_handler_disconnect (priv->send_src[0], priv->caps_sig);
+ gst_element_release_request_pad (rtpbin, priv->send_rtp_sink);
+ gst_object_unref (priv->send_rtp_sink);
+ priv->send_rtp_sink = NULL;
+ } else if (priv->recv_rtp_src) {
+ gst_pad_unlink (priv->recv_rtp_src, priv->sinkpad);
+ gst_object_unref (priv->recv_rtp_src);
+ priv->recv_rtp_src = NULL;
+ }
+
+ is_tcp = priv->protocols & GST_RTSP_LOWER_TRANS_TCP;
+
+ is_udp = ((priv->protocols & GST_RTSP_LOWER_TRANS_UDP) ||
+ (priv->protocols & GST_RTSP_LOWER_TRANS_UDP_MCAST));
+
for (i = 0; i < 2; i++) {
if (priv->udpsink[i])
gst_element_set_state (priv->appsink[i], GST_STATE_NULL);
if (priv->appqueue[i])
gst_element_set_state (priv->appqueue[i], GST_STATE_NULL);
+ if (priv->udpqueue[i])
+ gst_element_set_state (priv->udpqueue[i], GST_STATE_NULL);
if (priv->tee[i])
gst_element_set_state (priv->tee[i], GST_STATE_NULL);
if (priv->funnel[i])
gst_element_set_state (priv->funnel[i], GST_STATE_NULL);
if (priv->appsrc[i])
gst_element_set_state (priv->appsrc[i], GST_STATE_NULL);
+
if (priv->udpsrc_v4[i]) {
- /* and set udpsrc to NULL now before removing */
- gst_element_set_locked_state (priv->udpsrc_v4[i], FALSE);
- gst_element_set_state (priv->udpsrc_v4[i], GST_STATE_NULL);
- /* removing them should also nicely release the request
- * pads when they finalize */
- gst_bin_remove (bin, priv->udpsrc_v4[i]);
+ if (priv->sinkpad || i == 1) {
+ /* and set udpsrc to NULL now before removing */
+ gst_element_set_locked_state (priv->udpsrc_v4[i], FALSE);
+ gst_element_set_state (priv->udpsrc_v4[i], GST_STATE_NULL);
+ /* removing them should also nicely release the request
+ * pads when they finalize */
+ gst_bin_remove (bin, priv->udpsrc_v4[i]);
+ } else {
+ /* we need to set the state to NULL before unref */
+ gst_element_set_state (priv->udpsrc_v4[i], GST_STATE_NULL);
+ gst_object_unref (priv->udpsrc_v4[i]);
+ }
}
+
if (priv->udpsrc_v6[i]) {
- gst_element_set_locked_state (priv->udpsrc_v6[i], FALSE);
- gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_NULL);
- gst_bin_remove (bin, priv->udpsrc_v6[i]);
+ if (priv->sinkpad || i == 1) {
+ gst_element_set_locked_state (priv->udpsrc_v6[i], FALSE);
+ gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_NULL);
+ gst_bin_remove (bin, priv->udpsrc_v6[i]);
+ } else {
+ gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_NULL);
+ gst_object_unref (priv->udpsrc_v6[i]);
+ }
}
- if (priv->udpsink[i])
+
+ for (l = priv->transport_sources; l; l = l->next) {
+ GstRTSPMulticastTransportSource *s = l->data;
+
+ if (!s->udpsrc[i])
+ continue;
+
+ gst_element_set_locked_state (s->udpsrc[i], FALSE);
+ gst_element_set_state (s->udpsrc[i], GST_STATE_NULL);
+ gst_bin_remove (bin, s->udpsrc[i]);
+ }
+
+ if (priv->udpsink[i] && is_udp && (priv->srcpad || i == 1))
gst_bin_remove (bin, priv->udpsink[i]);
- if (priv->appsrc[i])
+ if (priv->appsrc[i] && (priv->sinkpad || i == 1))
gst_bin_remove (bin, priv->appsrc[i]);
- if (priv->appsink[i])
+ if (priv->appsink[i] && is_tcp && (priv->srcpad || i == 1))
gst_bin_remove (bin, priv->appsink[i]);
- if (priv->appqueue[i])
+ if (priv->appqueue[i] && is_tcp && is_udp && (priv->srcpad || i == 1))
gst_bin_remove (bin, priv->appqueue[i]);
- if (priv->tee[i])
+ if (priv->udpqueue[i] && is_tcp && is_udp && (priv->srcpad || i == 1))
+ gst_bin_remove (bin, priv->udpqueue[i]);
+ if (priv->tee[i] && is_tcp && is_udp && (priv->srcpad || i == 1))
gst_bin_remove (bin, priv->tee[i]);
- if (priv->funnel[i])
+ if (priv->funnel[i] && (priv->sinkpad || i == 1))
gst_bin_remove (bin, priv->funnel[i]);
- gst_element_release_request_pad (rtpbin, priv->recv_sink[i]);
- gst_object_unref (priv->recv_sink[i]);
- priv->recv_sink[i] = NULL;
+ if (priv->sinkpad || i == 1) {
+ gst_element_release_request_pad (rtpbin, priv->recv_sink[i]);
+ gst_object_unref (priv->recv_sink[i]);
+ priv->recv_sink[i] = NULL;
+ }
priv->udpsrc_v4[i] = NULL;
priv->udpsrc_v6[i] = NULL;
priv->appsrc[i] = NULL;
priv->appsink[i] = NULL;
priv->appqueue[i] = NULL;
+ priv->udpqueue[i] = NULL;
priv->tee[i] = NULL;
priv->funnel[i] = NULL;
}
- gst_object_unref (priv->send_src[0]);
- priv->send_src[0] = NULL;
+
+ for (l = priv->transport_sources; l; l = l->next) {
+ GstRTSPMulticastTransportSource *s = l->data;
+ g_slice_free (GstRTSPMulticastTransportSource, s);
+ }
+ g_list_free (priv->transport_sources);
+ priv->transport_sources = NULL;
+
+ if (priv->srcpad) {
+ gst_object_unref (priv->send_src[0]);
+ priv->send_src[0] = NULL;
+ }
gst_element_release_request_pad (rtpbin, priv->send_src[1]);
gst_object_unref (priv->send_src[1]);
if (priv->srtpenc)
gst_object_unref (priv->srtpenc);
+ if (priv->srtpdec)
+ gst_object_unref (priv->srtpdec);
priv->is_joined = FALSE;
g_mutex_unlock (&priv->lock);
g_mutex_unlock (&priv->lock);
return TRUE;
}
+transports_not_removed:
+ {
+ GST_ERROR_OBJECT (stream, "can't leave bin (transports not removed)");
+ g_mutex_unlock (&priv->lock);
+ return FALSE;
+ }
}
/**
g_mutex_lock (&priv->lock);
+ /* First try to extract the information from the last buffer on the sinks.
+ * This will have a more accurate sequence number and timestamp, as between
+ * the payloader and the sink there can be some queues
+ */
+ if (priv->udpsink[0] || priv->appsink[0]) {
+ GstSample *last_sample;
+
+ if (priv->udpsink[0])
+ g_object_get (priv->udpsink[0], "last-sample", &last_sample, NULL);
+ else
+ g_object_get (priv->appsink[0], "last-sample", &last_sample, NULL);
+
+ if (last_sample) {
+ GstCaps *caps;
+ GstBuffer *buffer;
+ GstSegment *segment;
+ GstRTPBuffer rtp_buffer = GST_RTP_BUFFER_INIT;
+
+ caps = gst_sample_get_caps (last_sample);
+ buffer = gst_sample_get_buffer (last_sample);
+ segment = gst_sample_get_segment (last_sample);
+
+ if (gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp_buffer)) {
+ if (seq) {
+ *seq = gst_rtp_buffer_get_seq (&rtp_buffer);
+ }
+
+ if (rtptime) {
+ *rtptime = gst_rtp_buffer_get_timestamp (&rtp_buffer);
+ }
+
+ gst_rtp_buffer_unmap (&rtp_buffer);
+
+ if (running_time) {
+ *running_time =
+ gst_segment_to_running_time (segment, GST_FORMAT_TIME,
+ GST_BUFFER_TIMESTAMP (buffer));
+ }
+
+ if (clock_rate) {
+ GstStructure *s = gst_caps_get_structure (caps, 0);
+
+ gst_structure_get_int (s, "clock-rate", (gint *) clock_rate);
+
+ if (*clock_rate == 0 && running_time)
+ *running_time = GST_CLOCK_TIME_NONE;
+ }
+ gst_sample_unref (last_sample);
+
+ goto done;
+ } else {
+ gst_sample_unref (last_sample);
+ }
+ }
+ }
+
if (g_object_class_find_property (payobjclass, "stats")) {
g_object_get (priv->payloader, "stats", &stats, NULL);
if (stats == NULL)
if (running_time)
*running_time = GST_CLOCK_TIME_NONE;
}
+
+done:
g_mutex_unlock (&priv->lock);
return TRUE;
g_mutex_unlock (&priv->lock);
if (element) {
+ if (priv->appsrc_base_time[0] == -1) {
+ /* Take current running_time. This timestamp will be put on
+ * the first buffer of each stream because we are a live source and so we
+ * timestamp with the running_time. When we are dealing with TCP, we also
+ * only timestamp the first buffer (using the DISCONT flag) because a server
+ * typically bursts data, for which we don't want to compensate by speeding
+ * up the media. The other timestamps will be interpollated from this one
+ * using the RTP timestamps. */
+ GST_OBJECT_LOCK (element);
+ if (GST_ELEMENT_CLOCK (element)) {
+ GstClockTime now;
+ GstClockTime base_time;
+
+ now = gst_clock_get_time (GST_ELEMENT_CLOCK (element));
+ base_time = GST_ELEMENT_CAST (element)->base_time;
+
+ priv->appsrc_base_time[0] = now - base_time;
+ GST_BUFFER_TIMESTAMP (buffer) = priv->appsrc_base_time[0];
+ GST_DEBUG ("stream %p: first buffer at time %" GST_TIME_FORMAT
+ ", base %" GST_TIME_FORMAT, stream, GST_TIME_ARGS (now),
+ GST_TIME_ARGS (base_time));
+ }
+ GST_OBJECT_UNLOCK (element);
+ }
+
ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (element), buffer);
gst_object_unref (element);
} else {
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), GST_FLOW_ERROR);
priv = stream->priv;
g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
- g_return_val_if_fail (priv->is_joined, FALSE);
+ if (!priv->is_joined) {
+ gst_buffer_unref (buffer);
+ return GST_FLOW_NOT_LINKED;
+ }
g_mutex_lock (&priv->lock);
if (priv->appsrc[1])
element = gst_object_ref (priv->appsrc[1]);
g_mutex_unlock (&priv->lock);
if (element) {
+ if (priv->appsrc_base_time[1] == -1) {
+ /* Take current running_time. This timestamp will be put on
+ * the first buffer of each stream because we are a live source and so we
+ * timestamp with the running_time. When we are dealing with TCP, we also
+ * only timestamp the first buffer (using the DISCONT flag) because a server
+ * typically bursts data, for which we don't want to compensate by speeding
+ * up the media. The other timestamps will be interpollated from this one
+ * using the RTP timestamps. */
+ GST_OBJECT_LOCK (element);
+ if (GST_ELEMENT_CLOCK (element)) {
+ GstClockTime now;
+ GstClockTime base_time;
+
+ now = gst_clock_get_time (GST_ELEMENT_CLOCK (element));
+ base_time = GST_ELEMENT_CAST (element)->base_time;
+
+ priv->appsrc_base_time[1] = now - base_time;
+ GST_BUFFER_TIMESTAMP (buffer) = priv->appsrc_base_time[1];
+ GST_DEBUG ("stream %p: first buffer at time %" GST_TIME_FORMAT
+ ", base %" GST_TIME_FORMAT, stream, GST_TIME_ARGS (now),
+ GST_TIME_ARGS (base_time));
+ }
+ GST_OBJECT_UNLOCK (element);
+ }
+
ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (element), buffer);
gst_object_unref (element);
} else {
tr = gst_rtsp_stream_transport_get_transport (trans);
switch (tr->lower_transport) {
- case GST_RTSP_LOWER_TRANS_UDP:
case GST_RTSP_LOWER_TRANS_UDP_MCAST:
{
+ GstRTSPMulticastTransportSource *source;
+ GstBin *bin;
+
+ bin = GST_BIN (gst_object_get_parent (GST_OBJECT (priv->funnel[0])));
+
+ if (add) {
+ gchar *host;
+ gint i;
+ GstPad *selpad, *pad;
+
+ source = g_slice_new0 (GstRTSPMulticastTransportSource);
+ source->transport = trans;
+
+ for (i = 0; i < 2; i++) {
+ host =
+ g_strdup_printf ("udp://%s:%d", tr->destination,
+ (i == 0) ? tr->port.min : tr->port.max);
+ source->udpsrc[i] =
+ gst_element_make_from_uri (GST_URI_SRC, host, NULL, NULL);
+ g_free (host);
+ g_object_set (source->udpsrc[i], "loop", FALSE, NULL);
+
+ if (priv->srcpad) {
+ /* we set and keep these to playing so that they don't cause NO_PREROLL return
+ * values. This is only relevant for PLAY pipelines */
+ gst_element_set_state (source->udpsrc[i], GST_STATE_PLAYING);
+ gst_element_set_locked_state (source->udpsrc[i], TRUE);
+ }
+ /* add udpsrc */
+ gst_bin_add (bin, source->udpsrc[i]);
+
+ /* and link to the funnel v4 */
+ source->selpad[i] = selpad =
+ gst_element_get_request_pad (priv->funnel[i], "sink_%u");
+ pad = gst_element_get_static_pad (source->udpsrc[i], "src");
+ gst_pad_link (pad, selpad);
+ gst_object_unref (pad);
+ gst_object_unref (selpad);
+ }
+
+ priv->transport_sources =
+ g_list_prepend (priv->transport_sources, source);
+ } else {
+ GList *l;
+
+ for (l = priv->transport_sources; l; l = l->next) {
+ source = l->data;
+
+ if (source->transport == trans) {
+ priv->transport_sources =
+ g_list_delete_link (priv->transport_sources, l);
+ break;
+ }
+ }
+
+ if (l != NULL) {
+ gint i;
+
+ for (i = 0; i < 2; i++) {
+ /* Will automatically unlink everything */
+ gst_bin_remove (bin,
+ GST_ELEMENT (gst_object_ref (source->udpsrc[i])));
+
+ gst_element_set_state (source->udpsrc[i], GST_STATE_NULL);
+ gst_object_unref (source->udpsrc[i]);
+
+ gst_element_release_request_pad (priv->funnel[i],
+ source->selpad[i]);
+ }
+
+ g_slice_free (GstRTSPMulticastTransportSource, source);
+ }
+ }
+
+ gst_object_unref (bin);
+
+ /* fall through for the generic case */
+ }
+ case GST_RTSP_LOWER_TRANS_UDP:
+ {
gchar *dest;
gint min, max;
guint ttl = 0;
}
/**
+ * gst_rtsp_stream_set_seqnum:
+ * @stream: a #GstRTSPStream
+ * @seqnum: a new sequence number
+ *
+ * Configure the sequence number in the payloader of @stream to @seqnum.
+ */
+void
+gst_rtsp_stream_set_seqnum_offset (GstRTSPStream * stream, guint16 seqnum)
+{
+ GstRTSPStreamPrivate *priv;
+
+ g_return_if_fail (GST_IS_RTSP_STREAM (stream));
+
+ priv = stream->priv;
+
+ g_object_set (G_OBJECT (priv->payloader), "seqnum-offset", seqnum, NULL);
+}
+
+/**
+ * gst_rtsp_stream_get_seqnum:
+ * @stream: a #GstRTSPStream
+ *
+ * Get the configured sequence number in the payloader of @stream.
+ *
+ * Returns: the sequence number of the payloader.
+ */
+guint16
+gst_rtsp_stream_get_current_seqnum (GstRTSPStream * stream)
+{
+ GstRTSPStreamPrivate *priv;
+ guint seqnum;
+
+ g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), 0);
+
+ priv = stream->priv;
+
+ g_object_get (G_OBJECT (priv->payloader), "seqnum", &seqnum, NULL);
+
+ return seqnum;
+}
+
+/**
* gst_rtsp_stream_transport_filter:
* @stream: a #GstRTSPStream
* @func: (scope call) (allow-none): a callback
{
GstRTSPStreamPrivate *priv;
GList *result, *walk, *next;
- GHashTable *visited;
+ GHashTable *visited = NULL;
guint cookie;
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
return result;
}
+
+/**
+ * gst_rtsp_stream_query_position:
+ * @stream: a #GstRTSPStream
+ *
+ * Query the position of the stream in %GST_FORMAT_TIME. This only considers
+ * the RTP parts of the pipeline and not the RTCP parts.
+ *
+ * Returns: %TRUE if the position could be queried
+ */
+gboolean
+gst_rtsp_stream_query_position (GstRTSPStream * stream, gint64 * position)
+{
+ GstRTSPStreamPrivate *priv;
+ GstElement *sink;
+ gboolean ret;
+
+ g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
+
+ priv = stream->priv;
+
+ g_mutex_lock (&priv->lock);
+ /* depending on the transport type, it should query corresponding sink */
+ if ((priv->protocols & GST_RTSP_LOWER_TRANS_UDP) ||
+ (priv->protocols & GST_RTSP_LOWER_TRANS_UDP_MCAST))
+ sink = priv->udpsink[0];
+ else
+ sink = priv->appsink[0];
+
+ if (sink)
+ gst_object_ref (sink);
+ g_mutex_unlock (&priv->lock);
+
+ if (!sink)
+ return FALSE;
+
+ ret = gst_element_query_position (sink, GST_FORMAT_TIME, position);
+ gst_object_unref (sink);
+
+ return ret;
+}
+
+/**
+ * gst_rtsp_stream_query_stop:
+ * @stream: a #GstRTSPStream
+ *
+ * Query the stop of the stream in %GST_FORMAT_TIME. This only considers
+ * the RTP parts of the pipeline and not the RTCP parts.
+ *
+ * Returns: %TRUE if the stop could be queried
+ */
+gboolean
+gst_rtsp_stream_query_stop (GstRTSPStream * stream, gint64 * stop)
+{
+ GstRTSPStreamPrivate *priv;
+ GstElement *sink;
+ GstQuery *query;
+ gboolean ret;
+
+ g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
+
+ priv = stream->priv;
+
+ g_mutex_lock (&priv->lock);
+ /* depending on the transport type, it should query corresponding sink */
+ if ((priv->protocols & GST_RTSP_LOWER_TRANS_UDP) ||
+ (priv->protocols & GST_RTSP_LOWER_TRANS_UDP_MCAST))
+ sink = priv->udpsink[0];
+ else
+ sink = priv->appsink[0];
+
+ if (sink)
+ gst_object_ref (sink);
+ g_mutex_unlock (&priv->lock);
+
+ if (!sink)
+ return FALSE;
+
+ query = gst_query_new_segment (GST_FORMAT_TIME);
+ if ((ret = gst_element_query (sink, query))) {
+ GstFormat format;
+
+ gst_query_parse_segment (query, NULL, &format, NULL, stop);
+ if (format != GST_FORMAT_TIME)
+ *stop = -1;
+ }
+ gst_query_unref (query);
+ gst_object_unref (sink);
+
+ return ret;
+
+}
projects / platform / upstream / gstreamer.git / blobdiff