appsink: add buffer fallback in case the application doesn't handle buffer lists

We shouldn't assume the application handles buffer lists, for
ease-of-use reasons and for backwards compatibility reasons.

gst-libs/gst/app/gstappsink.c

@@ -111,6 +111,8 @@ struct _GstAppSinkPrivate
   GstAppSinkCallbacks callbacks;
   gpointer user_data;
   GDestroyNotify notify;
+
+  gboolean buffer_lists_supported;
 };
 
 GST_DEBUG_CATEGORY_STATIC (app_sink_debug);
@@ -623,6 +625,21 @@ gst_app_sink_flush_unlocked (GstAppSink * appsink)
   g_cond_signal (priv->cond);
 }
 
+#define NEW_BUFFER_LIST_SIGID \
+    gst_app_sink_signals[SIGNAL_NEW_BUFFER_LIST]
+
+static gboolean
+gst_app_sink_check_buffer_lists_support (GstAppSink * appsink)
+{
+  gboolean ret;
+
+  ret = (appsink->priv->callbacks.new_buffer_list != NULL) ||
+      g_signal_has_handler_pending (appsink, NEW_BUFFER_LIST_SIGID, 0, FALSE);
+
+  GST_INFO_OBJECT (appsink, "application supports buffer lists: %d", ret);
+  return ret;
+}
+
 static gboolean
 gst_app_sink_start (GstBaseSink * psink)
 {
@@ -633,6 +650,8 @@ gst_app_sink_start (GstBaseSink * psink)
   GST_DEBUG_OBJECT (appsink, "starting");
   priv->flushing = FALSE;
   priv->started = TRUE;
+  priv->buffer_lists_supported =
+      gst_app_sink_check_buffer_lists_support (appsink);
   g_mutex_unlock (priv->mutex);
 
   return TRUE;
@@ -812,9 +831,46 @@ gst_app_sink_render (GstBaseSink * psink, GstBuffer * buffer)
 }
 
 static GstFlowReturn
-gst_app_sink_render_list (GstBaseSink * psink, GstBufferList * list)
+gst_app_sink_render_list (GstBaseSink * sink, GstBufferList * list)
 {
-  return gst_app_sink_render_common (psink, GST_MINI_OBJECT_CAST (list), TRUE);
+  GstBufferListIterator *it;
+  GstFlowReturn flow;
+  GstAppSink *appsink;
+  GstBuffer *group;
+
+  appsink = GST_APP_SINK_CAST (sink);
+
+  if (appsink->priv->buffer_lists_supported)
+    return gst_app_sink_render_common (sink, GST_MINI_OBJECT_CAST (list), TRUE);
+
+  /* The application doesn't support buffer lists, extract individual buffers
+   * then and push them one-by-one */
+  GST_INFO_OBJECT (sink, "chaining each group in list as a merged buffer");
+
+  it = gst_buffer_list_iterate (list);
+
+  if (gst_buffer_list_iterator_next_group (it)) {
+    do {
+      group = gst_buffer_list_iterator_merge_group (it);
+      if (group == NULL) {
+        group = gst_buffer_new ();
+        GST_DEBUG_OBJECT (sink, "chaining empty group");
+      } else {
+        GST_DEBUG_OBJECT (sink, "chaining group");
+      }
+      flow = gst_app_sink_render (sink, group);
+      gst_buffer_unref (group);
+    } while (flow == GST_FLOW_OK && gst_buffer_list_iterator_next_group (it));
+  } else {
+    GST_DEBUG_OBJECT (sink, "chaining empty group");
+    group = gst_buffer_new ();
+    flow = gst_app_sink_render (sink, group);
+    gst_buffer_unref (group);
+  }
+
+  gst_buffer_list_iterator_free (it);
+
+  return flow;
 }
 
 static GstCaps *
@@ -1335,6 +1391,8 @@ gst_app_sink_set_callbacks (GstAppSink * appsink,
   priv->callbacks = *callbacks;
   priv->user_data = user_data;
   priv->notify = notify;
+  priv->buffer_lists_supported =
+      gst_app_sink_check_buffer_lists_support (appsink);
   GST_OBJECT_UNLOCK (appsink);
 }
 

tests/check/elements/appsink.c

@@ -317,6 +317,91 @@ GST_START_TEST (test_buffer_list)
 
 GST_END_TEST;
 
+static GstFlowReturn
+callback_function_buffer (GstAppSink * appsink, gpointer p_counter)
+{
+  GstBuffer *buf;
+  gint *p_int_counter = p_counter;
+
+  buf = gst_app_sink_pull_buffer (appsink);
+  fail_unless (GST_IS_BUFFER (buf));
+
+  /* buffer list has 3 buffers in two groups */
+  switch (*p_int_counter) {
+    case 0:
+      fail_unless_equals_int (GST_BUFFER_SIZE (buf), sizeof (gint));
+      fail_unless_equals_int ((((gint *) GST_BUFFER_DATA (buf))[0]), 1);
+      break;
+    case 1:
+      fail_unless_equals_int (GST_BUFFER_SIZE (buf), 2 * sizeof (gint));
+      fail_unless_equals_int ((((gint *) GST_BUFFER_DATA (buf))[0]), 2);
+      fail_unless_equals_int ((((gint *) GST_BUFFER_DATA (buf))[1]), 4);
+      break;
+    default:
+      g_warn_if_reached ();
+      break;
+  }
+
+  gst_buffer_unref (buf);
+
+  *p_int_counter += 1;
+
+  return GST_FLOW_OK;
+}
+
+GST_START_TEST (test_buffer_list_fallback)
+{
+  GstElement *sink;
+  GstBufferList *list;
+  GstAppSinkCallbacks callbacks = { NULL };
+  gint counter = 0;
+
+  sink = setup_appsink ();
+
+  callbacks.new_buffer = callback_function_buffer;
+
+  gst_app_sink_set_callbacks (GST_APP_SINK (sink), &callbacks, &counter, NULL);
+
+  ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
+
+  list = create_buffer_list ();
+  fail_unless (gst_pad_push_list (mysrcpad, list) == GST_FLOW_OK);
+
+  fail_unless_equals_int (counter, 2);
+
+  ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
+  cleanup_appsink (sink);
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_buffer_list_fallback_signal)
+{
+  GstElement *sink;
+  GstBufferList *list;
+  gint counter = 0;
+
+  sink = setup_appsink ();
+
+  /* C calling convention to the rescue.. */
+  g_signal_connect (sink, "new-buffer", G_CALLBACK (callback_function_buffer),
+      &counter);
+
+  g_object_set (sink, "emit-signals", TRUE, NULL);
+
+  ASSERT_SET_STATE (sink, GST_STATE_PLAYING, GST_STATE_CHANGE_ASYNC);
+
+  list = create_buffer_list ();
+  fail_unless (gst_pad_push_list (mysrcpad, list) == GST_FLOW_OK);
+
+  fail_unless_equals_int (counter, 2);
+
+  ASSERT_SET_STATE (sink, GST_STATE_NULL, GST_STATE_CHANGE_SUCCESS);
+  cleanup_appsink (sink);
+}
+
+GST_END_TEST;
+
 static Suite *
 appsink_suite (void)
 {
@@ -329,6 +414,8 @@ appsink_suite (void)
   tcase_add_test (tc_chain, test_notify0);
   tcase_add_test (tc_chain, test_notify1);
   tcase_add_test (tc_chain, test_buffer_list);
+  tcase_add_test (tc_chain, test_buffer_list_fallback);
+  tcase_add_test (tc_chain, test_buffer_list_fallback_signal);
 
   return s;
 }