From d69d0f87384ec861a325b9daca1659d69c922d01 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sebastian=20Dr=C3=B6ge?= <sebastian@centricular.com>
Date: Tue, 17 Sep 2024 09:52:54 +0300
Subject: [PATCH] cea708mux: Stop with EOS if all pads are EOS instead of
 continuing forever

Also don't drop buffers if multiple tries are needed for aggregating
because some pads are not ready yet.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1774>
---
 video/closedcaption/src/cea708mux/imp.rs | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/video/closedcaption/src/cea708mux/imp.rs b/video/closedcaption/src/cea708mux/imp.rs
index e2d604ae..87677de7 100644
--- a/video/closedcaption/src/cea708mux/imp.rs
+++ b/video/closedcaption/src/cea708mux/imp.rs
@@ -127,6 +127,7 @@ impl AggregatorImpl for Cea708Mux {
             .nseconds();
         let end_running_time = start_running_time + duration;
         let mut need_data = false;
+        let mut all_eos = true;
         gst::debug!(
             CAT,
             imp = self,
@@ -144,15 +145,20 @@ impl AggregatorImpl for Cea708Mux {
                 .expect("Not a Cea708MuxSinkPad?!")
         }) {
             let mut pad_state = pad.imp().pad_state.lock().unwrap();
-            pad_state.pending_buffer = None;
             // any data we currently have stored
             let have_pending = pad_state
                 .pending_services
                 .values()
                 .any(|codes| !codes.is_empty());
+
             if pad.is_eos() {
+                if pad_state.pending_buffer.is_some() {
+                    all_eos = false;
+                }
                 continue;
             }
+            all_eos = false;
+
             let buffer = if let Some(buffer) = pad.peek_buffer() {
                 buffer
             } else {
@@ -229,6 +235,9 @@ impl AggregatorImpl for Cea708Mux {
         if need_data && !timeout {
             return Err(gst_base::AGGREGATOR_FLOW_NEED_DATA);
         }
+        if all_eos {
+            return Err(gst::FlowError::Eos);
+        }
 
         self.obj()
             .selected_samples(start_running_time, None, duration, None);