From be6074f4f432c90736734e2997c91f3e07e68834 Mon Sep 17 00:00:00 2001
From: Jochen Henneberg <jochen@centricular.com>
Date: Thu, 28 Nov 2024 14:09:39 +0100
Subject: [PATCH] fmp4mux: Fix state cleanup on flush

State cleanup was incomplete. If a flush happens we do not have any
buffers anymore so all buffer queue related settings must be cleared
on the state. Flush action has been moved into state impl.

The GstAggregatorPad flush() was implemented as well as the pad flush
handler - with different implementations. Unify it.

Fixes #636

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1945>
---
 mux/fmp4/src/fmp4mux/imp.rs | 60 ++++++++++++++++++++-----------------
 1 file changed, 33 insertions(+), 27 deletions(-)

diff --git a/mux/fmp4/src/fmp4mux/imp.rs b/mux/fmp4/src/fmp4mux/imp.rs
index 5c2885ff..530f4f64 100644
--- a/mux/fmp4/src/fmp4mux/imp.rs
+++ b/mux/fmp4/src/fmp4mux/imp.rs
@@ -335,6 +335,15 @@ impl Stream {
     fn caps_or_tag_change(&self) -> bool {
         self.next_caps.is_some() || self.tag_changed
     }
+
+    fn flush(&mut self) {
+        self.queued_gops.clear();
+        self.dts_offset = None;
+        self.current_position = gst::ClockTime::ZERO;
+        self.fragment_filled = false;
+        self.pre_queue.clear();
+        self.running_time_utc_time_mapping = None;
+    }
 }
 
 #[derive(Default)]
@@ -389,6 +398,22 @@ struct State {
     manual_fragment_boundaries: BTreeSet<gst::ClockTime>,
 }
 
+impl State {
+    fn flush(&mut self) {
+        for stream in &mut self.streams {
+            stream.flush();
+        }
+
+        self.current_offset = 0;
+        self.fragment_offsets.clear();
+        self.manual_fragment_boundaries.clear();
+        self.end_pts = None;
+        self.fragment_start_pts = None;
+        self.fragment_end_pts = None;
+        self.chunk_start_pts = None;
+    }
+}
+
 #[derive(Default)]
 pub(crate) struct FMP4Mux {
     state: Mutex<State>,
@@ -3772,23 +3797,8 @@ impl AggregatorImpl for FMP4Mux {
     }
 
     fn flush(&self) -> Result<gst::FlowSuccess, gst::FlowError> {
-        let mut state = self.state.lock().unwrap();
-
-        for stream in &mut state.streams {
-            stream.queued_gops.clear();
-            stream.dts_offset = None;
-            stream.current_position = gst::ClockTime::ZERO;
-            stream.fragment_filled = false;
-            stream.pre_queue.clear();
-            stream.running_time_utc_time_mapping = None;
-        }
-
-        state.current_offset = 0;
-        state.fragment_offsets.clear();
-        state.manual_fragment_boundaries.clear();
-
-        drop(state);
-
+        gst::trace!(CAT, imp = self, "Flush");
+        self.state.lock().unwrap().flush();
         self.parent_flush()
     }
 
@@ -4544,16 +4554,12 @@ impl AggregatorPadImpl for FMP4MuxPad {
         let mux = aggregator.downcast_ref::<super::FMP4Mux>().unwrap();
         let mut mux_state = mux.imp().state.lock().unwrap();
 
-        for stream in &mut mux_state.streams {
-            if stream.sinkpad == *self.obj() {
-                stream.queued_gops.clear();
-                stream.dts_offset = None;
-                stream.current_position = gst::ClockTime::ZERO;
-                stream.fragment_filled = false;
-                stream.pre_queue.clear();
-                stream.running_time_utc_time_mapping = None;
-                break;
-            }
+        if let Some(stream) = mux_state
+            .streams
+            .iter_mut()
+            .find(|s| s.sinkpad == *self.obj())
+        {
+            stream.flush();
         }
 
         drop(mux_state);