adaptivedemux: replace ghostpad with a standard pad

Handling the ghostpad and its internal pad was causing more issues
than helping because of their coupled activation/deactivation
actions.

As we have to install custom chain,event and query functions it is
better to use a floating sink pad internally in the demuxer and just
use those pad functions to push through a standard pad in the demuxer

https://bugzilla.gnome.org/show_bug.cgi?id=757951

ext/dash/gstdashdemux.c

@@ -827,7 +827,7 @@ gst_dash_demux_create_pad (GstDashDemux * demux, GstActiveStream * stream)
   }
 
   /* Create and activate new pads */
-  pad = gst_ghost_pad_new_no_target_from_template (name, tmpl);
+  pad = gst_pad_new_from_template (tmpl, name);
   g_free (name);
   gst_object_unref (tmpl);
 

ext/hls/gsthlsdemux.c

@@ -207,13 +207,10 @@ static GstPad *
 gst_hls_demux_create_pad (GstHLSDemux * hlsdemux)
 {
   gchar *name;
-  GstPadTemplate *tmpl;
   GstPad *pad;
 
   name = g_strdup_printf ("src_%u", hlsdemux->srcpad_counter++);
-  tmpl = gst_static_pad_template_get (&srctemplate);
+  pad = gst_pad_new_from_static_template (&srctemplate, name);
-  pad = gst_ghost_pad_new_no_target_from_template (name, tmpl);
-  gst_object_unref (tmpl);
   g_free (name);
 
   return pad;

ext/smoothstreaming/gstmssdemux.c

@@ -354,8 +354,7 @@ _create_pad (GstMssDemux * mssdemux, GstMssStream * manifeststream)
   }
 
   if (tmpl != NULL) {
-    srcpad =
+    srcpad = GST_PAD_CAST (gst_pad_new_from_template (tmpl, name));
-        GST_PAD_CAST (gst_ghost_pad_new_no_target_from_template (name, tmpl));
     g_free (name);
     gst_object_unref (tmpl);
   }

gst-libs/gst/adaptivedemux/gstadaptivedemux.c

