rtpbin: do better cleanup of the src ghostpads

Connect to the pad-removed signal of the ptdemux elements so that we remove the
ghostpads for them. Fixes cleanup when going to NULL as well as when releasing
the sinkpads.

Fixes #561752

2 files changed, 114 insertions, 14 deletions

diff --git a/gst/rtpmanager/gstrtpbin.c b/gst/rtpmanager/gstrtpbin.c
index ea391cf1a..c09b0ab90 100644
--- a/gst/rtpmanager/gstrtpbin.c
+++ b/gst/rtpmanager/gstrtpbin.c
@@ -285,16 +285,15 @@ struct _GstRtpBinStream
   gulong buffer_ptreq_sig;
   gulong buffer_ntpstop_sig;
 
   /* the PT demuxer of the SSRC */
   GstElement *demux;
   gulong demux_newpad_sig;
+  gulong demux_padremoved_sig;
   gulong demux_ptreq_sig;
   gulong demux_pt_change_sig;
-  /* ghostpads from the ptdemuxer */
-  GSList *pads;
 
   /* if we have calculated a valid unix_delta for this stream */
   gboolean have_sync;
   /* mapping to local RTP and NTP time */
   gint64 unix_delta;
 };
@@ -1149,36 +1148,31 @@ no_demux:
 }
 
 static void
 free_stream (GstRtpBinStream * stream)
 {
   GstRtpBinSession *session;
-  GSList *walk;
 
   session = stream->session;
 
   g_signal_handler_disconnect (stream->demux, stream->demux_newpad_sig);
   g_signal_handler_disconnect (stream->demux, stream->demux_ptreq_sig);
   g_signal_handler_disconnect (stream->buffer, stream->buffer_handlesync_sig);
   g_signal_handler_disconnect (stream->buffer, stream->buffer_ptreq_sig);
   g_signal_handler_disconnect (stream->buffer, stream->buffer_ntpstop_sig);
 
   gst_element_set_state (stream->demux, GST_STATE_NULL);
   gst_element_set_state (stream->buffer, GST_STATE_NULL);
 
+  /* now remove this signal, we need this while going to NULL because it to 
+   * do some cleanups */
+  g_signal_handler_disconnect (stream->demux, stream->demux_padremoved_sig);
+
   gst_bin_remove (GST_BIN_CAST (session->bin), stream->buffer);
   gst_bin_remove (GST_BIN_CAST (session->bin), stream->demux);
 
-  for (walk = stream->pads; walk; walk = g_slist_next (walk)) {
-    GstPad *gpad = GST_PAD_CAST (walk->data);
-
-    gst_pad_set_active (gpad, FALSE);
-    gst_element_remove_pad (GST_ELEMENT_CAST (session->bin), gpad);
-  }
-  g_slist_free (stream->pads);
-
   g_free (stream);
 }
 
 /* GObject vmethods */
 static void gst_rtp_bin_dispose (GObject * object);
 static void gst_rtp_bin_finalize (GObject * object);
@@ -1741,30 +1735,49 @@ new_payload_found (GstElement * element, guint pt, GstPad * pad,
   klass = GST_ELEMENT_GET_CLASS (rtpbin);
   templ = gst_element_class_get_pad_template (klass, "recv_rtp_src_%d_%d_%d");
   padname = g_strdup_printf ("recv_rtp_src_%d_%u_%d",
       stream->session->id, stream->ssrc, pt);
   gpad = gst_ghost_pad_new_from_template (padname, pad, templ);
   g_free (padname);
+  g_object_set_data (G_OBJECT (pad), "GstRTPBin.ghostpad", gpad);
 
   gst_pad_set_caps (gpad, GST_PAD_CAPS (pad));
   gst_pad_set_active (gpad, TRUE);
   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), gpad);
-
-  stream->pads = g_slist_prepend (stream->pads, gpad);
-
   GST_RTP_BIN_SHUTDOWN_UNLOCK (rtpbin);
 
   return;
 
 shutdown:
   {
     GST_DEBUG ("ignoring, we are shutting down");
     return;
   }
 }
 
