decklinksrc: aggregate audio and video flow returns

And pause task in some more cases where we should
pause the task.

1 files changed, 28 insertions, 24 deletions

diff --git a/sys/decklink/gstdecklinksrc.cpp b/sys/decklink/gstdecklinksrc.cpp
index c76c7a432..13df571ca 100644
--- a/sys/decklink/gstdecklinksrc.cpp
+++ b/sys/decklink/gstdecklinksrc.cpp
@@ -734,7 +734,7 @@ gst_decklink_src_task (void *priv)
   void *data;
   gsize data_size;
   int n_samples;
-  GstFlowReturn ret;
+  GstFlowReturn video_flow, audio_flow, flow;
   const GstDecklinkMode *mode;
   gboolean discont = FALSE;
 
@@ -838,15 +838,7 @@ gst_decklink_src_task (void *priv)
   else
     GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT);
 
-  /* FIXME: proper flow aggregation with audio flow */
-  ret = gst_pad_push (decklinksrc->videosrcpad, buffer);
-  if (!(ret == GST_FLOW_OK || ret == GST_FLOW_NOT_LINKED ||
-          ret == GST_FLOW_FLUSHING)) {
-    GST_ELEMENT_ERROR (decklinksrc, STREAM, FAILED,
-        ("Internal data stream error."),
-        ("stream stopped, reason %s", gst_flow_get_name (ret)));
-      goto pause;
-  }
+  video_flow = gst_pad_push (decklinksrc->videosrcpad, buffer);
 
   if (gst_pad_is_linked (decklinksrc->audiosrcpad)) {
     n_samples = audio_frame->GetSampleFrameCount ();
@@ -866,23 +858,35 @@ gst_decklink_src_task (void *priv)
 
     decklinksrc->num_audio_samples += n_samples;
 
-  /* FIXME: proper flow aggregation with video flow */
-    ret = gst_pad_push (decklinksrc->audiosrcpad, audio_buffer);
-    if (!(ret == GST_FLOW_OK || ret == GST_FLOW_NOT_LINKED ||
-            ret == GST_FLOW_FLUSHING)) {
-      GST_ELEMENT_ERROR (decklinksrc, STREAM, FAILED,
-          ("Internal data stream error."),
-          ("stream stopped, reason %s", gst_flow_get_name (ret)));
-      goto pause;
-    }
+    audio_flow = gst_pad_push (decklinksrc->audiosrcpad, audio_buffer);
+  } else {
+    audio_flow = GST_FLOW_NOT_LINKED;
   }
 
+  if (audio_flow == GST_FLOW_NOT_LINKED)
+    flow = video_flow;
+  else if (video_flow == GST_FLOW_NOT_LINKED)
+    flow = audio_flow;
+  else if (video_flow == GST_FLOW_FLUSHING || audio_flow == GST_FLOW_FLUSHING)
+    flow = GST_FLOW_FLUSHING;
+  else if (video_flow < GST_FLOW_EOS)
+    flow = video_flow;
+  else if (audio_flow < GST_FLOW_EOS)
+    flow = audio_flow;
+  else if (video_flow == GST_FLOW_EOS || audio_flow == GST_FLOW_EOS)
+    flow = GST_FLOW_EOS;
+  else
+    flow = video_flow;
+
   if (g_atomic_int_compare_and_exchange (&decklinksrc->pending_eos, TRUE,
       FALSE)) {
     GST_INFO_OBJECT (decklinksrc, "EOS pending");
-    ret = GST_FLOW_EOS;
+    flow = GST_FLOW_EOS;
   }
 
+  if (flow != GST_FLOW_OK)
+    goto pause;
+
 done:
 
   if (audio_frame)
@@ -892,15 +896,15 @@ done:
 
 pause:
   {
-    const gchar *reason = gst_flow_get_name (ret);
+    const gchar *reason = gst_flow_get_name (flow);
     GstEvent *event = NULL;
 
     GST_DEBUG_OBJECT (decklinksrc, "pausing task, reason %s", reason);
     gst_task_pause (decklinksrc->task);
-    if (ret == GST_FLOW_EOS) {
+    if (flow == GST_FLOW_EOS) {
       /* perform EOS logic (very crude, we don't even keep a GstSegment) */
       event = gst_event_new_eos ();
-    } else if (ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS) {
+    } else if (flow == GST_FLOW_NOT_LINKED || flow < GST_FLOW_EOS) {
       event = gst_event_new_eos ();
       /* for fatal errors we post an error message, post the error
        * first so the app knows about the error first.
@@ -910,7 +914,7 @@ pause:
        * a flushing seek. */
       GST_ELEMENT_ERROR (decklinksrc, STREAM, FAILED,
           ("Internal data flow error."),
-          ("streaming task paused, reason %s (%d)", reason, ret));
+          ("streaming task paused, reason %s (%d)", reason, flow));
     }
     if (event != NULL) {
       GST_INFO_OBJECT (decklinksrc->videosrcpad, "pushing EOS event");