#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;
/* transports we stream to */
guint n_active;
GList *transports;
- gboolean tr_changed;
+ guint transports_cookie;
GList *tr_cache;
+ guint tr_cache_cookie;
+
+ /* UDP sources for UDP multicast transports */
+ GList *transport_sources;
gint dscp_qos;
*
* Get the multicast address of @stream for @family.
*
- * Returns: (transfer full): the #GstRTSPAddress of @stream or %NULL when no
- * address could be allocated. gst_rtsp_address_free() after usage.
+ * Returns: (transfer full) (nullable): the #GstRTSPAddress of @stream
+ * or %NULL when no address could be allocated. gst_rtsp_address_free()
+ * after usage.
*/
GstRTSPAddress *
gst_rtsp_stream_get_multicast_address (GstRTSPStream * stream,
*
* Reserve @address and @port as the address and port of @stream.
*
- * Returns: the #GstRTSPAddress of @stream or %NULL when the address could be
- * reserved. gst_rtsp_address_free() after usage.
+ * Returns: (nullable): the #GstRTSPAddress of @stream or %NULL when
+ * the address could be reserved. gst_rtsp_address_free() after usage.
*/
GstRTSPAddress *
gst_rtsp_stream_reserve_address (GstRTSPStream * stream,
udpsrc_out[1] = udpsrc1;
udpsink_out[0] = udpsink0;
udpsink_out[1] = udpsink1;
+
server_port_out->min = rtpport;
server_port_out->max = rtcpport;
is_rtp = GST_ELEMENT_CAST (sink) == priv->appsink[0];
g_mutex_lock (&priv->lock);
- if (priv->tr_changed) {
+ 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));
}
- priv->tr_changed = FALSE;
+ priv->tr_cache_cookie = priv->transports_cookie;
}
g_mutex_unlock (&priv->lock);
{
GstRTSPStreamPrivate *priv;
gint i;
+ GList *l;
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_NULL);
gst_bin_remove (bin, priv->udpsrc_v6[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])
gst_bin_remove (bin, priv->udpsink[i]);
if (priv->appsrc[i])
priv->tee[i] = NULL;
priv->funnel[i] = 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;
+
gst_object_unref (priv->send_src[0]);
priv->send_src[0] = NULL;
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);
+
+ /* we set and keep these to playing so that they don't cause NO_PREROLL return
+ * values */
+ 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);
+ }
+ gst_object_unref (bin);
+
+ 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);
+ }
+ }
+
+ /* fall through for the generic case */
+ }
+ case GST_RTSP_LOWER_TRANS_UDP:
+ {
gchar *dest;
gint min, max;
guint ttl = 0;
g_signal_emit_by_name (priv->udpsink[1], "remove", dest, max, NULL);
priv->transports = g_list_remove (priv->transports, trans);
}
- priv->tr_changed = TRUE;
+ priv->transports_cookie++;
break;
}
case GST_RTSP_LOWER_TRANS_TCP:
GST_INFO ("removing TCP %s", tr->destination);
priv->transports = g_list_remove (priv->transports, trans);
}
- priv->tr_changed = TRUE;
+ priv->transports_cookie++;
break;
default:
goto unknown_transport;
* gst_rtsp_stream_update_crypto:
* @stream: a #GstRTSPStream
* @ssrc: the SSRC
- * @crypto: (transfer none) (allow none): a #GstCaps with crypto info
+ * @crypto: (transfer none) (allow-none): a #GstCaps with crypto info
*
* Update the new crypto information for @ssrc in @stream. If information
* for @ssrc did not exist, it will be added. If information
GstRTSPStreamPrivate *priv;
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
- g_return_val_if_fail (GST_IS_CAPS (crypto), FALSE);
+ g_return_val_if_fail (crypto == NULL || GST_IS_CAPS (crypto), FALSE);
priv = stream->priv;
*
* @stream must be joined to a bin.
*
- * Returns: (transfer full): the RTP socket or %NULL if no socket could be
- * allocated for @family. Unref after usage
+ * Returns: (transfer full) (nullable): the RTP socket or %NULL if no
+ * socket could be allocated for @family. Unref after usage
*/
GSocket *
gst_rtsp_stream_get_rtp_socket (GstRTSPStream * stream, GSocketFamily family)
*
* @stream must be joined to a bin.
*
- * Returns: (transfer full): the RTCP socket or %NULL if no socket could be
- * allocated for @family. Unref after usage
+ * Returns: (transfer full) (nullable): the RTCP socket or %NULL if no
+ * socket could be allocated for @family. Unref after usage
*/
GSocket *
gst_rtsp_stream_get_rtcp_socket (GstRTSPStream * stream, GSocketFamily family)
{
GstRTSPStreamPrivate *priv;
GList *result, *walk, *next;
+ GHashTable *visited;
+ guint cookie;
g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
priv = stream->priv;
result = NULL;
+ if (func)
+ visited = g_hash_table_new_full (NULL, NULL, g_object_unref, NULL);
g_mutex_lock (&priv->lock);
+restart:
+ cookie = priv->transports_cookie;
for (walk = priv->transports; walk; walk = next) {
GstRTSPStreamTransport *trans = walk->data;
GstRTSPFilterResult res;
+ gboolean changed;
next = g_list_next (walk);
- if (func)
+ if (func) {
+ /* only visit each transport once */
+ if (g_hash_table_contains (visited, trans))
+ continue;
+
+ g_hash_table_add (visited, g_object_ref (trans));
+ g_mutex_unlock (&priv->lock);
+
res = func (stream, trans, user_data);
- else
+
+ g_mutex_lock (&priv->lock);
+ } else
res = GST_RTSP_FILTER_REF;
+ changed = (cookie != priv->transports_cookie);
+
switch (res) {
case GST_RTSP_FILTER_REMOVE:
update_transport (stream, trans, FALSE);
default:
break;
}
+ if (changed)
+ goto restart;
}
g_mutex_unlock (&priv->lock);
+ if (func)
+ g_hash_table_unref (visited);
+
return result;
}
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);
+ if ((sink = priv->udpsink[0]))
+ 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);
+ if ((sink = priv->udpsink[0]))
+ 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