+static void
+payload_pad_removed (GstElement * element, GstPad * pad,
+    GstRtpBinStream * stream)
+{
+  GstRtpBin *rtpbin;
+  GstPad *gpad;
+
+  rtpbin = stream->bin;
+
+  GST_DEBUG ("payload pad removed");
+
+  GST_RTP_BIN_DYN_LOCK (rtpbin);
+  if ((gpad = g_object_get_data (G_OBJECT (pad), "GstRTPBin.ghostpad"))) {
+    g_object_set_data (G_OBJECT (pad), "GstRTPBin.ghostpad", NULL);
+
+    gst_pad_set_active (gpad, FALSE);
+    gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin), gpad);
+  }
+  GST_RTP_BIN_DYN_UNLOCK (rtpbin);
+}
+
 static GstCaps *
 pt_map_requested (GstElement * element, guint pt, GstRtpBinSession * session)
 {
   GstRtpBin *rtpbin;
   GstCaps *caps;
 
@@ -1866,12 +1879,15 @@ new_ssrc_pad_found (GstElement * element, guint ssrc, GstPad * pad,
       "handle-sync", (GCallback) gst_rtp_bin_handle_sync, stream);
 
   /* connect to the new-pad signal of the payload demuxer, this will expose the
    * new pad by ghosting it. */
   stream->demux_newpad_sig = g_signal_connect (stream->demux,
       "new-payload-type", (GCallback) new_payload_found, stream);
+  stream->demux_padremoved_sig = g_signal_connect (stream->demux,
+      "pad-removed", (GCallback) payload_pad_removed, stream);
+
   /* connect to the request-pt-map signal. This signal will be emited by the
    * demuxer so that it can apply a proper caps on the buffers for the
    * depayloaders. */
   stream->demux_ptreq_sig = g_signal_connect (stream->demux,
       "request-pt-map", (GCallback) pt_map_requested, session);
 
diff --git a/tests/check/elements/rtpbin.c b/tests/check/elements/rtpbin.c
index bc30c9181..8764da54a 100644
--- a/tests/check/elements/rtpbin.c
+++ b/tests/check/elements/rtpbin.c
@@ -303,21 +303,105 @@ GST_START_TEST (test_cleanup_recv)
 
   clean_data (&data);
 }
 
 GST_END_TEST;
 
+GST_START_TEST (test_cleanup_recv2)
+{
+  GstElement *rtpbin;
+  GstPad *rtp_sink;
+  CleanupData data;
+  GstStateChangeReturn ret;
+  GstFlowReturn res;
+  GstBuffer *buffer;
+  gint count = 2;
+
+  init_data (&data);
+
+  rtpbin = gst_element_factory_make ("gstrtpbin", "rtpbin");
+
+  g_signal_connect (rtpbin, "pad-added", (GCallback) pad_added_cb, &data);
+  g_signal_connect (rtpbin, "pad-removed", (GCallback) pad_removed_cb, &data);
+
+  ret = gst_element_set_state (rtpbin, GST_STATE_PLAYING);
+  fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
+
+  /* request session 0 */
+  rtp_sink = gst_element_get_request_pad (rtpbin, "recv_rtp_sink_0");
+  fail_unless (rtp_sink != NULL);
+  ASSERT_OBJECT_REFCOUNT (rtp_sink, "rtp_sink", 2);
+
+  while (count--) {
+    /* no sourcepads are created yet */
+    fail_unless (rtpbin->numsinkpads == 1);
+    fail_unless (rtpbin->numsrcpads == 0);
+
+    buffer = make_rtp_packet (&data);
+    res = gst_pad_chain (rtp_sink, buffer);
+    GST_DEBUG ("res %d, %s\n", res, gst_flow_get_name (res));
+    fail_unless (res == GST_FLOW_OK);
+
+    buffer = make_rtp_packet (&data);
+    res = gst_pad_chain (rtp_sink, buffer);
+    GST_DEBUG ("res %d, %s\n", res, gst_flow_get_name (res));
+    fail_unless (res == GST_FLOW_OK);
+
+    /* we wait for the new pad to appear now */
+    g_mutex_lock (data.lock);
+    while (!data.pad_added)
+      g_cond_wait (data.cond, data.lock);
+    g_mutex_unlock (data.lock);
+
+    /* sourcepad created now */
+    fail_unless (rtpbin->numsinkpads == 1);
+    fail_unless (rtpbin->numsrcpads == 1);
+
+    /* change state */
+    ret = gst_element_set_state (rtpbin, GST_STATE_NULL);
+    fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
+
+    /* pad should be gone now */
+    g_mutex_lock (data.lock);
+    while (data.pad_added)
+      g_cond_wait (data.cond, data.lock);
+    g_mutex_unlock (data.lock);
+
+    /* back to playing for the next round */
+    ret = gst_element_set_state (rtpbin, GST_STATE_PLAYING);
+    fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
+  }
+
+  /* remove the session */
+  gst_element_release_request_pad (rtpbin, rtp_sink);
+  gst_object_unref (rtp_sink);
+
+  /* nothing left anymore now */
+  fail_unless (rtpbin->numsinkpads == 0);
+  fail_unless (rtpbin->numsrcpads == 0);
+
+  ret = gst_element_set_state (rtpbin, GST_STATE_NULL);
+  fail_unless (ret == GST_STATE_CHANGE_SUCCESS);
+
+  gst_object_unref (rtpbin);
+
+  clean_data (&data);
+}
+
+GST_END_TEST;
+
 Suite *
 gstrtpbin_suite (void)
 {
   Suite *s = suite_create ("gstrtpbin");
   TCase *tc_chain = tcase_create ("general");
 
   suite_add_tcase (s, tc_chain);
   tcase_add_test (tc_chain, test_cleanup_send);
   tcase_add_test (tc_chain, test_cleanup_recv);
+  tcase_add_test (tc_chain, test_cleanup_recv2);
 
   return s;
 }
 
 int
 main (int argc, char **argv)