videoaggregator: implement samples selection API

Call gst_aggregator_selected_samples() after filling the queues
(but before preparing frames).

Implement GstAggregator.peek_next_sample.

Add an example that demonstrates usage of the new API in combination
with the existing buffer-consumed signal.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/merge_requests/728>

gst-libs/gst/video/gstvideoaggregator.c

@@ -213,6 +213,22 @@ gst_video_aggregator_pad_clean_frame (GstVideoAggregatorPad * pad,
   }
 }
 
+static GstSample *
+gst_video_aggregator_peek_next_sample (GstAggregator * agg,
+    GstAggregatorPad * aggpad)
+{
+  GstVideoAggregatorPad *vaggpad = GST_VIDEO_AGGREGATOR_PAD (aggpad);
+  GstSample *res = NULL;
+
+  if (vaggpad->priv->buffer) {
+    GstCaps *caps = gst_pad_get_current_caps (GST_PAD (aggpad));
+    res = gst_sample_new (vaggpad->priv->buffer, caps, &aggpad->segment, NULL);
+    gst_caps_unref (caps);
+  }
+
+  return res;
+}
+
 static void
 gst_video_aggregator_pad_class_init (GstVideoAggregatorPadClass * klass)
 {
@@ -451,8 +467,8 @@ gst_video_aggregator_convert_pad_prepare_frame (GstVideoAggregatorPad * vpad,
     if (!gst_video_info_is_equal (&vpad->info, &pad->priv->conversion_info)) {
       pad->priv->convert =
           gst_video_converter_new (&vpad->info, &pad->priv->conversion_info,
-          pad->priv->converter_config ? gst_structure_copy (pad->
-              priv->converter_config) : NULL);
+          pad->priv->converter_config ? gst_structure_copy (pad->priv->
+              converter_config) : NULL);
       if (!pad->priv->convert) {
         GST_WARNING_OBJECT (pad, "No path found for conversion");
         return FALSE;
@@ -1564,7 +1580,7 @@ gst_video_aggregator_fill_queues (GstVideoAggregator * vagg,
         continue;
       }
 
-      if (end_time >= output_start_running_time
+      if (end_time > output_start_running_time
           && start_time < output_end_running_time) {
         GST_DEBUG_OBJECT (pad,
             "Taking new buffer with start time %" GST_TIME_FORMAT,
@@ -1755,6 +1771,9 @@ gst_video_aggregator_do_aggregate (GstVideoAggregator * vagg,
   gst_element_foreach_sink_pad (GST_ELEMENT_CAST (vagg), sync_pad_values,
       &out_stream_time);
 
+  /* Let the application know that input buffers have been staged */
+  gst_aggregator_selected_samples (agg);
+
   /* Convert all the frames the subclass has before aggregating */
   gst_element_foreach_sink_pad (GST_ELEMENT_CAST (vagg), prepare_frames, NULL);
 
@@ -2639,6 +2658,7 @@ gst_video_aggregator_class_init (GstVideoAggregatorClass * klass)
       gst_video_aggregator_default_negotiated_src_caps;
   agg_class->decide_allocation = gst_video_aggregator_decide_allocation;
   agg_class->propose_allocation = gst_video_aggregator_propose_allocation;
+  agg_class->peek_next_sample = gst_video_aggregator_peek_next_sample;
 
   klass->find_best_format = gst_video_aggregator_find_best_format;
   klass->create_output_buffer = gst_video_aggregator_create_output_buffer;

tests/check/elements/compositor.c

@@ -2166,6 +2166,88 @@ GST_START_TEST (test_gap_events)
 
 GST_END_TEST;
 
+static GstBuffer *expected_selected_buffer = NULL;
+
+static void
+samples_selected_cb (GstAggregator * agg, gint * called)
+{
+  GstPad *pad;
+  GstSample *sample;
+
+  pad = gst_element_get_static_pad (GST_ELEMENT (agg), "sink_0");
+  sample = gst_aggregator_peek_next_sample (agg, GST_AGGREGATOR_PAD (pad));
+  fail_unless (sample != NULL);
+  fail_unless (gst_sample_get_buffer (sample) == expected_selected_buffer);
+  gst_sample_unref (sample);
+  gst_object_unref (pad);
+
+  *called += 1;
+}
+
+static void
+buffer_consumed_cb (GstAggregator * agg, GstBuffer * unused, gint * called)
+{
+  *called += 1;
+}
+
+GST_START_TEST (test_signals)
+{
+  gint samples_selected_called = 0;
+  gint buffer_consumed_called = 0;
+  GstBuffer *buf;
+  GstElement *comp = gst_element_factory_make ("compositor", NULL);
+  GstHarness *h = gst_harness_new_with_element (comp, "sink_%u", "src");
+  GstPad *pad;
+
+  g_object_set (comp, "emit-signals", TRUE, NULL);
+  g_signal_connect (comp, "samples-selected", G_CALLBACK (samples_selected_cb),
+      &samples_selected_called);
+
+  pad = gst_element_get_static_pad (comp, "sink_0");
+  g_object_set (pad, "emit-signals", TRUE, NULL);
+  g_signal_connect (pad, "buffer-consumed", G_CALLBACK (buffer_consumed_cb),
+      &buffer_consumed_called);
+  gst_object_unref (pad);
+
+  gst_harness_set_sink_caps_str (h,
+      "video/x-raw, format=RGBA, width=1, height=1, framerate=1/1");
+  gst_harness_set_src_caps_str (h,
+      "video/x-raw, format=RGBA, width=1, height=1, framerate=2/1");
+
+  gst_harness_play (h);
+
+  buf = gst_buffer_new_allocate (NULL, 4, NULL);
+  GST_BUFFER_PTS (buf) = 0;
+  GST_BUFFER_DURATION (buf) = GST_SECOND / 2;
+  expected_selected_buffer = buf;
+  gst_harness_push (h, buf);
+  buf = gst_harness_pull (h);
+  gst_buffer_unref (buf);
+  fail_unless_equals_int (samples_selected_called, 1);
+  fail_unless_equals_int (buffer_consumed_called, 1);
+
+  /* This next buffer should be discarded */
+  buf = gst_buffer_new_allocate (NULL, 4, NULL);
+  GST_BUFFER_PTS (buf) = GST_SECOND / 2;
+  GST_BUFFER_DURATION (buf) = GST_SECOND / 2;
+  gst_harness_push (h, buf);
+
+  buf = gst_buffer_new_allocate (NULL, 4, NULL);
+  GST_BUFFER_PTS (buf) = GST_SECOND;
+  GST_BUFFER_DURATION (buf) = GST_SECOND / 2;
+  expected_selected_buffer = buf;
+  gst_harness_push (h, buf);
+  buf = gst_harness_pull (h);
+  gst_buffer_unref (buf);
+  fail_unless_equals_int (samples_selected_called, 2);
+  fail_unless_equals_int (buffer_consumed_called, 3);
+
+  gst_harness_teardown (h);
+  gst_object_unref (comp);
+}
+
+GST_END_TEST;
+
 static Suite *
 compositor_suite (void)
 {
@@ -2205,6 +2287,7 @@ compositor_suite (void)
   tcase_add_test (tc_chain, test_start_time_first_live_drop_3);
   tcase_add_test (tc_chain, test_start_time_first_live_drop_3_unlinked_1);
   tcase_add_test (tc_chain, test_gap_events);
+  tcase_add_test (tc_chain, test_signals);
 
   return s;
 }

tests/examples/compositor/meson.build

@@ -3,3 +3,9 @@ executable('crossfade', 'crossfade.c',
   c_args: ['-DHAVE_CONFIG_H'],
   dependencies: [gst_controller_dep, gst_dep],
   install: false)
+
+executable('signals', 'signals.c',
+  include_directories: [configinc],
+  c_args: ['-DHAVE_CONFIG_H'],
+  dependencies: [gst_dep, gst_base_dep],
+  install: false)

tests/examples/compositor/signals.c

@@ -0,0 +1,165 @@
+#include <gst/gst.h>
+#include <gst/base/gstaggregator.h>
+
+#define MAKE_AND_ADD(var, pipe, name, label)                                   \
+  G_STMT_START                                                                 \
+  {                                                                            \
+    GError* make_and_add_err = NULL;                                           \
+    if (G_UNLIKELY(!(var = (gst_parse_bin_from_description_full(               \
+                       name,                                                   \
+                       TRUE,                                                   \
+                       NULL,                                                   \
+                       GST_PARSE_FLAG_NO_SINGLE_ELEMENT_BINS,                  \
+                       &make_and_add_err))))) {                                \
+      GST_ERROR(                                                               \
+        "Could not create element %s (%s)", name, make_and_add_err->message);  \
+      g_clear_error(&make_and_add_err);                                        \
+      goto label;                                                              \
+    }                                                                          \
+    if (G_UNLIKELY(!gst_bin_add(GST_BIN_CAST(pipe), var))) {                   \
+      GST_ERROR("Could not add element %s", name);                             \
+      goto label;                                                              \
+    }                                                                          \
+  }                                                                            \
+  G_STMT_END
+
+static gboolean
+check_aggregated_buffer (GstElement * agg, GstPad * pad,
+    GHashTable * consumed_buffers)
+{
+  GstSample *sample;
+  GList *pad_consumed_buffers;
+  GList *tmp;
+
+  sample =
+      gst_aggregator_peek_next_sample (GST_AGGREGATOR (agg),
+      GST_AGGREGATOR_PAD (pad));
+
+  g_hash_table_steal_extended (consumed_buffers, pad, NULL,
+      (gpointer *) & pad_consumed_buffers);
+
+  for (tmp = pad_consumed_buffers; tmp; tmp = tmp->next) {
+    GstBuffer *consumed_buffer = (GstBuffer *) tmp->data;
+    gboolean aggregated = FALSE;
+
+    if (sample) {
+      aggregated =
+          consumed_buffer == gst_sample_get_buffer (sample) ? TRUE : FALSE;
+    }
+
+    gst_printerr ("One consumed buffer: %" GST_PTR_FORMAT
+        ", it was%s aggregated\n", consumed_buffer, aggregated ? "" : " not");
+  }
+
+  if (sample) {
+    gst_sample_unref (sample);
+  }
+
+  g_list_free_full (pad_consumed_buffers, (GDestroyNotify) gst_buffer_unref);
+
+  return TRUE;
+}
+
+static void
+samples_selected_cb (GstElement * agg, GHashTable * consumed_buffers)
+{
+  gst_printerr ("Compositor has selected the samples it will aggregate\n");
+  gst_element_foreach_sink_pad (agg,
+      (GstElementForeachPadFunc) check_aggregated_buffer, consumed_buffers);
+}
+
+static void
+pad_buffer_consumed_cb (GstAggregatorPad * pad, GstBuffer * buffer,
+    GHashTable * consumed_buffers)
+{
+  GList *pad_consumed_buffers;
+
+  g_hash_table_steal_extended (consumed_buffers, pad, NULL,
+      (gpointer *) & pad_consumed_buffers);
+
+  pad_consumed_buffers =
+      g_list_append (pad_consumed_buffers, gst_buffer_ref (buffer));
+
+  g_hash_table_insert (consumed_buffers, pad, pad_consumed_buffers);
+}
+
+static gboolean
+unref_consumed_buffers (gpointer key, GList * pad_consumed_buffers)
+{
+  g_list_free_full (pad_consumed_buffers, (GDestroyNotify) gst_buffer_unref);
+
+  return TRUE;
+}
+
+int
+main (int ac, char **av)
+{
+  int ret = 0;
+  GstElement *pipe;
+  GstBus *bus;
+  GstElement *vsrc, *vcfltr1, *compositor, *vcfltr2, *vsink;
+  GstCaps *caps;
+  GstPad *pad;
+  GHashTable *consumed_buffers =
+      g_hash_table_new (g_direct_hash, g_direct_equal);
+
+  gst_init (NULL, NULL);
+
+  pipe = gst_pipeline_new (NULL);
+
+  MAKE_AND_ADD (vsrc, pipe, "videotestsrc", err);
+  MAKE_AND_ADD (vcfltr1, pipe, "capsfilter", err);
+  MAKE_AND_ADD (compositor, pipe, "compositor", err);
+  MAKE_AND_ADD (vcfltr2, pipe, "capsfilter", err);
+  MAKE_AND_ADD (vsink, pipe, "autovideosink", err);
+
+  if (!gst_element_link_many (vsrc, vcfltr1, compositor, vcfltr2, vsink, NULL)) {
+    GST_ERROR ("Failed to link pipeline");
+    goto err;
+  }
+
+  caps =
+      gst_caps_new_simple ("video/x-raw", "framerate", GST_TYPE_FRACTION, 30, 1,
+      NULL);
+  g_object_set (vcfltr1, "caps", caps, NULL);
+  gst_caps_unref (caps);
+
+  caps =
+      gst_caps_new_simple ("video/x-raw", "framerate", GST_TYPE_FRACTION, 6, 1,
+      NULL);
+  g_object_set (vcfltr2, "caps", caps, NULL);
+  gst_caps_unref (caps);
+
+  g_object_set (vsrc, "num-buffers", 300, NULL);
+
+  g_object_set (compositor, "emit-signals", TRUE, NULL);
+  g_signal_connect (compositor, "samples-selected",
+      G_CALLBACK (samples_selected_cb), consumed_buffers);
+
+  pad = gst_element_get_static_pad (compositor, "sink_0");
+  g_object_set (pad, "emit-signals", TRUE, NULL);
+  g_signal_connect (pad, "buffer-consumed", G_CALLBACK (pad_buffer_consumed_cb),
+      consumed_buffers);
+  gst_object_unref (pad);
+
+  gst_element_set_state (pipe, GST_STATE_PLAYING);
+
+  bus = gst_pipeline_get_bus (GST_PIPELINE (pipe));
+
+  gst_message_unref (gst_bus_timed_pop_filtered (bus, GST_CLOCK_TIME_NONE,
+          GST_MESSAGE_EOS));
+
+  gst_object_unref (bus);
+
+done:
+  g_hash_table_foreach_remove (consumed_buffers,
+      (GHRFunc) unref_consumed_buffers, NULL);
+  g_hash_table_unref (consumed_buffers);
+  gst_element_set_state (pipe, GST_STATE_NULL);
+  gst_object_unref (pipe);
+  return ret;
+
+err:
+  ret = 1;
+  goto done;
+}