smart-mixer: Use the new 'samples-selected' signal to handle queuing in aggregator pads

Since aggregator introduced queueing in its sinkpads the way we set
properties on the pads is incorrect as it doesn't take it into account.
This fixes the issue by using the newly introduced `samples-selected`
signal in aggregator to set the properties right before the compositing
is done.

Also require the compositor we use to be an aggregator.

And add a validate test for it.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-editing-services/-/merge_requests/204>

6 files changed, 270 insertions, 30 deletions

diff --git a/ges/ges-smart-video-mixer.c b/ges/ges-smart-video-mixer.c
index 5197a3eb..26147fff 100644
--- a/ges/ges-smart-video-mixer.c
+++ b/ges/ges-smart-video-mixer.c
@@ -24,6 +24,7 @@
 #include "ges-types.h"
 #include "ges-internal.h"
 #include "ges-smart-video-mixer.h"
+#include <gst/base/base.h>
 
 #define GES_TYPE_SMART_MIXER_PAD             (ges_smart_mixer_pad_get_type ())
 typedef struct _GESSmartMixerPad GESSmartMixerPad;
@@ -123,29 +124,48 @@ static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink_%u",
 
 typedef struct _PadInfos
 {
+  volatile gint refcount;
+
   GESSmartMixer *self;
   GstPad *mixer_pad;
+  GstPad *ghostpad;
   GstElement *bin;
-  gulong probe_id;
 } PadInfos;
 
 static void
