utils/fallbacksrc: Add property to optionally restart the source on EOS

EOS might be unexpected and the stream might be supposed to run forever,
in which case it should transparently be restarted on EOS.

1 files changed, 65 insertions, 1 deletions

diff --git a/utils/fallbackswitch/src/fallbacksrc.rs b/utils/fallbackswitch/src/fallbacksrc.rs
index a637dc3bf..e3c397840 100644
--- a/utils/fallbackswitch/src/fallbacksrc.rs
+++ b/utils/fallbackswitch/src/fallbacksrc.rs
@@ -45,6 +45,7 @@ struct Settings {
     timeout: u64,
     restart_timeout: u64,
     retry_timeout: u64,
+    restart_on_eos: bool,
 }
 
 impl Default for Settings {
@@ -58,6 +59,7 @@ impl Default for Settings {
             timeout: 5 * gst::SECOND_VAL,
             restart_timeout: 5 * gst::SECOND_VAL,
             retry_timeout: 60 * gst::SECOND_VAL,
+            restart_on_eos: false,
         }
     }
 }
@@ -144,7 +146,7 @@ enum Status {
     Running,
 }
 
-static PROPERTIES: [subclass::Property; 9] = [
+static PROPERTIES: [subclass::Property; 10] = [
     subclass::Property("enable-audio", |name| {
         glib::ParamSpec::boolean(
             name,
@@ -217,6 +219,15 @@ static PROPERTIES: [subclass::Property; 9] = [
             glib::ParamFlags::READWRITE,
         )
     }),
+    subclass::Property("restart-on-eos", |name| {
+        glib::ParamSpec::boolean(
+            name,
+            "Restart on EOS",
+            "Restart source on EOS",
+            false,
+            glib::ParamFlags::READWRITE,
+        )
+    }),
     subclass::Property("status", |name| {
         glib::ParamSpec::enum_(
             name,
@@ -378,6 +389,18 @@ impl ObjectImpl for FallbackSrc {
                 );
                 settings.retry_timeout = new_value;
             }
+            subclass::Property("restart-on-eos", ..) => {
+                let mut settings = self.settings.lock().unwrap();
+                let new_value = value.get_some().expect("type checked upstream");
+                gst_info!(
+                    CAT,
+                    obj: element,
+                    "Changing restart-on-eos from {:?} to {:?}",
+                    settings.restart_on_eos,
+                    new_value,
+                );
+                settings.restart_on_eos = new_value;
+            }
             _ => unimplemented!(),
         }
     }
@@ -421,6 +444,10 @@ impl ObjectImpl for FallbackSrc {
                 let settings = self.settings.lock().unwrap();
                 Ok(settings.retry_timeout.to_value())
             }
+            subclass::Property("restart-on-eos", ..) => {
+                let settings = self.settings.lock().unwrap();
+                Ok(settings.restart_on_eos.to_value())
+            }
             subclass::Property("status", ..) => {
                 let state_guard = self.state.lock().unwrap();
 
@@ -1159,6 +1186,43 @@ impl FallbackSrc {
             )
         })?;
 
+        if state.settings.restart_on_eos {
+            let element_weak = element.downgrade();
+            pad.add_probe(gst::PadProbeType::EVENT_DOWNSTREAM, move |pad, info| {
+                let element = match element_weak.upgrade() {
+                    None => return gst::PadProbeReturn::Ok,
+                    Some(element) => element,
+                };
+
+                let src = FallbackSrc::from_instance(&element);
+
+                match info.data {
+                    Some(gst::PadProbeData::Event(ref ev))
+                        if ev.get_type() == gst::EventType::Eos =>
+                    {
+                        gst_debug!(
+                            CAT,
+                            obj: &element,
+                            "Received EOS from source on pad {}, restarting",
+                            pad.get_name()
+                        );
+
+                        let mut state_guard = src.state.lock().unwrap();
+                        let state = match &mut *state_guard {
+                            None => {
+                                return gst::PadProbeReturn::Ok;
+                            }
+                            Some(state) => state,
+                        };
+                        src.handle_source_error(&element, state);
+
+                        gst::PadProbeReturn::Drop
+                    }
+                    _ => gst::PadProbeReturn::Ok,
+                }
+            });
+        }
+
         stream.source_srcpad = Some(pad.clone());
         stream.source_srcpad_block = Some(self.add_pad_probe(element, pad));