From 0a45f776e0a38b994b3e49bab390c64602f2de8b Mon Sep 17 00:00:00 2001
From: "Jan Alexander Steffens (heftig)" <jan.steffens@ltnglobal.com>
Date: Tue, 24 Oct 2023 23:09:11 +0200
Subject: [PATCH] livesync: Simplify start_src_task and src_loop

This should effect no change in behavior.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1369>
---
 utils/livesync/src/livesync/imp.rs | 39 ++++++++++--------------------
 1 file changed, 13 insertions(+), 26 deletions(-)

diff --git a/utils/livesync/src/livesync/imp.rs b/utils/livesync/src/livesync/imp.rs
index 72d54573..418862db 100644
--- a/utils/livesync/src/livesync/imp.rs
+++ b/utils/livesync/src/livesync/imp.rs
@@ -968,31 +968,17 @@ impl LiveSync {
         Ok(gst::FlowSuccess::Ok)
     }
 
-    fn start_src_task(&self) -> Result<(), glib::BoolError> {
-        self.srcpad.start_task({
-            let pad = self.srcpad.downgrade();
-            move || {
-                let pad = pad.upgrade().unwrap();
-                let parent = pad.parent_element().unwrap();
-                let livesync = parent.downcast_ref::<super::LiveSync>().unwrap();
-                let ret = livesync.imp().src_loop(&pad);
-
-                if !ret {
-                    gst::log!(CAT, obj: &parent, "Loop stopping");
-                    let _ = pad.pause_task();
-                }
-            }
-        })
+    fn start_src_task(&self, state: &mut State) -> Result<(), glib::BoolError> {
+        let imp = self.ref_counted();
+        self.srcpad.start_task(move || imp.src_loop())
     }
 
-    fn src_loop(&self, pad: &gst::Pad) -> bool {
-        let mut err = match self.src_loop_inner() {
-            Ok(_) => return true,
-            Err(e) => e,
+    fn src_loop(&self) {
+        let Err(mut err) = self.src_loop_inner() else {
+            return;
         };
-        let eos;
 
-        {
+        let eos = {
             let mut state = self.state.lock();
 
             match state.srcresult {
@@ -1002,21 +988,22 @@ impl LiveSync {
                 // Communicate our flow return
                 Ok(_) => state.srcresult = Err(err),
             }
-            eos = state.eos;
             state.clock_id = None;
-
             self.cond.notify_all();
-        }
+
+            state.eos
+        };
 
         // Following GstQueue's behavior:
         // > let app know about us giving up if upstream is not expected to do so
         // > EOS is already taken care of elsewhere
         if eos && !matches!(err, gst::FlowError::Flushing | gst::FlowError::Eos) {
             self.flow_error(err);
-            pad.push_event(gst::event::Eos::new());
+            self.srcpad.push_event(gst::event::Eos::new());
         }
 
-        false
+        gst::log!(CAT, imp: self, "Loop stopping");
+        let _ = self.srcpad.pause_task();
     }
 
     fn src_loop_inner(&self) -> Result<gst::FlowSuccess, gst::FlowError> {