rtsp-stream: Remove the multicast group udp sources when removing from the bin
[platform/upstream/gstreamer.git] / gst / rtsp-server / rtsp-stream.c
index 8382d3e..68d7bec 100644 (file)
 #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;
@@ -79,6 +88,11 @@ struct _GstRTSPStreamPrivate
   /* the RTPSession object */
   GObject *session;
 
+  /* SRTP encoder/decoder */
+  GstElement *srtpenc;
+  GstElement *srtpdec;
+  GHashTable *keys;
+
   /* sinks used for sending and receiving RTP and RTCP over ipv4, they share
    * sockets */
   GstElement *udpsrc_v4[2];
@@ -119,6 +133,12 @@ struct _GstRTSPStreamPrivate
   /* transports we stream to */
   guint n_active;
   GList *transports;
+  guint transports_cookie;
+  GList *tr_cache;
+  guint tr_cache_cookie;
+
+  /* UDP sources for UDP multicast transports */
+  GList *transport_sources;
 
   gint dscp_qos;
 
@@ -141,6 +161,13 @@ enum
   PROP_LAST
 };
 
+enum
+{
+  SIGNAL_NEW_RTP_ENCODER,
+  SIGNAL_NEW_RTCP_ENCODER,
+  SIGNAL_LAST
+};
+
 GST_DEBUG_CATEGORY_STATIC (rtsp_stream_debug);
 #define GST_CAT_DEFAULT rtsp_stream_debug
 
@@ -153,6 +180,8 @@ static void gst_rtsp_stream_set_property (GObject * object, guint propid,
 
 static void gst_rtsp_stream_finalize (GObject * obj);
 
+static guint gst_rtsp_stream_signals[SIGNAL_LAST] = { 0 };
+
 G_DEFINE_TYPE (GstRTSPStream, gst_rtsp_stream, G_TYPE_OBJECT);
 
 static void
@@ -183,6 +212,16 @@ gst_rtsp_stream_class_init (GstRTSPStreamClass * klass)
           "Allowed lower transport protocols", GST_TYPE_RTSP_LOWER_TRANS,
           DEFAULT_PROTOCOLS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  gst_rtsp_stream_signals[SIGNAL_NEW_RTP_ENCODER] =
+      g_signal_new ("new-rtp-encoder", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, 0, NULL, NULL, g_cclosure_marshal_generic,
+      G_TYPE_NONE, 1, GST_TYPE_ELEMENT);
+
+  gst_rtsp_stream_signals[SIGNAL_NEW_RTCP_ENCODER] =
+      g_signal_new ("new-rtcp-encoder", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, 0, NULL, NULL, g_cclosure_marshal_generic,
+      G_TYPE_NONE, 1, GST_TYPE_ELEMENT);
+
   GST_DEBUG_CATEGORY_INIT (rtsp_stream_debug, "rtspstream", 0, "GstRTSPStream");
 
   ssrc_stream_map_key = g_quark_from_static_string ("GstRTSPServer.stream");
@@ -203,6 +242,9 @@ gst_rtsp_stream_init (GstRTSPStream * stream)
   priv->protocols = DEFAULT_PROTOCOLS;
 
   g_mutex_init (&priv->lock);
+
+  priv->keys = g_hash_table_new_full (g_direct_hash, g_direct_equal,
+      NULL, (GDestroyNotify) gst_caps_unref);
 }
 
 static void
@@ -234,6 +276,8 @@ gst_rtsp_stream_finalize (GObject * obj)
   g_free (priv->control);
   g_mutex_clear (&priv->lock);
 
+  g_hash_table_unref (priv->keys);
+
   G_OBJECT_CLASS (gst_rtsp_stream_parent_class)->finalize (obj);
 }
 
