rtpbin: handle multiple encoder instances

Keep track of elements that are added to multiple sessions and make sure
we only add them to the rtpbin once and that we clean them when no
session refers to them anymore.

gst/rtpmanager/gstrtpbin.c

@@ -227,6 +227,9 @@ struct _GstRtpBinPrivate
 
   /* UNIX (ntp) time of last SR sync used */
   guint64 last_unix;
+
+  /* list of extra elements */
+  GList *elements;
 };
 
 /* signals and args */
@@ -666,10 +669,49 @@ no_demux:
   }
 }
 
+static gboolean
+bin_manage_element (GstRtpBin * bin, GstElement * element)
+{
+  GstRtpBinPrivate *priv = bin->priv;
+
+  if (g_list_find (priv->elements, element)) {
+    GST_DEBUG_OBJECT (bin, "requested element %p already in bin", element);
+  } else {
+    GST_DEBUG_OBJECT (bin, "adding requested element %p", element);
+    if (!gst_bin_add (GST_BIN_CAST (bin), element))
+      goto add_failed;
+    if (!gst_element_sync_state_with_parent (element))
+      GST_WARNING_OBJECT (bin, "unable to sync element state with rtpbin");
+  }
+  /* we add the element multiple times, each we need an equal number of
+   * removes to really remove the element from the bin */
+  priv->elements = g_list_prepend (priv->elements, element);
+
+  return TRUE;
+
+  /* ERRORS */
+add_failed:
+  {
+    GST_WARNING_OBJECT (bin, "unable to add element");
+    return FALSE;
+  }
+}
+
 static void
 remove_bin_element (GstElement * element, GstRtpBin * bin)
 {
-  gst_bin_remove (GST_BIN_CAST (bin), element);
+  GstRtpBinPrivate *priv = bin->priv;
+  GList *find;
+
+  find = g_list_find (priv->elements, element);
+  if (find) {
+    priv->elements = g_list_delete_link (priv->elements, find);
+
+    if (!g_list_find (priv->elements, element))
+      gst_bin_remove (GST_BIN_CAST (bin), element);
+    else
+      gst_object_unref (element);
+  }
 }
 
 /* called with RTP_BIN_LOCK */
