fallbackswitch: Replace with priorityswitch

fallbackswitch now supports multiple sink pads, and on a timeout of the
active pad, it will automatically switch to the next lowest priority pad
that has data available.

fallbackswitch sink pads follow the `sink_%u` template and have
`priority` as a pad property.

Co-authored-by: Vivia Nikolaidou <vivia.nikolaidou@ltnglobal.com>

utils/fallbackswitch/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "gst-plugin-fallbackswitch"
 version = "0.9.0"
-authors = ["Sebastian Dröge <sebastian@centricular.com>"]
+authors = ["Sebastian Dröge <sebastian@centricular.com>", "Jan Schmidt <jan@centricular.com>"]
 repository = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs"
 license = "MPL-2.0"
 edition = "2021"
@@ -11,12 +11,13 @@ description = "Fallback Switcher Plugin"
 [dependencies]
 libc = { version = "0.2", optional = true }
 gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_14"] }
-gst-base = { package = "gstreamer-base", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_18"] }
+gst-base = { package = "gstreamer-base", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_14"] }
 gst-audio = { package = "gstreamer-audio", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_14"] }
 gst-video = { package = "gstreamer-video", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_14"] }
 gtk = { git = "https://github.com/gtk-rs/gtk3-rs", optional = true }
 gio = { git = "https://github.com/gtk-rs/gtk-rs-core", optional = true }
 once_cell = "1.0"
+parking_lot = "0.12"
 
 [dev-dependencies]
 gst-app = { package = "gstreamer-app", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs",  features = ["v1_14"]}
@@ -37,7 +38,6 @@ gst-plugin-version-helper = { path="../../version-helper" }
 
 [features]
 default = ["libc"]
-v1_20 = ["gst/v1_20"]
 # We already use 1.14 which is new enough for static build
 static = []
 capi = []

utils/fallbackswitch/examples/gtk_fallbackswitch.rs

@@ -74,11 +74,12 @@ fn create_pipeline() -> (gst::Pipeline, gst::Pad, gst::Element, gtk::Widget) {
         ])
         .unwrap();
 
+    /* The first pad requested will be automatically preferred */
     video_src
-        .link_pads(Some("src"), &fallbackswitch, Some("sink"))
+        .link_pads(Some("src"), &fallbackswitch, Some("sink_%u"))
         .unwrap();
     fallback_video_src
-        .link_pads(Some("src"), &fallbackswitch, Some("fallback_sink"))
+        .link_pads(Some("src"), &fallbackswitch, Some("sink_%u"))
         .unwrap();
     fallbackswitch
         .link_pads(Some("src"), &decodebin, Some("sink"))

utils/fallbackswitch/src/fallbacksrc/imp.rs

@@ -10,8 +10,8 @@ use gst::glib;
 use gst::prelude::*;
 use gst::subclass::prelude::*;
 
+use parking_lot::Mutex;
 use std::mem;
-use std::sync::Mutex;
 use std::time::Instant;
 
 use once_cell::sync::Lazy;
@@ -316,7 +316,7 @@ impl ObjectImpl for FallbackSrc {
     ) {
         match pspec.name() {
             "enable-audio" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -328,7 +328,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.enable_audio = new_value;
             }
             "enable-video" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -340,7 +340,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.enable_video = new_value;
             }
             "uri" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -352,7 +352,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.uri = new_value;
             }
             "source" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -364,7 +364,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.source = new_value;
             }
             "fallback-uri" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -376,7 +376,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.fallback_uri = new_value;
             }
             "timeout" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -388,7 +388,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.timeout = new_value;
             }
             "restart-timeout" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -400,7 +400,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.restart_timeout = new_value;
             }
             "retry-timeout" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -412,7 +412,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.retry_timeout = new_value;
             }
             "restart-on-eos" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -424,7 +424,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.restart_on_eos = new_value;
             }
             "min-latency" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -436,7 +436,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.min_latency = new_value;
             }
             "buffer-duration" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -448,7 +448,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.buffer_duration = new_value;
             }
             "immediate-fallback" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -460,7 +460,7 @@ impl ObjectImpl for FallbackSrc {
                 settings.immediate_fallback = new_value;
             }
             "manual-unblock" => {
-                let mut settings = self.settings.lock().unwrap();
+                let mut settings = self.settings.lock();
                 let new_value = value.get().expect("type checked upstream");
                 gst::info!(
                     CAT,
@@ -481,43 +481,43 @@ impl ObjectImpl for FallbackSrc {
     fn property(&self, _obj: &Self::Type, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
         match pspec.name() {
             "enable-audio" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.enable_audio.to_value()
             }
             "enable-video" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.enable_video.to_value()
             }
             "uri" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.uri.to_value()
             }
             "source" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.source.to_value()
             }
             "fallback-uri" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.fallback_uri.to_value()
             }
             "timeout" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.timeout.to_value()
             }
             "restart-timeout" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.restart_timeout.to_value()
             }
             "retry-timeout" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.retry_timeout.to_value()
             }
             "restart-on-eos" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.restart_on_eos.to_value()
             }
             "status" => {
-                let state_guard = self.state.lock().unwrap();
+                let state_guard = self.state.lock();
 
                 // If we have no state then we'r stopped
                 let state = match &*state_guard {
@@ -569,20 +569,20 @@ impl ObjectImpl for FallbackSrc {
                 Status::Running.to_value()
             }
             "min-latency" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.min_latency.to_value()
             }
             "buffer-duration" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.buffer_duration.to_value()
             }
             "statistics" => self.stats().to_value(),
             "immediate-fallback" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.immediate_fallback.to_value()
             }
             "manual-unblock" => {
-                let settings = self.settings.lock().unwrap();
+                let settings = self.settings.lock();
                 settings.manual_unblock.to_value()
             }
             _ => unimplemented!(),
