From daa6cfbb6a1608f5221abc341bab5ca514010060 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sebastian=20Dr=C3=B6ge?= <sebastian@centricular.com>
Date: Fri, 3 Jul 2020 15:37:36 +0300
Subject: [PATCH] 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.
---
 utils/fallbackswitch/src/fallbacksrc.rs | 66 ++++++++++++++++++++++++-
 1 file changed, 65 insertions(+), 1 deletion(-)

diff --git a/utils/fallbackswitch/src/fallbacksrc.rs b/utils/fallbackswitch/src/fallbacksrc.rs
index a637dc3b..e3c39784 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));