@@ -2532,31 +2574,21 @@ static GstElement *
 session_request_encoder (GstRtpBinSession * session, guint signal)
 {
   GstElement *encoder = NULL;
+  GstRtpBin *bin = session->bin;
 
-  g_signal_emit (session->bin, gst_rtp_bin_signals[signal], 0, session->id,
+  g_signal_emit (bin, gst_rtp_bin_signals[signal], 0, session->id, &encoder);
-      &encoder);
 
   if (encoder) {
-    if (g_slist_find (session->encoders, encoder)) {
+    if (!bin_manage_element (bin, encoder))
-      GST_DEBUG_OBJECT (session->bin, "requested encoder %p already in bin",
+      goto manage_failed;
-          encoder);
+    session->encoders = g_slist_prepend (session->encoders, encoder);
-    } else {
-      GST_DEBUG_OBJECT (session->bin, "adding requested encoder %p", encoder);
-      if (!gst_bin_add (GST_BIN_CAST (session->bin), encoder))
-        goto add_failed;
-      if (!gst_element_sync_state_with_parent (encoder))
-        GST_WARNING_OBJECT (session->bin,
-            "unable to sync encoder state with rtpbin");
-      session->encoders = g_slist_append (session->encoders, encoder);
-    }
   }
-
   return encoder;
 
   /* ERRORS */
-add_failed:
+manage_failed:
   {
-    GST_WARNING_OBJECT (session->bin, "unable to add encoder");
+    GST_WARNING_OBJECT (bin, "unable to manage encoder");
     gst_object_unref (encoder);
     return NULL;
   }
@@ -2566,31 +2598,21 @@ static GstElement *
 session_request_decoder (GstRtpBinSession * session, guint signal)
 {
   GstElement *decoder = NULL;
+  GstRtpBin *bin = session->bin;
 
-  g_signal_emit (session->bin, gst_rtp_bin_signals[signal], 0, session->id,
+  g_signal_emit (bin, gst_rtp_bin_signals[signal], 0, session->id, &decoder);
-      &decoder);
 
   if (decoder) {
-    if (g_slist_find (session->decoders, decoder)) {
+    if (!bin_manage_element (bin, decoder))
-      GST_DEBUG_OBJECT (session->bin, "requested decoder %p already in bin",
+      goto manage_failed;
-          decoder);
+    session->decoders = g_slist_prepend (session->decoders, decoder);
-    } else {
-      GST_DEBUG_OBJECT (session->bin, "adding requested decoder %p", decoder);
-      if (!gst_bin_add (GST_BIN_CAST (session->bin), decoder))
-        goto add_failed;
-      if (!gst_element_sync_state_with_parent (decoder))
-        GST_WARNING_OBJECT (session->bin,
-            "unable to sync decoder state with rtpbin");
-      session->decoders = g_slist_append (session->decoders, decoder);
-    }
   }
-
   return decoder;
 
   /* ERRORS */
-add_failed:
+manage_failed:
   {
-    GST_WARNING_OBJECT (session->bin, "unable to add decoder");
+    GST_WARNING_OBJECT (bin, "unable to manage decoder");
     gst_object_unref (decoder);
     return NULL;
   }

tests/check/elements/rtpbin.c

@@ -431,43 +431,78 @@ GST_START_TEST (test_request_pad_by_template_name)
 GST_END_TEST;
 
 static GstElement *
-encoder_cb (GstElement * rtpbin, guint sessid, gpointer user_data)
+encoder_cb (GstElement * rtpbin, guint sessid, GstElement * bin)
 {
-  GstElement *bin;
   GstPad *srcpad, *sinkpad;
 
   fail_unless (sessid == 2);
 
-  bin = gst_bin_new ("rtpenc");
   GST_DEBUG ("making encoder");
-
   sinkpad = gst_ghost_pad_new_no_target ("rtp_sink_2", GST_PAD_SINK);
   srcpad = gst_ghost_pad_new_no_target ("rtp_src_2", GST_PAD_SRC);
 
   gst_element_add_pad (bin, sinkpad);
   gst_element_add_pad (bin, srcpad);
 
-  return bin;
+  return gst_object_ref (bin);
+}
+
+static GstElement *
+encoder_cb2 (GstElement * rtpbin, guint sessid, GstElement * bin)
+{
+  GstPad *srcpad, *sinkpad;
+
+  fail_unless (sessid == 3);
+
+  GST_DEBUG ("making encoder");
+  sinkpad = gst_ghost_pad_new_no_target ("rtp_sink_3", GST_PAD_SINK);
+  srcpad = gst_ghost_pad_new_no_target ("rtp_src_3", GST_PAD_SRC);
+
+  gst_element_add_pad (bin, sinkpad);
+  gst_element_add_pad (bin, srcpad);
+
+  return gst_object_ref (bin);
 }
 
 GST_START_TEST (test_encoder)
 {
-  GstElement *rtpbin;
+  GstElement *rtpbin, *bin;
-  GstPad *rtp_sink1;
+  GstPad *rtp_sink1, *rtp_sink2;
+  gulong id;
+
+  bin = gst_bin_new ("rtpenc");
 
   rtpbin = gst_element_factory_make ("rtpbin", "rtpbin");
 
-  g_signal_connect (rtpbin, "request-rtp-encoder", (GCallback) encoder_cb,
+  id = g_signal_connect (rtpbin, "request-rtp-encoder", (GCallback) encoder_cb,
-      NULL);
+      bin);
 
   rtp_sink1 = gst_element_get_request_pad (rtpbin, "send_rtp_sink_2");
   fail_unless (rtp_sink1 != NULL);
   fail_unless_equals_string (GST_PAD_NAME (rtp_sink1), "send_rtp_sink_2");
   ASSERT_OBJECT_REFCOUNT (rtp_sink1, "rtp_sink1", 2);
 
+  g_signal_handler_disconnect (rtpbin, id);
+
+  id = g_signal_connect (rtpbin, "request-rtp-encoder", (GCallback) encoder_cb2,
+      bin);
+
+  rtp_sink2 = gst_element_get_request_pad (rtpbin, "send_rtp_sink_3");
+  fail_unless (rtp_sink2 != NULL);
+
+  /* remove the session */
+  gst_element_release_request_pad (rtpbin, rtp_sink1);
   gst_object_unref (rtp_sink1);
 
+  gst_element_release_request_pad (rtpbin, rtp_sink2);
+  gst_object_unref (rtp_sink2);
+
+  /* nothing left anymore now */
+  fail_unless (rtpbin->numsinkpads == 0);
+  fail_unless (rtpbin->numsrcpads == 0);
+
   gst_object_unref (rtpbin);
+  gst_object_unref (bin);
 }
 
 GST_END_TEST;