From 207196a0334da74c4db9db7c565d882cb9ebc07d 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/1968>
---
 mux/fmp4/src/fmp4mux/imp.rs | 59 +++++++++++++++++++++----------------
 1 file changed, 33 insertions(+), 26 deletions(-)

diff --git a/mux/fmp4/src/fmp4mux/imp.rs b/mux/fmp4/src/fmp4mux/imp.rs
index 8e4598652..a27d3b0e7 100644
--- a/mux/fmp4/src/fmp4mux/imp.rs
+++ b/mux/fmp4/src/fmp4mux/imp.rs
@@ -230,6 +230,17 @@ struct Stream {
     extra_header_data: Option<Vec<u8>>,
 }
 
+impl Stream {
+    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)]
 struct State {
     /// Currently configured streams.
@@ -273,6 +284,20 @@ struct State {
     sent_headers: bool,
 }
 
+impl State {
+    fn flush(&mut self) {
+        for stream in &mut self.streams {
+            stream.flush();
+        }
+
+        self.current_offset = 0;
+        self.fragment_offsets.clear();
+        self.end_pts = None;
+        self.fragment_start_pts = None;
+        self.chunk_start_pts = None;
+    }
+}
+
 #[derive(Default)]
 pub(crate) struct FMP4Mux {
     state: Mutex<State>,
@@ -3302,22 +3327,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();
-
-        drop(state);
-
+        gst::trace!(CAT, imp = self, "Flush");
+        self.state.lock().unwrap().flush();
         self.parent_flush()
     }
 
@@ -4011,16 +4022,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);