@@ -612,7 +612,7 @@ impl ObjectImpl for FallbackSrc {
                     .class_handler(|_token, args| {
                         let element = args[0].get::<super::FallbackSrc>().expect("signal arg");
                         let src = element.imp();
-                        let mut state_guard = src.state.lock().unwrap();
+                        let mut state_guard = src.state.lock();
                         let state = match &mut *state_guard {
                             None => {
                                 return None;
@@ -730,7 +730,7 @@ impl ElementImpl for FallbackSrc {
             gst::EventView::Eos(..) => {
                 gst::debug!(CAT, "Handling element-level EOS, forwarding to all streams");
 
-                let mut state_guard = self.state.lock().unwrap();
+                let mut state_guard = self.state.lock();
                 let state = match &mut *state_guard {
                     None => {
                         return true;
@@ -942,10 +942,17 @@ impl FallbackSrc {
         switch.set_property("min-upstream-latency", min_latency.nseconds());
         switch.set_property("immediate-fallback", immediate_fallback);
 
-        gst::Element::link_pads(&fallback_input, Some("src"), &switch, Some("fallback_sink"))
+        let fallback_srcpad = fallback_input.static_pad("src").unwrap();
-            .unwrap();
+        let switch_fallbacksink = switch.request_pad_simple("sink_%u").unwrap();
+        fallback_srcpad.link(&switch_fallbacksink).unwrap();
+        switch_fallbacksink.set_property("priority", 1u32);
+
         gst::Element::link_pads(&clocksync_queue, Some("src"), &clocksync, Some("sink")).unwrap();
-        gst::Element::link_pads(&clocksync, Some("src"), &switch, Some("sink")).unwrap();
+
+        let clocksync_srcpad = clocksync.static_pad("src").unwrap();
+        let switch_mainsink = switch.request_pad_simple("sink_%u").unwrap();
+        clocksync_srcpad.link(&switch_mainsink).unwrap();
+        switch_mainsink.set_property("priority", 0u32);
         // clocksync_queue sink pad is not connected to anything yet at this point!
 
         let srcpad = switch.static_pad("src").unwrap();
@@ -984,12 +991,12 @@ impl FallbackSrc {
 
     fn start(&self, element: &super::FallbackSrc) -> Result<(), gst::StateChangeError> {
         gst::debug!(CAT, obj: element, "Starting");
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         if state_guard.is_some() {
             return Err(gst::StateChangeError);
         }
 
-        let settings = self.settings.lock().unwrap().clone();
+        let settings = self.settings.lock().clone();
         let configured_source = match settings
             .uri
             .as_ref()
@@ -1082,7 +1089,7 @@ impl FallbackSrc {
 
     fn stop(&self, element: &super::FallbackSrc) {
         gst::debug!(CAT, obj: element, "Stopping");
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let mut state = match state_guard.take() {
             Some(state) => state,
             None => return,
@@ -1138,7 +1145,7 @@ impl FallbackSrc {
 
     fn change_source_state(&self, element: &super::FallbackSrc, transition: gst::StateChange) {
         gst::debug!(CAT, obj: element, "Changing source state: {:?}", transition);
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             Some(state) => state,
             None => return,
@@ -1174,7 +1181,7 @@ impl FallbackSrc {
                 // Try again later if we're not shutting down
                 if transition != gst::StateChange::ReadyToNull {
                     let _ = source.set_state(gst::State::Null);
-                    let mut state_guard = self.state.lock().unwrap();
+                    let mut state_guard = self.state.lock();
                     let state = state_guard.as_mut().expect("no state");
                     self.handle_source_error(element, state, RetryReason::StateChangeFailure);
                     drop(state_guard);
@@ -1189,7 +1196,7 @@ impl FallbackSrc {
                     res
                 );
 
-                let mut state_guard = self.state.lock().unwrap();
+                let mut state_guard = self.state.lock();
                 let state = state_guard.as_mut().expect("no state");
 
                 // Remember if the source is live
@@ -1216,7 +1223,7 @@ impl FallbackSrc {
     ) -> Result<gst::FlowSuccess, gst::FlowError> {
         let res = gst::ProxyPad::chain_default(pad, Some(element), buffer);
 
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => return res,
             Some(state) => state,
@@ -1232,7 +1239,7 @@ impl FallbackSrc {
     ) -> Result<(), gst::ErrorMessage> {
         gst::debug!(CAT, obj: element, "Pad {} added to source", pad.name(),);
 
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => {
                 return Ok(());
@@ -1360,7 +1367,7 @@ impl FallbackSrc {
                         pad.name()
                     );
 
-                    let mut state_guard = src.state.lock().unwrap();
+                    let mut state_guard = src.state.lock();
                     let state = match &mut *state_guard {
                         None => {
                             return gst::PadProbeReturn::Ok;
@@ -1469,7 +1476,7 @@ impl FallbackSrc {
         pad: &gst::Pad,
         pts: impl Into<Option<gst::ClockTime>>,
     ) -> Result<(), gst::ErrorMessage> {
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => {
                 return Ok(());
@@ -1810,7 +1817,7 @@ impl FallbackSrc {
     fn handle_source_pad_removed(&self, element: &super::FallbackSrc, pad: &gst::Pad) {
         gst::debug!(CAT, obj: element, "Pad {} removed from source", pad.name());
 
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => {
                 return;
@@ -1846,7 +1853,7 @@ impl FallbackSrc {
     }
 
     fn handle_buffering(&self, element: &super::FallbackSrc, m: &gst::message::Buffering) {
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => {
                 return;
@@ -1890,7 +1897,7 @@ impl FallbackSrc {
         element: &super::FallbackSrc,
         m: &gst::message::StreamsSelected,
     ) {
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => {
                 return;
@@ -1950,7 +1957,7 @@ impl FallbackSrc {
     }
 
     fn handle_error(&self, element: &super::FallbackSrc, m: &gst::message::Error) -> bool {
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => {
                 return false;
@@ -2074,7 +2081,7 @@ impl FallbackSrc {
 
             // Remove blocking pad probes if they are still there as otherwise shutting down the
             // source will deadlock on the probes.
-            let mut state_guard = src.state.lock().unwrap();
+            let mut state_guard = src.state.lock();
             let state = match &mut *state_guard {
                 None
                 | Some(State {
@@ -2115,7 +2122,7 @@ impl FallbackSrc {
 
             // Sleep for 1s before retrying
 
-            let mut state_guard = src.state.lock().unwrap();
+            let mut state_guard = src.state.lock();
             let state = match &mut *state_guard {
                 None
                 | Some(State {
@@ -2154,7 +2161,7 @@ impl FallbackSrc {
                     element.call_async(|element| {
                         let src = element.imp();
 
-                        let mut state_guard = src.state.lock().unwrap();
+                        let mut state_guard = src.state.lock();
                         let state = match &mut *state_guard {
                             None
                             | Some(State {
@@ -2211,7 +2218,7 @@ impl FallbackSrc {
                         if source.sync_state_with_parent().is_err() {
                             gst::error!(CAT, obj: element, "Source failed to change state");
                             let _ = source.set_state(gst::State::Null);
-                            let mut state_guard = src.state.lock().unwrap();
+                            let mut state_guard = src.state.lock();
                             let state = state_guard.as_mut().expect("no state");
                             src.handle_source_error(
                                 element,
@@ -2221,7 +2228,7 @@ impl FallbackSrc {
                             drop(state_guard);
                             element.notify("statistics");
                         } else {
-                            let mut state_guard = src.state.lock().unwrap();
+                            let mut state_guard = src.state.lock();
                             let state = state_guard.as_mut().expect("no state");
                             assert!(state.source_restart_timeout.is_none());
                             src.schedule_source_restart_timeout(
@@ -2293,7 +2300,7 @@ impl FallbackSrc {
                     let src = element.imp();
 
                     gst::debug!(CAT, obj: element, "Source restart timeout triggered");
-                    let mut state_guard = src.state.lock().unwrap();
+                    let mut state_guard = src.state.lock();
                     let state = match &mut *state_guard {
                         None => {
                             gst::debug!(CAT, obj: element, "Restarting source not needed anymore");
@@ -2358,7 +2365,7 @@ impl FallbackSrc {
                     .audio_stream
                     .as_ref()
                     .and_then(|s| s.switch.property::<Option<gst::Pad>>("active-pad"))
-                    .map(|p| p.name() == "fallback_sink")
+                    .map(|p| p.property::<u32>("priority") != 0)
                     .unwrap_or(true))
             || (have_video
                 && state.video_stream.is_some()
@@ -2366,12 +2373,12 @@ impl FallbackSrc {
                     .video_stream
                     .as_ref()
                     .and_then(|s| s.switch.property::<Option<gst::Pad>>("active-pad"))
-                    .map(|p| p.name() == "fallback_sink")
+                    .map(|p| p.property::<u32>("priority") != 0)
                     .unwrap_or(true))
     }
 
     fn handle_switch_active_pad_change(&self, element: &super::FallbackSrc) {
-        let mut state_guard = self.state.lock().unwrap();
+        let mut state_guard = self.state.lock();
         let state = match &mut *state_guard {
             None => {
                 return;
@@ -2404,7 +2411,7 @@ impl FallbackSrc {
     }
 
     fn stats(&self) -> gst::Structure {
-        let state_guard = self.state.lock().unwrap();
+        let state_guard = self.state.lock();
 
         let state = match &*state_guard {
             None => return Stats::default().to_structure(),

utils/fallbackswitch/src/fallbackswitch/mod.rs

@@ -1,4 +1,7 @@
-// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2020 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2021 Jan Schmidt <jan@centricular.com>
+// Copyright (C) 2020 Mathieu Duponchelle <mathieu@centricular.com>
+// Copyright (C) 2022 Vivia Nikolaidou <vivia.nikolaidou@ltnglobal.com>
 //
 // This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
 // If a copy of the MPL was not distributed with this file, You can obtain one at
@@ -11,12 +14,19 @@ use gst::prelude::*;
 
 mod imp;
 
-pub use imp::StreamHealth;
+// The public Rust wrapper type for our element
-
 glib::wrapper! {
-    pub struct FallbackSwitch(ObjectSubclass<imp::FallbackSwitch>) @extends gst_base::Aggregator, gst::Element, gst::Object;
+    pub struct FallbackSwitch(ObjectSubclass<imp::FallbackSwitch>) @extends gst::Element, gst::Object, @implements gst::ChildProxy;
 }
 
+// The public Rust wrapper type for our sink pad
+glib::wrapper! {
+    pub struct FallbackSwitchSinkPad(ObjectSubclass<imp::FallbackSwitchSinkPad>) @extends gst::Pad, gst::Object;
+}
+
+// Registers the type for our element, and then registers in GStreamer under
+// the name "fallbackswitch" for being able to instantiate it via e.g.
+// gst::ElementFactory::make().
 pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
     gst::Element::register(
         Some(plugin),

utils/fallbackswitch/src/lib.rs

@@ -13,7 +13,6 @@ mod fallbacksrc;
 mod fallbackswitch;
 
 pub use fallbacksrc::{RetryReason, Status};
-pub use fallbackswitch::StreamHealth;
 
 fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
     fallbacksrc::register(plugin)?;

utils/fallbackswitch/tests/fallbackswitch.rs

@@ -1,4 +1,5 @@
 // Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2021 Jan Schmidt <jan@centricular.com>
 //
 // This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
 // If a copy of the MPL was not distributed with this file, You can obtain one at
@@ -6,10 +7,13 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
+use gst::debug;
 use gst::prelude::*;
 
 use once_cell::sync::Lazy;
 
+const LATENCY: gst::ClockTime = gst::ClockTime::from_mseconds(10);
+
 static TEST_CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
     gst::DebugCategory::new(
         "fallbackswitch-test",
@@ -44,20 +48,20 @@ macro_rules! assert_buffer {
 
 #[test]
 fn test_no_fallback_no_drops() {
-    let pipeline = setup_pipeline(None);
+    let pipeline = setup_pipeline(None, None, None);
 
     push_buffer(&pipeline, gst::ClockTime::ZERO);
-    set_time(&pipeline, gst::ClockTime::ZERO);
+    set_time(&pipeline, gst::ClockTime::ZERO + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::ZERO));
 
     push_buffer(&pipeline, gst::ClockTime::SECOND);
-    set_time(&pipeline, gst::ClockTime::SECOND);
+    set_time(&pipeline, gst::ClockTime::SECOND + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::SECOND));
 
     push_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
-    set_time(&pipeline, 2 * gst::ClockTime::SECOND);
+    set_time(&pipeline, 2 * gst::ClockTime::SECOND + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(2 * gst::ClockTime::SECOND));
 
@@ -78,23 +82,23 @@ fn test_no_drops_not_live() {
 }
 
 fn test_no_drops(live: bool) {
-    let pipeline = setup_pipeline(Some(live));
+    let pipeline = setup_pipeline(Some(live), None, None);
 
     push_buffer(&pipeline, gst::ClockTime::ZERO);
     push_fallback_buffer(&pipeline, gst::ClockTime::ZERO);
-    set_time(&pipeline, gst::ClockTime::ZERO);
+    set_time(&pipeline, gst::ClockTime::ZERO + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::ZERO));
 
     push_fallback_buffer(&pipeline, gst::ClockTime::SECOND);
     push_buffer(&pipeline, gst::ClockTime::SECOND);
-    set_time(&pipeline, gst::ClockTime::SECOND);
+    set_time(&pipeline, gst::ClockTime::SECOND + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::SECOND));
 
     push_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
     push_fallback_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
-    set_time(&pipeline, 2 * gst::ClockTime::SECOND);
+    set_time(&pipeline, 2 * gst::ClockTime::SECOND + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(2 * gst::ClockTime::SECOND));
 
@@ -116,22 +120,22 @@ fn test_no_drops_but_no_fallback_frames_not_live() {
 }
 
 fn test_no_drops_but_no_fallback_frames(live: bool) {
-    let pipeline = setup_pipeline(Some(live));
+    let pipeline = setup_pipeline(Some(live), None, None);
 
     push_buffer(&pipeline, gst::ClockTime::ZERO);
     // +10ms needed here because the immediate timeout will be always at running time 0, but
     // aggregator also adds the latency to it so we end up at 10ms instead.
-    set_time(&pipeline, 10 * gst::ClockTime::MSECOND);
+    set_time(&pipeline, LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::ZERO));
 
     push_buffer(&pipeline, gst::ClockTime::SECOND);
-    set_time(&pipeline, gst::ClockTime::SECOND);
+    set_time(&pipeline, gst::ClockTime::SECOND + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::SECOND));
 
     push_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
-    set_time(&pipeline, 2 * gst::ClockTime::SECOND);
+    set_time(&pipeline, 2 * gst::ClockTime::SECOND + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(2 * gst::ClockTime::SECOND));
 
@@ -153,11 +157,11 @@ fn test_short_drop_not_live() {
 }
 
 fn test_short_drop(live: bool) {
-    let pipeline = setup_pipeline(Some(live));
+    let pipeline = setup_pipeline(Some(live), None, None);
 
     push_buffer(&pipeline, gst::ClockTime::ZERO);
     push_fallback_buffer(&pipeline, gst::ClockTime::ZERO);
-    set_time(&pipeline, gst::ClockTime::ZERO);
+    set_time(&pipeline, gst::ClockTime::ZERO + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::ZERO));
 
@@ -172,7 +176,7 @@ fn test_short_drop(live: bool) {
 
     push_fallback_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
     push_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
-    set_time(&pipeline, 2 * gst::ClockTime::SECOND);
+    set_time(&pipeline, 2 * gst::ClockTime::SECOND + LATENCY);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(2 * gst::ClockTime::SECOND));
 
@@ -194,7 +198,7 @@ fn test_long_drop_and_eos_not_live() {
 }
 
 fn test_long_drop_and_eos(live: bool) {
-    let pipeline = setup_pipeline(Some(live));
+    let pipeline = setup_pipeline(Some(live), None, None);
 
     // Produce the first frame
     push_buffer(&pipeline, gst::ClockTime::ZERO);
@@ -256,7 +260,9 @@ fn test_long_drop_and_recover_not_live() {
 }
 
 fn test_long_drop_and_recover(live: bool) {
-    let pipeline = setup_pipeline(Some(live));
+    let pipeline = setup_pipeline(Some(live), None, None);
+    let switch = pipeline.by_name("switch").unwrap();
+    let mainsink = switch.static_pad("sink_0").unwrap();
 
     // Produce the first frame
     push_buffer(&pipeline, gst::ClockTime::ZERO);
@@ -264,6 +270,7 @@ fn test_long_drop_and_recover(live: bool) {
     set_time(&pipeline, gst::ClockTime::ZERO);
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(gst::ClockTime::ZERO));
+    assert!(mainsink.property::<bool>("is-healthy"));
 
     // Produce a second frame but only from the fallback source
     push_fallback_buffer(&pipeline, gst::ClockTime::SECOND);
@@ -301,22 +308,33 @@ fn test_long_drop_and_recover(live: bool) {
 
     // Produce a sixth frame from the normal source
     push_buffer(&pipeline, 5 * gst::ClockTime::SECOND);
-    push_fallback_buffer(&pipeline, 5 * gst::ClockTime::SECOND);
+    set_time(
-    set_time(&pipeline, 5 * gst::ClockTime::SECOND);
+        &pipeline,
+        5 * gst::ClockTime::SECOND + 10 * gst::ClockTime::MSECOND,
+    );
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(5 * gst::ClockTime::SECOND));
+    assert!(!mainsink.property::<bool>("is-healthy"));
+    drop(mainsink);
+    drop(switch);
 
     // Produce a seventh frame from the normal source but no fallback.
     // This should still be output immediately
     push_buffer(&pipeline, 6 * gst::ClockTime::SECOND);
-    set_time(&pipeline, 6 * gst::ClockTime::SECOND);
+    set_time(
+        &pipeline,
+        6 * gst::ClockTime::SECOND + 10 * gst::ClockTime::MSECOND,
+    );
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(6 * gst::ClockTime::SECOND));
 
     // Produce a eight frame from the normal source
     push_buffer(&pipeline, 7 * gst::ClockTime::SECOND);
     push_fallback_buffer(&pipeline, 7 * gst::ClockTime::SECOND);
-    set_time(&pipeline, 7 * gst::ClockTime::SECOND);
+    set_time(
+        &pipeline,
+        7 * gst::ClockTime::SECOND + 10 * gst::ClockTime::MSECOND,
+    );
     let buffer = pull_buffer(&pipeline);
     assert_buffer!(buffer, Some(7 * gst::ClockTime::SECOND));
 
@@ -328,6 +346,144 @@ fn test_long_drop_and_recover(live: bool) {
     stop_pipeline(pipeline);
 }
 
+#[test]
+fn test_initial_timeout_live() {
+    test_initial_timeout(true);
+}
+
+#[test]
+fn test_initial_timeout_not_live() {
+    test_initial_timeout(false);
+}
+
+fn test_initial_timeout(live: bool) {
+    let pipeline = setup_pipeline(Some(live), None, None);
+
+    // Produce the first frame but only from the fallback source
+    push_fallback_buffer(&pipeline, gst::ClockTime::ZERO);
+    set_time(&pipeline, gst::ClockTime::ZERO);
+
+    // Produce a second frame but only from the fallback source
+    push_fallback_buffer(&pipeline, gst::ClockTime::SECOND);
+    set_time(
+        &pipeline,
+        gst::ClockTime::SECOND + 10 * gst::ClockTime::MSECOND,
+    );
+
+    // Produce a third frame but only from the fallback source
+    push_fallback_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
+    set_time(
+        &pipeline,
+        2 * gst::ClockTime::SECOND + 10 * gst::ClockTime::MSECOND,
+    );
+
+    // Produce a fourth frame but only from the fallback source
+    // This should be output now
+    push_fallback_buffer(&pipeline, 3 * gst::ClockTime::SECOND);
+    set_time(
+        &pipeline,
+        3 * gst::ClockTime::SECOND + 10 * gst::ClockTime::MSECOND,
+    );
+    let buffer = pull_buffer(&pipeline);
+    assert_fallback_buffer!(buffer, Some(3 * gst::ClockTime::SECOND));
+
+    // Produce a fifth frame but only from the fallback source
+    // This should be output now
+    push_fallback_buffer(&pipeline, 4 * gst::ClockTime::SECOND);
+    set_time(
+        &pipeline,
+        4 * gst::ClockTime::SECOND + 10 * gst::ClockTime::MSECOND,
+    );
+    let buffer = pull_buffer(&pipeline);
+    assert_fallback_buffer!(buffer, Some(4 * gst::ClockTime::SECOND));
+
+    // Wait for EOS to arrive at appsink
+    push_eos(&pipeline);
+    push_fallback_eos(&pipeline);
+    wait_eos(&pipeline);
+
+    stop_pipeline(pipeline);
+}
+
+#[test]
+fn test_immediate_fallback_live() {
+    test_immediate_fallback(true);
+}
+
+#[test]
+fn test_immediate_fallback_not_live() {
+    test_immediate_fallback(false);
+}
+
+fn test_immediate_fallback(live: bool) {
+    let pipeline = setup_pipeline(Some(live), Some(true), None);
+
+    // Produce the first frame but only from the fallback source
+    push_fallback_buffer(&pipeline, gst::ClockTime::ZERO);
+    set_time(&pipeline, gst::ClockTime::ZERO);
+
+    let buffer = pull_buffer(&pipeline);
+    assert_fallback_buffer!(buffer, Some(gst::ClockTime::ZERO));
+
+    // Wait for EOS to arrive at appsink
+    push_eos(&pipeline);
+    push_fallback_eos(&pipeline);
+    wait_eos(&pipeline);
+
+    stop_pipeline(pipeline);
+}
+
+#[test]
+fn test_manual_switch_live() {
+    test_manual_switch(true);
+}
+
+#[test]
+fn test_manual_switch_not_live() {
+    test_manual_switch(false);
+}
+
+fn test_manual_switch(live: bool) {
+    let pipeline = setup_pipeline(Some(live), None, Some(false));
+    let switch = pipeline.by_name("switch").unwrap();
+    let mainsink = switch.static_pad("sink_0").unwrap();
+    let fallbacksink = switch.static_pad("sink_1").unwrap();
+
+    switch.set_property("active-pad", &mainsink);
+    push_buffer(&pipeline, gst::ClockTime::ZERO);
+    push_fallback_buffer(&pipeline, gst::ClockTime::ZERO);
+    set_time(&pipeline, gst::ClockTime::ZERO + LATENCY);
+    let buffer = pull_buffer(&pipeline);
+    assert_buffer!(buffer, Some(gst::ClockTime::ZERO));
+
+    switch.set_property("active-pad", &fallbacksink);
+    push_fallback_buffer(&pipeline, gst::ClockTime::SECOND);
+    push_buffer(&pipeline, gst::ClockTime::SECOND);
+    set_time(&pipeline, gst::ClockTime::SECOND + LATENCY);
+    let mut buffer = pull_buffer(&pipeline);
+    // FIXME: Sometimes we get the ZERO buffer from the fallback sink instead
+    if buffer.pts() == Some(gst::ClockTime::ZERO) {
+        buffer = pull_buffer(&pipeline);
+    }
+    assert_fallback_buffer!(buffer, Some(gst::ClockTime::SECOND));
+
+    switch.set_property("active-pad", &mainsink);
+    push_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
+    push_fallback_buffer(&pipeline, 2 * gst::ClockTime::SECOND);
+    set_time(&pipeline, 2 * gst::ClockTime::SECOND + LATENCY);
+    let buffer = pull_buffer(&pipeline);
+    assert_buffer!(buffer, Some(2 * gst::ClockTime::SECOND));
+
+    drop(mainsink);
+    drop(fallbacksink);
+    drop(switch);
+    // EOS on the fallback should not be required
+    push_eos(&pipeline);
+    wait_eos(&pipeline);
+
+    stop_pipeline(pipeline);
+}
+
 struct Pipeline {
     pipeline: gst::Pipeline,
     clock_join_handle: Option<std::thread::JoinHandle<()>>,
@@ -341,10 +497,14 @@ impl std::ops::Deref for Pipeline {
     }
 }
 
-fn setup_pipeline(with_live_fallback: Option<bool>) -> Pipeline {
+fn setup_pipeline(
+    with_live_fallback: Option<bool>,
+    immediate_fallback: Option<bool>,
+    auto_switch: Option<bool>,
+) -> Pipeline {
     init();
 
-    gst::debug!(TEST_CAT, "Setting up pipeline");
+    debug!(TEST_CAT, "Setting up pipeline");
 
     let clock = gst_check::TestClock::new();
     clock.set_time(gst::ClockTime::ZERO);
@@ -363,19 +523,25 @@ fn setup_pipeline(with_live_fallback: Option<bool>) -> Pipeline {
         .unwrap();
     src.set_property("is-live", true);
     src.set_property("format", gst::Format::Time);
-    src.set_property("min-latency", 10i64);
+    src.set_property("min-latency", LATENCY.nseconds() as i64);
     src.set_property(
         "caps",
-        &gst::Caps::builder("video/x-raw")
+        gst::Caps::builder("video/x-raw")
             .field("format", "ARGB")
-            .field("width", 320)
+            .field("width", 320i32)
-            .field("height", 240)
+            .field("height", 240i32)
             .field("framerate", gst::Fraction::new(1, 1))
             .build(),
     );
 
     let switch = gst::ElementFactory::make("fallbackswitch", Some("switch")).unwrap();
     switch.set_property("timeout", 3 * gst::ClockTime::SECOND);
+    if let Some(imm) = immediate_fallback {
+        switch.set_property("immediate-fallback", imm);
+    }
+    if let Some(auto_switch) = auto_switch {
+        switch.set_property("auto-switch", auto_switch);
+    }
 
     let sink = gst::ElementFactory::make("appsink", Some("sink"))
         .unwrap()
@@ -388,7 +554,7 @@ fn setup_pipeline(with_live_fallback: Option<bool>) -> Pipeline {
     pipeline
         .add_many(&[src.upcast_ref(), &switch, &queue, sink.upcast_ref()])
         .unwrap();
-    src.link_pads(Some("src"), &switch, Some("sink")).unwrap();
+    src.link_pads(Some("src"), &switch, Some("sink_0")).unwrap();
     switch.link_pads(Some("src"), &queue, Some("sink")).unwrap();
     queue.link_pads(Some("src"), &sink, Some("sink")).unwrap();
 
@@ -399,13 +565,13 @@ fn setup_pipeline(with_live_fallback: Option<bool>) -> Pipeline {
             .unwrap();
         fallback_src.set_property("is-live", live);
         fallback_src.set_property("format", gst::Format::Time);
-        fallback_src.set_property("min-latency", 10i64);
+        fallback_src.set_property("min-latency", LATENCY.nseconds() as i64);
         fallback_src.set_property(
             "caps",
-            &gst::Caps::builder("video/x-raw")
+            gst::Caps::builder("video/x-raw")
                 .field("format", "ARGB")
-                .field("width", 160)
+                .field("width", 160i32)
-                .field("height", 120)
+                .field("height", 120i32)
                 .field("framerate", gst::Fraction::new(1, 1))
                 .build(),
         );
@@ -413,7 +579,7 @@ fn setup_pipeline(with_live_fallback: Option<bool>) -> Pipeline {
         pipeline.add(&fallback_src).unwrap();
 
         fallback_src
-            .link_pads(Some("src"), &switch, Some("fallback_sink"))
+            .link_pads(Some("src"), &switch, Some("sink_1"))
             .unwrap();
     }
 
@@ -429,11 +595,16 @@ fn setup_pipeline(with_live_fallback: Option<bool>) -> Pipeline {
                     None
                 }
             }) {
-                gst::debug!(TEST_CAT, "Processing clock ID at {}", clock_id.time());
+                debug!(
+                    TEST_CAT,
+                    "Processing clock ID {} at {:?}",
+                    clock_id.time(),
+                    clock.time()
+                );
                 if let Some(clock_id) = clock.process_next_clock_id() {
-                    gst::debug!(TEST_CAT, "Processed clock ID at {}", clock_id.time());
+                    debug!(TEST_CAT, "Processed clock ID {}", clock_id.time());
                     if clock_id.time().is_zero() {
-                        gst::debug!(TEST_CAT, "Stopping clock thread");
+                        debug!(TEST_CAT, "Stopping clock thread");
                         return;
                     }
                 }
@@ -528,7 +699,7 @@ fn set_time(pipeline: &Pipeline, time: gst::ClockTime) {
         .downcast::<gst_check::TestClock>()
         .unwrap();
 
-    gst::debug!(TEST_CAT, "Setting time to {}", time);
+    debug!(TEST_CAT, "Setting time to {}", time);
     clock.set_time(gst::ClockTime::SECOND + time);
 }
 
@@ -543,7 +714,7 @@ fn wait_eos(pipeline: &Pipeline) {
         use std::{thread, time};
 
         if sink.is_eos() {
-            gst::debug!(TEST_CAT, "Waited for EOS");
+            debug!(TEST_CAT, "Waited for EOS");
             break;
         }
         thread::sleep(time::Duration::from_millis(10));