@@ -1860,12 +1860,15 @@ gst_adaptive_demux_stream_data_received_default (GstAdaptiveDemux * demux,
 static GstFlowReturn
 _src_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 {
-  GstPad *srcpad = (GstPad *) parent;
+  GstAdaptiveDemuxStream *stream;
-  GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (srcpad);
+  GstAdaptiveDemux *demux;
-  GstAdaptiveDemux *demux = stream->demux;
+  GstAdaptiveDemuxClass *klass;
-  GstAdaptiveDemuxClass *klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
   GstFlowReturn ret = GST_FLOW_OK;
 
+  demux = GST_ADAPTIVE_DEMUX_CAST (parent);
+  stream = gst_pad_get_element_private (pad);
+  klass = GST_ADAPTIVE_DEMUX_GET_CLASS (demux);
+
   GST_MANIFEST_LOCK (demux);
 
   /* do not make any changes if the stream is cancelled */
@@ -2019,8 +2022,7 @@ gst_adaptive_demux_stream_fragment_download_finish (GstAdaptiveDemuxStream *
 static gboolean
 _src_event (GstPad * pad, GstObject * parent, GstEvent * event)
 {
-  GstPad *srcpad = GST_PAD_CAST (parent);
+  GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (pad);
-  GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (srcpad);
   GstAdaptiveDemux *demux = stream->demux;
 
   switch (GST_EVENT_TYPE (event)) {
@@ -2049,6 +2051,8 @@ _src_event (GstPad * pad, GstObject * parent, GstEvent * event)
 static gboolean
 _src_query (GstPad * pad, GstObject * parent, GstQuery * query)
 {
+  GstAdaptiveDemuxStream *stream = gst_pad_get_element_private (pad);
+
   switch (GST_QUERY_TYPE (query)) {
     case GST_QUERY_ALLOCATION:
       return FALSE;
@@ -2057,7 +2061,7 @@ _src_query (GstPad * pad, GstObject * parent, GstQuery * query)
       break;
   }
 
-  return gst_pad_query_default (pad, parent, query);
+  return gst_pad_peer_query (stream->pad, query);
 }
 
 /* must be called with manifest_lock taken.
@@ -2120,37 +2124,6 @@ gst_adaptive_demux_stream_wait_manifest_update (GstAdaptiveDemux * demux,
   return ret;
 }
 
-static gboolean
-_adaptive_demux_pad_remove_eos_sticky (GstPad * pad, GstEvent ** event,
-    gpointer udata)
-{
-  if (GST_EVENT_TYPE (*event) == GST_EVENT_EOS) {
-    gst_event_replace (event, NULL);
-    return FALSE;
-  }
-  return TRUE;
-}
-
-/* must be called with manifest_lock taken */
-static void
-gst_adaptive_demux_stream_clear_eos_and_flush_state (GstAdaptiveDemuxStream *
-    stream)
-{
-  GstPad *internal_pad;
-
-  internal_pad =
-      GST_PAD_CAST (gst_proxy_pad_get_internal (GST_PROXY_PAD (stream->pad)));
-  gst_pad_sticky_events_foreach (internal_pad,
-      _adaptive_demux_pad_remove_eos_sticky, NULL);
-  GST_OBJECT_FLAG_UNSET (internal_pad, GST_PAD_FLAG_EOS);
-  /* In case the stream is recovering from a flushing seek it is also needed
-   * to remove the flushing state from this pad. The flushing state is set
-   * because of the flow return propagating until the source element */
-  GST_PAD_UNSET_FLUSHING (internal_pad);
-
-  gst_object_unref (internal_pad);
-}
-
 static void
 gst_adaptive_demux_stream_queue_overrun (GstElement * queue, gpointer user_data)
 {
@@ -2216,7 +2189,7 @@ gst_adaptive_demux_stream_update_source (GstAdaptiveDemuxStream * stream,
     GstElement *queue;
     GstPadLinkReturn pad_link_ret;
     GObjectClass *gobject_class;
-    GstPad *internal_pad;
+    gchar *internal_name, *bin_name;
 
     /* Our src consists of a bin containing uri_handler -> queue2 . The
      * purpose of the queue2 is to allow the uri_handler to download an
@@ -2271,7 +2244,9 @@ gst_adaptive_demux_stream_update_source (GstAdaptiveDemuxStream * stream,
     }
 
     /* Source bin creation */
-    stream->src = gst_bin_new (NULL);
+    bin_name = g_strdup_printf ("srcbin-%s", GST_PAD_NAME (stream->pad));
+    stream->src = gst_bin_new (bin_name);
+    g_free (bin_name);
     if (stream->src == NULL) {
       gst_object_unref (queue);
       gst_object_unref (uri_handler);
@@ -2310,20 +2285,26 @@ gst_adaptive_demux_stream_update_source (GstAdaptiveDemuxStream * stream,
     gst_bin_add (GST_BIN_CAST (demux), stream->src);
     stream->src_srcpad = gst_element_get_static_pad (stream->src, "src");
 
-    gst_ghost_pad_set_target (GST_GHOST_PAD_CAST (stream->pad),
+    /* set up our internal floating pad to drop all events from
-        stream->src_srcpad);
-
-    /* set up our internal pad to drop all events from
      * the http src we don't care about. On the chain function
-     * we just push the buffer forward, but this way dash can get
+     * we just push the buffer forward */
-     * the flow return from downstream */
+    internal_name = g_strdup_printf ("internal-%s", GST_PAD_NAME (stream->pad));
-    internal_pad =
+    stream->internal_pad = gst_pad_new (internal_name, GST_PAD_SINK);
-        GST_PAD_CAST (gst_proxy_pad_get_internal (GST_PROXY_PAD (stream->pad)));
+    g_free (internal_name);
-    gst_pad_set_chain_function (GST_PAD_CAST (internal_pad), _src_chain);
+    gst_object_set_parent (GST_OBJECT_CAST (stream->internal_pad),
-    gst_pad_set_event_function (GST_PAD_CAST (internal_pad), _src_event);
+        GST_OBJECT_CAST (demux));
-    /* need to set query otherwise deadlocks happen with allocation queries */
+    gst_pad_set_element_private (stream->internal_pad, stream);
-    gst_pad_set_query_function (GST_PAD_CAST (internal_pad), _src_query);
+    gst_pad_set_active (stream->internal_pad, TRUE);
-    gst_object_unref (internal_pad);
+    gst_pad_set_chain_function (stream->internal_pad, _src_chain);
+    gst_pad_set_event_function (stream->internal_pad, _src_event);
+    gst_pad_set_query_function (stream->internal_pad, _src_query);
+
+    if (gst_pad_link_full (stream->src_srcpad, stream->internal_pad,
+            GST_PAD_LINK_CHECK_NOTHING) != GST_PAD_LINK_OK) {
+      GST_ERROR_OBJECT (stream->pad, "Failed to link internal pad");
+      return FALSE;
+    }
+
     stream->uri_handler = uri_handler;
     stream->queue = queue;
   }
@@ -2443,7 +2424,10 @@ gst_adaptive_demux_stream_download_uri (GstAdaptiveDemux * demux,
   }
   g_mutex_unlock (&stream->fragment_download_lock);
 
-  gst_adaptive_demux_stream_clear_eos_and_flush_state (stream);
+  /* deactivate and reactivate our ghostpad to make it fresh for a new
+   * stream */
+  gst_pad_set_active (stream->internal_pad, FALSE);
+  gst_pad_set_active (stream->internal_pad, TRUE);
 
   return ret;
 }

gst-libs/gst/adaptivedemux/gstadaptivedemux.h

@@ -114,6 +114,7 @@ struct _GstAdaptiveDemuxStreamFragment
 struct _GstAdaptiveDemuxStream
 {
   GstPad *pad;
+  GstPad *internal_pad;
 
   GstAdaptiveDemux *demux;
 

tests/check/elements/adaptive_demux_common.c

@@ -403,7 +403,7 @@ testSeekOnStateChanged (GstBus * bus, GstMessage * msg, gpointer user_data)
       gst_element_state_get_name (old_state),
       gst_element_state_get_name (new_state));
 
-  if (strstr (srcName, "bin") == srcName &&
+  if (strstr (srcName, "srcbin") == srcName &&
       old_state == GST_STATE_PLAYING && new_state == GST_STATE_PAUSED) {
     g_mutex_lock (&testData->test_task_state_lock);
     if (testData->test_task_state ==

tests/check/elements/adaptive_demux_engine.c

@@ -1,4 +1,4 @@
-/* A generic test engine for elements based upon GstAdaptiveDemux 
+/* A generic test engine for elements based upon GstAdaptiveDemux
  *
  * Copyright (c) <2015> YouView TV Ltd
  *
@@ -227,6 +227,43 @@ on_demuxReceivesEvent (GstPad * pad, GstPadProbeInfo * info, gpointer data)
   return GST_PAD_PROBE_OK;
 }
 
+
+static void
+on_demuxElementAdded (GstBin * demux, GstElement * element, gpointer user_data)
+{
+  GstAdaptiveDemuxTestEnginePrivate *priv =
+      (GstAdaptiveDemuxTestEnginePrivate *) user_data;
+  GstAdaptiveDemuxTestOutputStream *stream = NULL;
+  GstPad *internal_pad;
+  gchar *srcbin_name;
+  gint i;
+
+  srcbin_name = GST_ELEMENT_NAME (element);
+  GST_TEST_LOCK (priv);
+  for (i = 0; i < priv->engine.output_streams->len; i++) {
+    stream = g_ptr_array_index (priv->engine.output_streams, i);
+    if (strstr (srcbin_name, GST_PAD_NAME (stream->pad)) != NULL)
+      break;
+  }
+  fail_unless (stream != NULL);
+
+  /* keep the reference to the internal_pad.
+   * We will need it to identify the stream in the on_demuxReceivesEvent callback
+   */
+  if (stream->internal_pad) {
+    gst_pad_remove_probe (stream->internal_pad, stream->internal_pad_probe);
+    gst_object_unref (stream->internal_pad);
+  }
+  internal_pad = gst_element_get_static_pad (element, "src");
+  stream->internal_pad_probe =
+      gst_pad_add_probe (internal_pad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM,
+      (GstPadProbeCallback) on_demuxReceivesEvent, priv, NULL);
+  stream->internal_pad = internal_pad;
+  GST_TEST_UNLOCK (priv);
+
+}
+
+
 /* callback called when demux creates a src pad.
  * We will create an AppSink to get the data
  */
@@ -239,7 +276,7 @@ on_demuxNewPad (GstElement * demux, GstPad * pad, gpointer user_data)
   GstElement *sink;
   gboolean ret;
   gchar *name;
-  GstPad *internal_pad, *appsink_pad;
+  GstPad *appsink_pad;
   GstAppSinkCallbacks appSinkCallbacks;
   GstAdaptiveDemuxTestOutputStream *stream;
   GObjectClass *gobject_class;
@@ -272,7 +309,6 @@ on_demuxNewPad (GstElement * demux, GstPad * pad, gpointer user_data)
       (GstPadProbeCallback) on_appsink_event, priv, NULL);
   gst_object_unref (appsink_pad);
 
-
   gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BUFFER,
       (GstPadProbeCallback) on_demux_sent_data, priv, NULL);
   gobject_class = G_OBJECT_GET_CLASS (sink);
@@ -281,16 +317,7 @@ on_demuxNewPad (GstElement * demux, GstPad * pad, gpointer user_data)
     g_object_set (G_OBJECT (sink), "sync", FALSE, NULL);
   }
   stream->pad = gst_object_ref (pad);
-  internal_pad =
-      GST_PAD_CAST (gst_proxy_pad_get_internal (GST_PROXY_PAD (pad)));
 
-  gst_pad_add_probe (internal_pad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM,
-      (GstPadProbeCallback) on_demuxReceivesEvent, priv, NULL);
-
-  /* keep the reference to the internal_pad.
-   * We will need it to identify the stream in the on_demuxReceivesEvent callback
-   */
-  stream->internal_pad = internal_pad;
 
   g_ptr_array_add (priv->engine.output_streams, stream);
   GST_TEST_UNLOCK (priv);
@@ -439,6 +466,8 @@ gst_adaptive_demux_test_run (const gchar * element_name,
   priv->engine.demux = demux;
   GST_DEBUG ("created demux %" GST_PTR_FORMAT, demux);
 
+  g_signal_connect (demux, "element-added", G_CALLBACK (on_demuxElementAdded),
+      priv);
   g_signal_connect (demux, "pad-added", G_CALLBACK (on_demuxNewPad), priv);
   g_signal_connect (demux, "pad-removed",
       G_CALLBACK (on_demuxPadRemoved), priv);

tests/check/elements/adaptive_demux_engine.h

@@ -35,6 +35,7 @@ typedef struct _GstAdaptiveDemuxTestOutputStream {
   GstPad *pad;
   /* the internal pad of adaptivedemux element used to send data to the GstAppSink element */
   GstPad *internal_pad;
+  gulong internal_pad_probe;
   /* current segment start offset */
   guint64 segment_start;
   /* the size received so far on this segment */