-destroy_pad (PadInfos * infos)
+pad_infos_unref (PadInfos * infos)
 {
-  gst_pad_remove_probe (infos->mixer_pad, infos->probe_id);
+  if (g_atomic_int_dec_and_test (&infos->refcount)) {
+    GST_DEBUG_OBJECT (infos->mixer_pad, "Releasing pad");
+    if (G_LIKELY (infos->bin)) {
+      gst_element_set_state (infos->bin, GST_STATE_NULL);
+      gst_element_unlink (infos->bin, infos->self->mixer);
+      gst_bin_remove (GST_BIN (infos->self), infos->bin);
+    }
 
-  if (G_LIKELY (infos->bin)) {
-    gst_element_set_state (infos->bin, GST_STATE_NULL);
-    gst_element_unlink (infos->bin, infos->self->mixer);
-    gst_bin_remove (GST_BIN (infos->self), infos->bin);
-  }
+    if (infos->mixer_pad) {
+      gst_element_release_request_pad (infos->self->mixer, infos->mixer_pad);
+      gst_object_unref (infos->mixer_pad);
+    }
 
-  if (infos->mixer_pad) {
-    gst_element_release_request_pad (infos->self->mixer, infos->mixer_pad);
-    gst_object_unref (infos->mixer_pad);
+    g_free (infos);
   }
+}
 
-  g_slice_free (PadInfos, infos);
+static PadInfos *
+pad_infos_new (void)
+{
+  PadInfos *info = g_new0 (PadInfos, 1);
+  info->refcount = 1;
+
+  return info;
+}
+
+static PadInfos *
+pad_infos_ref (PadInfos * info)
+{
+  g_atomic_int_inc (&info->refcount);
+  return info;
 }
 
 static gboolean
@@ -190,20 +210,19 @@ ges_smart_mixer_get_mixer_pad (GESSmartMixer * self, GstPad ** mixerpad)
 
 /* These metadata will get set by the upstream framepositioner element,
    added in the video sources' bin */
-static GstPadProbeReturn
-parse_metadata (GstPad * mixer_pad, GstPadProbeInfo * info,
-    GESSmartMixerPad * ghost)
+static void
+parse_metadata (GstPad * mixer_pad, GstBuffer * buf, GESSmartMixerPad * ghost)
 {
   GstFramePositionerMeta *meta;
   GESSmartMixer *self = GES_SMART_MIXER (GST_OBJECT_PARENT (ghost));
 
   meta =
-      (GstFramePositionerMeta *) gst_buffer_get_meta ((GstBuffer *) info->data,
+      (GstFramePositionerMeta *) gst_buffer_get_meta (buf,
       gst_frame_positioner_meta_api_get_type ());
 
   if (!meta) {
     GST_WARNING ("The current source should use a framepositioner");
-    return GST_PAD_PROBE_OK;
+    return;
   }
 
   if (!self->disable_zorder_alpha) {
@@ -215,7 +234,7 @@ parse_metadata (GstPad * mixer_pad, GstPadProbeInfo * info,
 
     GST_OBJECT_LOCK (ghost);
     stream_time = gst_segment_to_stream_time (&ghost->segment, GST_FORMAT_TIME,
-        GST_BUFFER_PTS (info->data));
+        GST_BUFFER_PTS (buf));
     GST_OBJECT_UNLOCK (ghost);
 
     if (GST_CLOCK_TIME_IS_VALID (stream_time))
@@ -227,8 +246,6 @@ parse_metadata (GstPad * mixer_pad, GstPadProbeInfo * info,
 
   g_object_set (mixer_pad, "xpos", meta->posx, "ypos",
       meta->posy, "width", meta->width, "height", meta->height, NULL);
-
-  return GST_PAD_PROBE_OK;
 }
 
 /****************************************************
@@ -239,7 +256,7 @@ _request_new_pad (GstElement * element, GstPadTemplate * templ,
     const gchar * name, const GstCaps * caps)
 {
   GstPad *videoconvert_srcpad, *videoconvert_sinkpad, *tmpghost;
-  PadInfos *infos = g_slice_new0 (PadInfos);
+  PadInfos *infos = pad_infos_new ();
   GESSmartMixer *self = GES_SMART_MIXER (element);
   GstPad *ghost;
   GstElement *videoconvert;
@@ -250,7 +267,7 @@ _request_new_pad (GstElement * element, GstPadTemplate * templ,
 
   if (infos->mixer_pad == NULL) {
     GST_WARNING_OBJECT (element, "Could not get any pad from GstMixer");
-    g_slice_free (PadInfos, infos);
+    pad_infos_unref (infos);
 
     return NULL;
   }
@@ -271,6 +288,7 @@ _request_new_pad (GstElement * element, GstPadTemplate * templ,
   gst_bin_add (GST_BIN (self), infos->bin);
   ghost = g_object_new (ges_smart_mixer_pad_get_type (), "name", name,
       "direction", GST_PAD_DIRECTION (tmpghost), NULL);
+  infos->ghostpad = ghost;
   gst_ghost_pad_set_target (GST_GHOST_PAD_CAST (ghost), tmpghost);
   gst_pad_set_active (ghost, TRUE);
   if (!gst_element_add_pad (GST_ELEMENT (self), ghost))
@@ -283,14 +301,13 @@ _request_new_pad (GstElement * element, GstPadTemplate * templ,
   gst_element_add_pad (GST_ELEMENT (infos->bin), tmpghost);
   gst_pad_link (tmpghost, infos->mixer_pad);
 
-  infos->probe_id =
-      gst_pad_add_probe (infos->mixer_pad, GST_PAD_PROBE_TYPE_BUFFER,
-      (GstPadProbeCallback) parse_metadata, ghost, NULL);
   gst_pad_set_event_function (GST_PAD (ghost),
       ges_smart_mixer_sinkpad_event_func);
 
   LOCK (self);
   g_hash_table_insert (self->pads_infos, ghost, infos);
+  g_hash_table_insert (self->pads_infos, infos->mixer_pad,
+      pad_infos_ref (infos));
   UNLOCK (self);
 
   GST_DEBUG_OBJECT (self, "Returning new pad %" GST_PTR_FORMAT, ghost);
@@ -299,19 +316,38 @@ _request_new_pad (GstElement * element, GstPadTemplate * templ,
 could_not_add:
   {
     GST_ERROR_OBJECT (self, "could not add pad");
-    destroy_pad (infos);
+    pad_infos_unref (infos);
     return NULL;
   }
 }
 
+static PadInfos *
+ges_smart_mixer_find_pad_info (GESSmartMixer * self, GstPad * pad)
+{
+  PadInfos *info;
+
+  LOCK (self);
+  info = g_hash_table_lookup (self->pads_infos, pad);
+  UNLOCK (self);
+
+  if (info)
+    pad_infos_ref (info);
+
+  return info;
+}
+
 static void
 _release_pad (GstElement * element, GstPad * pad)
 {
   GstPad *peer;
+  GESSmartMixer *self = GES_SMART_MIXER (element);
+  PadInfos *info = ges_smart_mixer_find_pad_info (self, pad);
+
   GST_DEBUG_OBJECT (element, "Releasing pad %" GST_PTR_FORMAT, pad);
 
   LOCK (element);
   g_hash_table_remove (GES_SMART_MIXER (element)->pads_infos, pad);
+  g_hash_table_remove (GES_SMART_MIXER (element)->pads_infos, info->mixer_pad);
   peer = gst_pad_get_peer (pad);
   if (peer) {
     gst_pad_unlink (peer, pad);
@@ -321,6 +357,45 @@ _release_pad (GstElement * element, GstPad * pad)
   gst_pad_set_active (pad, FALSE);
   gst_element_remove_pad (element, pad);
   UNLOCK (element);
+
+  pad_infos_unref (info);
+}
+
+static gboolean
+compositor_sync_properties_with_meta (GstElement * compositor,
+    GstPad * sinkpad, GESSmartMixer * self)
+{
+  PadInfos *info = ges_smart_mixer_find_pad_info (self, sinkpad);
+  GstSample *sample;
+
+  if (!info) {
+    GST_WARNING_OBJECT (self, "Couldn't find pad info?!");
+
+    return TRUE;
+  }
+
+  sample = gst_aggregator_peek_next_sample (GST_AGGREGATOR (compositor),
+      GST_AGGREGATOR_PAD (sinkpad));
+
+  if (!sample) {
+    GST_INFO_OBJECT (sinkpad, "No sample set!");
+  } else {
+    parse_metadata (sinkpad, gst_sample_get_buffer (sample),
+        GES_SMART_MIXER_PAD (info->ghostpad));
+    gst_sample_unref (sample);
+  }
+  pad_infos_unref (info);
+
+  return TRUE;
+}
+
+static void
+ges_smart_mixer_samples_selected_cb (GstElement * compositor,
+    GstSegment * segment, GstClockTime pts, GstClockTime dts,
+    GstClockTime duration, GstStructure * info, GESSmartMixer * self)
+{
+  gst_element_foreach_sink_pad (compositor,
+      (GstElementForeachPadFunc) compositor_sync_properties_with_meta, self);
 }
 
 /****************************************************
@@ -361,7 +436,9 @@ ges_smart_mixer_constructed (GObject * obj)
   self->mixer =
       gst_element_factory_create (ges_get_compositor_factory (), cname);
   g_free (cname);
-  g_object_set (self->mixer, "background", 1, NULL);
+  g_object_set (self->mixer, "background", 1, "emit-signals", TRUE, NULL);
+  g_signal_connect (self->mixer, "samples-selected",
+      G_CALLBACK (ges_smart_mixer_samples_selected_cb), self);
 
   /* See https://gitlab.freedesktop.org/gstreamer/gstreamer/issues/310 */
   GST_FIXME ("Stop dropping allocation query when it is not required anymore.");
@@ -393,7 +470,7 @@ ges_smart_mixer_class_init (GESSmartMixerClass * klass)
   gst_element_class_add_static_pad_template (element_class, &sink_template);
   gst_element_class_set_static_metadata (element_class, "GES Smart mixer",
       "Generic/Audio",
-      "Use mixer making use of GES informations",
+      "Use mixer making use of GES information",
       "Thibault Saunier <thibault.saunier@collabora.com>");
 
   element_class->request_new_pad = GST_DEBUG_FUNCPTR (_request_new_pad);
@@ -409,7 +486,7 @@ ges_smart_mixer_init (GESSmartMixer * self)
 {
   g_mutex_init (&self->lock);
   self->pads_infos = g_hash_table_new_full (g_direct_hash, g_direct_equal,
-      NULL, (GDestroyNotify) destroy_pad);
+      NULL, (GDestroyNotify) pad_infos_unref);
 }
 
 GstElement *
diff --git a/ges/ges-timeline.c b/ges/ges-timeline.c
index 38cce07c..2f49e218 100644
--- a/ges/ges-timeline.c
+++ b/ges/ges-timeline.c
@@ -496,6 +496,7 @@ ges_timeline_handle_message (GstBin * bin, GstMessage * message)
     }
 
     if (amessage) {
+      gst_message_unref (message);
       gst_element_post_message (GST_ELEMENT_CAST (bin), amessage);
       return;
     }
diff --git a/ges/ges-utils.c b/ges/ges-utils.c
index d8baacf0..2ee9b06b 100644
--- a/ges/ges-utils.c
+++ b/ges/ges-utils.c
@@ -34,6 +34,7 @@
 #include "ges-track.h"
 #include "ges-layer.h"
 #include "ges.h"
+#include <gst/base/base.h>
 
 static GstElementFactory *compositor_factory = NULL;
 
@@ -133,7 +134,9 @@ ges_pspec_hash (gconstpointer key_spec)
 static gboolean
 find_compositor (GstPluginFeatureFilter * feature, gpointer udata)
 {
+  gboolean res = FALSE;
   const gchar *klass;
+  GstPluginFeature *loaded_feature = NULL;
 
   if (G_UNLIKELY (!GST_IS_ELEMENT_FACTORY (feature)))
     return FALSE;
@@ -141,7 +144,20 @@ find_compositor (GstPluginFeatureFilter * feature, gpointer udata)
   klass = gst_element_factory_get_metadata (GST_ELEMENT_FACTORY_CAST (feature),
       GST_ELEMENT_METADATA_KLASS);
 
-  return (strstr (klass, "Compositor") != NULL);
+  if (strstr (klass, "Compositor") == NULL)
+    return FALSE;
+
+  loaded_feature = gst_plugin_feature_load (feature);
+  if (!loaded_feature) {
+    GST_ERROR ("Could not load feature: %" GST_PTR_FORMAT, feature);
+    return FALSE;
+  }
+
+  res =
+      g_type_is_a (gst_element_factory_get_element_type (GST_ELEMENT_FACTORY
+          (loaded_feature)), GST_TYPE_AGGREGATOR);
+  gst_object_unref (loaded_feature);
+  return res;
 }
 
 gboolean
diff --git a/tests/check/meson.build b/tests/check/meson.build
index b377cb1d..d5f26664 100644
--- a/tests/check/meson.build
+++ b/tests/check/meson.build
@@ -83,6 +83,7 @@ if gstvalidate_dep.found()
     'seek_with_stop.check_clock_sync': true,
     'edit_while_seeked_with_stop': true,
     'complex_effect_bin_desc': true,
+    'check_keyframes_in_compositor_two_sources': true,
   }
 
   foreach scenario, is_validatetest: scenarios
diff --git a/tests/check/scenarios/check_keyframes_in_compositor_two_sources.validatetest b/tests/check/scenarios/check_keyframes_in_compositor_two_sources.validatetest
new file mode 100644
index 00000000..763bee4e
--- /dev/null
+++ b/tests/check/scenarios/check_keyframes_in_compositor_two_sources.validatetest
@@ -0,0 +1,111 @@
+meta,
+    tool = "ges-launch-$(gst_api_version)",
+    args = {
+        -t, video,
+        --videosink, "$(videosink) name=videosink sync=true",
+        --video-caps, "video/x-raw,width=500,height=500,framerate=10/1,format=I420",
+    },
+    handles-states=true,
+    ignore-eos=true,
+    configs = {
+        "$(validateflow), pad=videosink:sink, buffers-checksum=as-id, ignored-fields=\"stream-start={stream-id,group-id, stream}\"",
+    },
+    expected-issues = {
+        # Blending is not 100% reproducible when seeking/changing values for
+        # some reason. It has been manually checked and the contents are very
+        # slightly off and ssim would detect no difference at all between the
+        # images
+        "expected-issue, issue-id=validateflow::mismatch, details=\".*content-id=10.*\\\\n.*\\\\n.*content-id=12.*\", sometimes=true",
+        "expected-issue, issue-id=validateflow::mismatch, details=\".*content-id=12.*\\\\n.*\\\\n.*content-id=10.*\", sometimes=true",
+        "expected-issue, issue-id=validateflow::mismatch, details=\".*content-id=0.*\\\\n.*\\\\n.*content-id=12.*\", sometimes=true",
+        "expected-issue, issue-id=validateflow::mismatch, details=\".*content-id=0.*\\\\n.*\\\\n.*content-id=13.*\", sometimes=true",
+    }
+
+remove-feature, name=queue
+
+add-clip, name=c0, asset-id="pattern=blue", layer-priority=0, type=GESTestClip, start=0, duration=2.0, inpoint=2.0
+add-clip, name=c1, asset-id="pattern=green", layer-priority=1, type=GESTestClip, start=0, duration=2.0
+set-child-properties, element-name=c0, width=250, height=250, pattern=blue
+set-child-properties, element-name=c1, width=250, height=250, pattern=green
+
+set-control-source, element-name=c0, property-name=posx, binding-type=direct-absolute
+set-control-source, element-name=c0, property-name=posy, binding-type=direct-absolute
+set-control-source, element-name=c1, property-name=posx, binding-type=direct-absolute
+set-control-source, element-name=c1, property-name=posy, binding-type=direct-absolute
+
+# c0 starts in the top left corner going to the bottom right at 1sec and going back to the top right at 2s
+# Keyframes are in 'source stream time' so we take the inpoint into account for the timestamp
+add-keyframe, element-name=c0, timestamp=2.0, property-name=posx, value=0
+add-keyframe, element-name=c0, timestamp=2.0, property-name=posy, value=0
+add-keyframe, element-name=c1, timestamp=0.0, property-name=posx, value=250
+add-keyframe, element-name=c1, timestamp=0.0, property-name=posy, value=250
+
+add-keyframe, element-name=c0, timestamp=3.0, property-name=posx, value=250
+add-keyframe, element-name=c0, timestamp=3.0, property-name=posy, value=250
+add-keyframe, element-name=c1, timestamp=1.0, property-name=posx, value=0
+add-keyframe, element-name=c1, timestamp=1.0, property-name=posy, value=0
+
+add-keyframe, element-name=c0, timestamp=4.0, property-name=posx, value=0
+add-keyframe, element-name=c0, timestamp=4.0, property-name=posy, value=0
+add-keyframe, element-name=c1, timestamp=2.0, property-name=posx, value=250
+add-keyframe, element-name=c1, timestamp=2.0, property-name=posy, value=250
+
+play
+
+check-properties,
+    gessmartmixer0-compositor.sink_0::xpos=0,  gessmartmixer0-compositor.sink_0::ypos=0,
+    gessmartmixer0-compositor.sink_1::xpos=250, gessmartmixer0-compositor.sink_1::ypos=250
+
+crank-clock
+wait, on-clock=true
+
+check-properties,
+    gessmartmixer0-compositor.sink_0::xpos=25,  gessmartmixer0-compositor.sink_0::ypos=25,
+    gessmartmixer0-compositor.sink_1::xpos=225, gessmartmixer0-compositor.sink_1::ypos=225
+
+# Check the 5th buffer
+crank-clock, repeat=4
+wait, on-clock=true
+check-properties,
+    gessmartmixer0-compositor.sink_0::xpos=125, gessmartmixer0-compositor.sink_0::ypos=125,
+    gessmartmixer0-compositor.sink_1::xpos=125,   gessmartmixer0-compositor.sink_1::ypos=125
+
+crank-clock
+check-position, expected-position=0.5
+
+# Check the 10th buffer
+crank-clock, repeat=4
+wait, on-clock=true
+check-properties,
+    gessmartmixer0-compositor.sink_0::xpos=250, gessmartmixer0-compositor.sink_0::ypos=250,
+    gessmartmixer0-compositor.sink_1::xpos=0,   gessmartmixer0-compositor.sink_1::ypos=0
+
+crank-clock
+check-position, expected-position=1.0
+
+crank-clock, repeat=11
+check-position, on-message=eos, expected-position=2000000001
+
+seek, start=1.0, flags=accurate+flush
+wait, on-clock=true
+
+# 10th buffer
+check-properties,
+    gessmartmixer0-compositor.sink_3::xpos=250, gessmartmixer0-compositor.sink_3::ypos=250,
+    gessmartmixer0-compositor.sink_4::xpos=0,   gessmartmixer0-compositor.sink_4::ypos=0
+
+set-ges-properties, element-name=c0, start=1000000000
+set-ges-properties, element-name=c1, start=1000000000
+commit;
+wait, on-clock=true
+
+check-position, expected-position=1.0
+
+# First buffer
+check-properties,
+    gessmartmixer0-compositor.sink_3::xpos=0,
+    gessmartmixer0-compositor.sink_3::ypos=0,
+    gessmartmixer0-compositor.sink_4::xpos=250,
+    gessmartmixer0-compositor.sink_4::ypos=250
+
+stop
diff --git a/tests/check/scenarios/check_keyframes_in_compositor_two_sources/flow-expectations/log-videosink-sink-expected b/tests/check/scenarios/check_keyframes_in_compositor_two_sources/flow-expectations/log-videosink-sink-expected
new file mode 100644
index 00000000..fb0a655a
--- /dev/null
+++ b/tests/check/scenarios/check_keyframes_in_compositor_two_sources/flow-expectations/log-videosink-sink-expected
@@ -0,0 +1,34 @@
+event stream-start: GstEventStreamStart, flags=(GstStreamFlags)GST_STREAM_FLAG_NONE;
+event caps: video/x-raw, format=(string)I420, width=(int)500, height=(int)500, framerate=(fraction)10/1, chroma-site=(string)jpeg, colorimetry=(string)bt601;
+event segment: format=TIME, start=0:00:00.000000000, offset=0:00:00.000000000, stop=0:00:02.000000000, flags=0x01, time=0:00:00.000000000, base=0:00:00.000000000, position=none
+buffer: content-id=0, pts=0:00:00.000000000, dur=0:00:00.100000000
+buffer: content-id=1, pts=0:00:00.100000000, dur=0:00:00.100000000
+buffer: content-id=2, pts=0:00:00.200000000, dur=0:00:00.100000000
+buffer: content-id=3, pts=0:00:00.300000000, dur=0:00:00.100000000
+buffer: content-id=4, pts=0:00:00.400000000, dur=0:00:00.100000000
+buffer: content-id=5, pts=0:00:00.500000000, dur=0:00:00.100000000
+buffer: content-id=6, pts=0:00:00.600000000, dur=0:00:00.100000000
+buffer: content-id=7, pts=0:00:00.700000000, dur=0:00:00.100000000
+buffer: content-id=8, pts=0:00:00.800000000, dur=0:00:00.100000000
+buffer: content-id=9, pts=0:00:00.900000000, dur=0:00:00.100000000
+buffer: content-id=10, pts=0:00:01.000000000, dur=0:00:00.100000000
+buffer: content-id=9, pts=0:00:01.100000000, dur=0:00:00.100000000
+buffer: content-id=8, pts=0:00:01.200000000, dur=0:00:00.100000000
+buffer: content-id=7, pts=0:00:01.300000000, dur=0:00:00.100000000
+buffer: content-id=6, pts=0:00:01.400000000, dur=0:00:00.100000000
+buffer: content-id=5, pts=0:00:01.500000000, dur=0:00:00.100000000
+buffer: content-id=4, pts=0:00:01.600000000, dur=0:00:00.100000000
+buffer: content-id=3, pts=0:00:01.700000000, dur=0:00:00.100000000
+buffer: content-id=2, pts=0:00:01.800000000, dur=0:00:00.100000000
+buffer: content-id=1, pts=0:00:01.900000000, dur=0:00:00.100000000
+event segment: format=TIME, start=0:00:02.000000000, offset=0:00:00.000000000, stop=0:00:02.000000001, flags=0x01, time=0:00:02.000000000, base=0:00:02.000000000, position=none
+buffer: content-id=11, pts=0:00:02.000000000, dur=0:00:00.000000001
+event eos: (no structure)
+event flush-start: (no structure)
+event flush-stop: GstEventFlushStop, reset-time=(boolean)true;
+event segment: format=TIME, start=0:00:01.000000000, offset=0:00:00.000000000, stop=0:00:02.000000000, flags=0x01, time=0:00:01.000000000, base=0:00:00.000000000, position=none
+buffer: content-id=10, pts=0:00:01.000000000, dur=0:00:00.100000000
+event flush-start: (no structure)
+event flush-stop: GstEventFlushStop, reset-time=(boolean)true;
+event segment: format=TIME, start=0:00:01.000000000, offset=0:00:00.000000000, stop=0:00:03.000000000, flags=0x01, time=0:00:01.000000000, base=0:00:00.000000000, position=none
+buffer: content-id=0, pts=0:00:01.000000000, dur=0:00:00.100000000