@@ -288,7 +332,7 @@ gst_rtsp_stream_set_property (GObject * object, guint propid,
  * Create a new media stream with index @idx that handles RTP data on
  * @srcpad and has a payloader element @payloader.
  *
- * Returns: a new #GstRTSPStream
+ * Returns: (transfer full): a new #GstRTSPStream
  */
 GstRTSPStream *
 gst_rtsp_stream_new (guint idx, GstElement * payloader, GstPad * srcpad)
@@ -370,7 +414,7 @@ gst_rtsp_stream_get_srcpad (GstRTSPStream * stream)
  *
  * Get the control string to identify this stream.
  *
- * Returns: (transfer full): the control string. free after usage.
+ * Returns: (transfer full): the control string. g_free() after usage.
  */
 gchar *
 gst_rtsp_stream_get_control (GstRTSPStream * stream)
@@ -563,7 +607,7 @@ gst_rtsp_stream_get_dscp_qos (GstRTSPStream * stream)
 /**
  * gst_rtsp_stream_is_transport_supported:
  * @stream: a #GstRTSPStream
- * @transport: a #GstRTSPTransport
+ * @transport: (transfer none): a #GstRTSPTransport
  *
  * Check if @transport can be handled by stream
  *
@@ -597,16 +641,19 @@ gst_rtsp_stream_is_transport_supported (GstRTSPStream * stream,
 unsupported_transmode:
   {
     GST_DEBUG ("unsupported transport mode %d", transport->trans);
+    g_mutex_unlock (&priv->lock);
     return FALSE;
   }
 unsupported_profile:
   {
     GST_DEBUG ("unsupported profile %d", transport->profile);
+    g_mutex_unlock (&priv->lock);
     return FALSE;
   }
 unsupported_ltrans:
   {
     GST_DEBUG ("unsupported lower transport %d", transport->lower_transport);
+    g_mutex_unlock (&priv->lock);
     return FALSE;
   }
 }
@@ -708,7 +755,7 @@ gst_rtsp_stream_get_protocols (GstRTSPStream * stream)
 /**
  * gst_rtsp_stream_set_address_pool:
  * @stream: a #GstRTSPStream
- * @pool: a #GstRTSPAddressPool
+ * @pool: (transfer none): a #GstRTSPAddressPool
  *
  * configure @pool to be used as the address pool of @stream.
  */
@@ -770,8 +817,9 @@ gst_rtsp_stream_get_address_pool (GstRTSPStream * stream)
  *
  * Get the multicast address of @stream for @family.
  *
- * Returns: 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,
@@ -788,10 +836,10 @@ gst_rtsp_stream_get_multicast_address (GstRTSPStream * stream,
 
   if (family == G_SOCKET_FAMILY_IPV6) {
     flags = GST_RTSP_ADDRESS_FLAG_IPV6;
-    addrp = &priv->addr_v4;
+    addrp = &priv->addr_v6;
   } else {
     flags = GST_RTSP_ADDRESS_FLAG_IPV4;
-    addrp = &priv->addr_v6;
+    addrp = &priv->addr_v4;
   }
 
   g_mutex_lock (&priv->lock);
@@ -835,8 +883,8 @@ no_address:
  *
  * 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,
@@ -866,9 +914,9 @@ gst_rtsp_stream_reserve_address (GstRTSPStream * stream,
   }
 
   if (family == G_SOCKET_FAMILY_IPV6)
-    addrp = &priv->addr_v4;
-  else
     addrp = &priv->addr_v6;
+  else
+    addrp = &priv->addr_v4;
 
   g_mutex_lock (&priv->lock);
   if (*addrp == NULL) {
@@ -1048,10 +1096,10 @@ again:
   g_object_set (G_OBJECT (udpsrc0), "socket", rtp_socket, NULL);
   g_object_set (G_OBJECT (udpsrc1), "socket", rtcp_socket, NULL);
 
-  ret = gst_element_set_state (udpsrc0, GST_STATE_PAUSED);
+  ret = gst_element_set_state (udpsrc0, GST_STATE_READY);
   if (ret == GST_STATE_CHANGE_FAILURE)
     goto element_error;
-  ret = gst_element_set_state (udpsrc1, GST_STATE_PAUSED);
+  ret = gst_element_set_state (udpsrc1, GST_STATE_READY);
   if (ret == GST_STATE_CHANGE_FAILURE)
     goto element_error;
 
@@ -1101,6 +1149,7 @@ again:
   udpsrc_out[1] = udpsrc1;
   udpsink_out[0] = udpsink0;
   udpsink_out[1] = udpsink1;
+
   server_port_out->min = rtpport;
   server_port_out->max = rtcpport;
 
@@ -1439,6 +1488,14 @@ on_timeout (GObject * session, GObject * source, GstRTSPStream * stream)
   }
 }
 
+static void
+clear_tr_cache (GstRTSPStreamPrivate * priv)
+{
+  g_list_foreach (priv->tr_cache, (GFunc) g_object_unref, NULL);
+  g_list_free (priv->tr_cache);
+  priv->tr_cache = NULL;
+}
+
 static GstFlowReturn
 handle_new_sample (GstAppSink * sink, gpointer user_data)
 {
@@ -1447,6 +1504,7 @@ handle_new_sample (GstAppSink * sink, gpointer user_data)
   GstSample *sample;
   GstBuffer *buffer;
   GstRTSPStream *stream;
+  gboolean is_rtp;
 
   sample = gst_app_sink_pull_sample (sink);
   if (!sample)
@@ -1456,18 +1514,28 @@ handle_new_sample (GstAppSink * sink, gpointer user_data)
   priv = stream->priv;
   buffer = gst_sample_get_buffer (sample);
 
+  is_rtp = GST_ELEMENT_CAST (sink) == priv->appsink[0];
+
   g_mutex_lock (&priv->lock);
-  for (walk = priv->transports; walk; walk = g_list_next (walk)) {
+  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_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 (GST_ELEMENT_CAST (sink) == priv->appsink[0]) {
+    if (is_rtp) {
       gst_rtsp_stream_transport_send_rtp (tr, buffer);
     } else {
       gst_rtsp_stream_transport_send_rtcp (tr, buffer);
     }
   }
-  g_mutex_unlock (&priv->lock);
-
   gst_sample_unref (sample);
 
   return GST_FLOW_OK;
@@ -1479,11 +1547,121 @@ static GstAppSinkCallbacks sink_cb = {
   handle_new_sample,
 };
 
+static GstElement *
+get_rtp_encoder (GstRTSPStream * stream, guint session)
+{
+  GstRTSPStreamPrivate *priv = stream->priv;
+
+  if (priv->srtpenc == NULL) {
+    gchar *name;
+
+    name = g_strdup_printf ("srtpenc_%u", session);
+    priv->srtpenc = gst_element_factory_make ("srtpenc", name);
+    g_free (name);
+
+    g_object_set (priv->srtpenc, "random-key", TRUE, NULL);
+  }
+  return gst_object_ref (priv->srtpenc);
+}
+
+static GstElement *
+request_rtp_encoder (GstElement * rtpbin, guint session, GstRTSPStream * stream)
+{
+  GstRTSPStreamPrivate *priv = stream->priv;
+  GstElement *oldenc, *enc;
+  GstPad *pad;
+  gchar *name;
+
+  if (priv->idx != session)
+    return NULL;
+
+  GST_DEBUG_OBJECT (stream, "make RTP encoder for session %u", session);
+
+  oldenc = priv->srtpenc;
+  enc = get_rtp_encoder (stream, session);
+  name = g_strdup_printf ("rtp_sink_%d", session);
+  pad = gst_element_get_request_pad (enc, name);
+  g_free (name);
+  gst_object_unref (pad);
+
+  if (oldenc == NULL)
+    g_signal_emit (stream, gst_rtsp_stream_signals[SIGNAL_NEW_RTP_ENCODER], 0,
+        enc);
+
+  return enc;
+}
+
+static GstElement *
+request_rtcp_encoder (GstElement * rtpbin, guint session,
+    GstRTSPStream * stream)
+{
+  GstRTSPStreamPrivate *priv = stream->priv;
+  GstElement *oldenc, *enc;
+  GstPad *pad;
+  gchar *name;
+
+  if (priv->idx != session)
+    return NULL;
+
+  GST_DEBUG_OBJECT (stream, "make RTCP encoder for session %u", session);
+
+  oldenc = priv->srtpenc;
+  enc = get_rtp_encoder (stream, session);
+  name = g_strdup_printf ("rtcp_sink_%d", session);
+  pad = gst_element_get_request_pad (enc, name);
+  g_free (name);
+  gst_object_unref (pad);
+
+  if (oldenc == NULL)
+    g_signal_emit (stream, gst_rtsp_stream_signals[SIGNAL_NEW_RTCP_ENCODER], 0,
+        enc);
+
+  return enc;
+}
+
+static GstCaps *
+request_key (GstElement * srtpdec, guint ssrc, GstRTSPStream * stream)
+{
+  GstRTSPStreamPrivate *priv = stream->priv;
+  GstCaps *caps;
+
+  GST_DEBUG ("request key %08x", ssrc);
+
+  g_mutex_lock (&priv->lock);
+  if ((caps = g_hash_table_lookup (priv->keys, GINT_TO_POINTER (ssrc))))
+    gst_caps_ref (caps);
+  g_mutex_unlock (&priv->lock);
+
+  return caps;
+}
+
+static GstElement *
+request_rtcp_decoder (GstElement * rtpbin, guint session,
+    GstRTSPStream * stream)
+{
+  GstRTSPStreamPrivate *priv = stream->priv;
+
+  if (priv->idx != session)
+    return NULL;
+
+  if (priv->srtpdec == NULL) {
+    gchar *name;
+
+    name = g_strdup_printf ("srtpdec_%u", session);
+    priv->srtpdec = gst_element_factory_make ("srtpdec", name);
+    g_free (name);
+
+    g_signal_connect (priv->srtpdec, "request-key",
+        (GCallback) request_key, stream);
+  }
+  return gst_object_ref (priv->srtpdec);
+}
+
 /**
  * gst_rtsp_stream_join_bin:
  * @stream: a #GstRTSPStream
- * @bin: a #GstBin to join
- * @rtpbin: a rtpbin element in @bin
+ * @bin: (transfer none): a #GstBin to join
+ * @rtpbin: (transfer none): a rtpbin element in @bin
  * @state: the target state of the new elements
  *
  * Join the #GstBin @bin that contains the element @rtpbin.
@@ -1525,6 +1703,17 @@ gst_rtsp_stream_join_bin (GstRTSPStream * stream, GstBin * bin,
   /* update the dscp qos field in the sinks */
   update_dscp_qos (stream);
 
+  if (priv->profiles & GST_RTSP_PROFILE_SAVP
+      || priv->profiles & GST_RTSP_PROFILE_SAVPF) {
+    /* For SRTP */
+    g_signal_connect (rtpbin, "request-rtp-encoder",
+        (GCallback) request_rtp_encoder, stream);
+    g_signal_connect (rtpbin, "request-rtcp-encoder",
+        (GCallback) request_rtcp_encoder, stream);
+    g_signal_connect (rtpbin, "request-rtcp-decoder",
+        (GCallback) request_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);
@@ -1714,7 +1903,7 @@ gst_rtsp_stream_join_bin (GstRTSPStream * stream, GstBin * bin,
   }
 
   /* be notified of caps changes */
-  priv->caps_sig = g_signal_connect (priv->send_rtp_sink, "notify::caps",
+  priv->caps_sig = g_signal_connect (priv->send_src[0], "notify::caps",
       (GCallback) caps_notify, stream);
 
   priv->is_joined = TRUE;
@@ -1747,8 +1936,8 @@ link_failed:
 /**
  * gst_rtsp_stream_leave_bin:
  * @stream: a #GstRTSPStream
- * @bin: a #GstBin
- * @rtpbin: a rtpbin #GstElement
+ * @bin: (transfer none): a #GstBin
+ * @rtpbin: (transfer none): a rtpbin #GstElement
  *
  * Remove the elements of @stream from @bin.
  *
@@ -1760,6 +1949,7 @@ gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
 {
   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);
@@ -1774,10 +1964,12 @@ gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
   /* all transports must be removed by now */
   g_return_val_if_fail (priv->transports == NULL, FALSE);
 
+  clear_tr_cache (priv);
+
   GST_INFO ("stream %p leaving bin", stream);
 
   gst_pad_unlink (priv->srcpad, priv->send_rtp_sink);
-  g_signal_handler_disconnect (priv->send_rtp_sink, priv->caps_sig);
+  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;
@@ -1808,6 +2000,18 @@ gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
       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])
@@ -1834,6 +2038,14 @@ gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
     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;
 
@@ -1847,6 +2059,9 @@ gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
     gst_caps_unref (priv->caps);
   priv->caps = NULL;
 
+  if (priv->srtpenc)
+    gst_object_unref (priv->srtpenc);
+
   priv->is_joined = FALSE;
   g_mutex_unlock (&priv->lock);
 
@@ -1854,6 +2069,7 @@ gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
 
 was_not_joined:
   {
+    g_mutex_unlock (&priv->lock);
     return TRUE;
   }
 }
@@ -1863,7 +2079,7 @@ was_not_joined:
  * @stream: a #GstRTSPStream
  * @rtptime: (allow-none): result RTP timestamp
  * @seq: (allow-none): result RTP seqnum
- * @clock_rate: the clock rate
+ * @clock_rate: (allow-none): the clock rate
  * @running_time: (allow-none): result running-time
  *
  * Retrieve the current rtptime, seq and running-time. This is used to
@@ -1877,6 +2093,7 @@ gst_rtsp_stream_get_rtpinfo (GstRTSPStream * stream,
     GstClockTime * running_time)
 {
   GstRTSPStreamPrivate *priv;
+  GstStructure *stats;
   GObjectClass *payobjclass;
 
   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
@@ -1886,27 +2103,52 @@ gst_rtsp_stream_get_rtpinfo (GstRTSPStream * stream,
   payobjclass = G_OBJECT_GET_CLASS (priv->payloader);
 
   g_mutex_lock (&priv->lock);
-  if (seq && g_object_class_find_property (payobjclass, "seqnum"))
-    g_object_get (priv->payloader, "seqnum", seq, NULL);
 
-  if (rtptime && g_object_class_find_property (payobjclass, "timestamp"))
-    g_object_get (priv->payloader, "timestamp", rtptime, NULL);
+  if (g_object_class_find_property (payobjclass, "stats")) {
+    g_object_get (priv->payloader, "stats", &stats, NULL);
+    if (stats == NULL)
+      goto no_stats;
 
-  if (running_time
-      && g_object_class_find_property (payobjclass, "running-time"))
-    g_object_get (priv->payloader, "running-time", running_time, NULL);
+    if (seq)
+      gst_structure_get_uint (stats, "seqnum", seq);
 
-  if (clock_rate && priv->caps) {
-    GstStructure *s;
+    if (rtptime)
+      gst_structure_get_uint (stats, "timestamp", rtptime);
 
-    s = gst_caps_get_structure (priv->caps, 0);
-    if (!gst_structure_get_int (s, "clock-rate", (gint *) clock_rate))
-      if (running_time)
+    if (running_time)
+      gst_structure_get_clock_time (stats, "running-time", running_time);
+
+    if (clock_rate) {
+      gst_structure_get_uint (stats, "clock-rate", clock_rate);
+      if (*clock_rate == 0 && running_time)
         *running_time = GST_CLOCK_TIME_NONE;
+    }
+    gst_structure_free (stats);
+  } else {
+    if (!g_object_class_find_property (payobjclass, "seqnum") ||
+        !g_object_class_find_property (payobjclass, "timestamp"))
+      goto no_stats;
+
+    if (seq)
+      g_object_get (priv->payloader, "seqnum", seq, NULL);
+
+    if (rtptime)
+      g_object_get (priv->payloader, "timestamp", rtptime, NULL);
+
+    if (running_time)
+      *running_time = GST_CLOCK_TIME_NONE;
   }
   g_mutex_unlock (&priv->lock);
 
   return TRUE;
+
+  /* ERRORS */
+no_stats:
+  {
+    GST_WARNING ("Could not get payloader stats");
+    g_mutex_unlock (&priv->lock);
+    return FALSE;
+  }
 }
 
 /**
@@ -1916,7 +2158,7 @@ gst_rtsp_stream_get_rtpinfo (GstRTSPStream * stream,
  * Retrieve the current caps of @stream.
  *
  * Returns: (transfer full): the #GstCaps of @stream. use gst_caps_unref()
- *    after usage.
+ * after usage.
  */
 GstCaps *
 gst_rtsp_stream_get_caps (GstRTSPStream * stream)
@@ -2012,6 +2254,7 @@ gst_rtsp_stream_recv_rtcp (GstRTSPStream * stream, GstBuffer * buffer)
     gst_object_unref (element);
   } else {
     ret = GST_FLOW_OK;
+    gst_buffer_unref (buffer);
   }
   return ret;
 }
@@ -2027,9 +2270,84 @@ update_transport (GstRTSPStream * stream, GstRTSPStreamTransport * trans,
   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;
@@ -2045,14 +2363,14 @@ update_transport (GstRTSPStream * stream, GstRTSPStreamTransport * trans,
       }
 
       if (add) {
-        GST_INFO ("adding %s:%d-%d", dest, min, max);
-        g_signal_emit_by_name (priv->udpsink[0], "add", dest, min, NULL);
-        g_signal_emit_by_name (priv->udpsink[1], "add", dest, max, NULL);
         if (ttl > 0) {
           GST_INFO ("setting ttl-mc %d", ttl);
           g_object_set (G_OBJECT (priv->udpsink[0]), "ttl-mc", ttl, NULL);
           g_object_set (G_OBJECT (priv->udpsink[1]), "ttl-mc", ttl, NULL);
         }
+        GST_INFO ("adding %s:%d-%d", dest, min, max);
+        g_signal_emit_by_name (priv->udpsink[0], "add", dest, min, NULL);
+        g_signal_emit_by_name (priv->udpsink[1], "add", dest, max, NULL);
         priv->transports = g_list_prepend (priv->transports, trans);
       } else {
         GST_INFO ("removing %s:%d-%d", dest, min, max);
@@ -2060,6 +2378,7 @@ update_transport (GstRTSPStream * stream, GstRTSPStreamTransport * trans,
         g_signal_emit_by_name (priv->udpsink[1], "remove", dest, max, NULL);
         priv->transports = g_list_remove (priv->transports, trans);
       }
+      priv->transports_cookie++;
       break;
     }
     case GST_RTSP_LOWER_TRANS_TCP:
@@ -2070,6 +2389,7 @@ update_transport (GstRTSPStream * stream, GstRTSPStreamTransport * trans,
         GST_INFO ("removing TCP %s", tr->destination);
         priv->transports = g_list_remove (priv->transports, trans);
       }
+      priv->transports_cookie++;
       break;
     default:
       goto unknown_transport;
@@ -2088,7 +2408,7 @@ unknown_transport:
 /**
  * gst_rtsp_stream_add_transport:
  * @stream: a #GstRTSPStream
- * @trans: a #GstRTSPStreamTransport
+ * @trans: (transfer none): a #GstRTSPStreamTransport
  *
  * Add the transport in @trans to @stream. The media of @stream will
  * then also be send to the values configured in @trans.
@@ -2121,7 +2441,7 @@ gst_rtsp_stream_add_transport (GstRTSPStream * stream,
 /**
  * gst_rtsp_stream_remove_transport:
  * @stream: a #GstRTSPStream
- * @trans: a #GstRTSPStreamTransport
+ * @trans: (transfer none): a #GstRTSPStreamTransport
  *
  * Remove the transport in @trans from @stream. The media of @stream will
  * not be sent to the values configured in @trans.
@@ -2152,6 +2472,43 @@ gst_rtsp_stream_remove_transport (GstRTSPStream * stream,
 }
 
 /**
+ * gst_rtsp_stream_update_crypto:
+ * @stream: a #GstRTSPStream
+ * @ssrc: the SSRC
+ * @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
+ * for @ssrc existed, it will be replaced. If @crypto is %NULL, it will
+ * be removed from @stream.
+ *
+ * Returns: %TRUE if @crypto could be updated
+ */
+gboolean
+gst_rtsp_stream_update_crypto (GstRTSPStream * stream,
+    guint ssrc, GstCaps * crypto)
+{
+  GstRTSPStreamPrivate *priv;
+
+  g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
+  g_return_val_if_fail (crypto == NULL || GST_IS_CAPS (crypto), FALSE);
+
+  priv = stream->priv;
+
+  GST_DEBUG_OBJECT (stream, "update key for %08x", ssrc);
+
+  g_mutex_lock (&priv->lock);
+  if (crypto)
+    g_hash_table_insert (priv->keys, GINT_TO_POINTER (ssrc),
+        gst_caps_ref (crypto));
+  else
+    g_hash_table_remove (priv->keys, GINT_TO_POINTER (ssrc));
+  g_mutex_unlock (&priv->lock);
+
+  return TRUE;
+}
+
+/**
  * gst_rtsp_stream_get_rtp_socket:
  * @stream: a #GstRTSPStream
  * @family: the socket family
@@ -2160,15 +2517,15 @@ gst_rtsp_stream_remove_transport (GstRTSPStream * stream,
  *
  * @stream must be joined to a bin.
  *
- * Returns: 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)
 {
   GstRTSPStreamPrivate *priv = GST_RTSP_STREAM_GET_PRIVATE (stream);
   GSocket *socket;
-  gchar *name;
+  const gchar *name;
 
   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
   g_return_val_if_fail (family == G_SOCKET_FAMILY_IPV4 ||
@@ -2194,15 +2551,15 @@ gst_rtsp_stream_get_rtp_socket (GstRTSPStream * stream, GSocketFamily family)
  *
  * @stream must be joined to a bin.
  *
- * Returns: 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 = GST_RTSP_STREAM_GET_PRIVATE (stream);
   GSocket *socket;
-  gchar *name;
+  const gchar *name;
 
   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
   g_return_val_if_fail (family == G_SOCKET_FAMILY_IPV4 ||
@@ -2223,7 +2580,7 @@ gst_rtsp_stream_get_rtcp_socket (GstRTSPStream * stream, GSocketFamily family)
  * gst_rtsp_stream_transport_filter:
  * @stream: a #GstRTSPStream
  * @func: (scope call) (allow-none): a callback
- * @user_data: user data passed to @func
+ * @user_data: (closure): user data passed to @func
  *
  * Call @func for each transport managed by @stream. The result value of @func
  * determines what happens to the transport. @func will be called with @stream
@@ -2250,25 +2607,43 @@ gst_rtsp_stream_transport_filter (GstRTSPStream * stream,
 {
   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);
@@ -2280,9 +2655,14 @@ gst_rtsp_stream_transport_filter (GstRTSPStream * stream,
       default:
         break;
     }
+    if (changed)
+      goto restart;
   }
   g_mutex_unlock (&priv->lock);
 
+  if (func)
+    g_hash_table_unref (visited);
+
   return result;
 }
 
@@ -2371,3 +2751,81 @@ gst_rtsp_stream_is_blocking (GstRTSPStream * stream)
 
   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;
+
+}