fallbackswitch: New plugin providing a fallbackswitch element

Allows having a live input stream and falling back to another input
stream after a configurable timeout without any buffers received on the
main input.

Cargo.toml

@@ -14,6 +14,7 @@ members = [
     "gst-plugin-cdg",
     "gst-plugin-rav1e",
     "gst-plugin-s3",
+    "gst-plugin-fallbackswitch",
 ]
 
 [profile.release]

gst-plugin-fallbackswitch/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "gst-plugin-fallbackswitch"
+version = "0.6.0"
+authors = ["Sebastian Dröge <sebastian@centricular.com>"]
+repository = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs"
+license = "LGPL-2.1+"
+edition = "2018"
+description = "Fallback Switcher Plugin"
+
+[dependencies]
+libc = "0.2"
+glib = { git = "https://github.com/gtk-rs/glib", features = ["subclassing"] }
+glib-sys = { git = "https://github.com/gtk-rs/sys" }
+gobject-sys = { git = "https://github.com/gtk-rs/sys" }
+gstreamer = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_14", "subclassing"] }
+gstreamer-base = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_16", "subclassing"], optional = true }
+gstreamer-sys = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs-sys" }
+gstreamer-audio = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_14"] }
+gstreamer-video = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_14"] }
+
+[lib]
+name = "gstfallbackswitch"
+crate-type = ["cdylib", "rlib"]
+path = "src/lib.rs"
+
+[build-dependencies]
+gst-plugin-version-helper = { path="../gst-plugin-version-helper" }
+cc = "1.0"
+pkg-config = "0.3"
+
+[features]
+default = []
+#v1_18 = ["gstreamer/v1_18", "gstreamer-base"]

gst-plugin-fallbackswitch/build.rs

@@ -0,0 +1,41 @@
+extern crate cc;
+extern crate gst_plugin_version_helper;
+extern crate pkg_config;
+
+fn main() {
+    gst_plugin_version_helper::get_info();
+
+    if cfg!(feature = "v1_18") {
+        return;
+    }
+
+    let gstreamer = pkg_config::probe_library("gstreamer-1.0").unwrap();
+    let includes = [gstreamer.include_paths];
+
+    let files = ["src/base/gstaggregator.c"];
+
+    let mut build = cc::Build::new();
+    build.include("src/base");
+
+    for f in files.iter() {
+        build.file(f);
+    }
+
+    for p in includes.iter().flat_map(|i| i) {
+        build.include(p);
+    }
+
+    build.define(
+        "PACKAGE_BUGREPORT",
+        "\"https://gitlab.freedesktop.org/gstreamer/gstreamer/issues/new\"",
+    );
+    build.define("GstAggregator", "GstAggregatorFallback");
+    build.define("GstAggregatorClass", "GstAggregatorFallbackClass");
+    build.define("GstAggregatorPrivate", "GstAggregatorFallbackPrivate");
+    build.define("GstAggregatorPad", "GstAggregatorFallbackPad");
+    build.define("GstAggregatorPadClass", "GstAggregatorFallbackPadClass");
+    build.define("GstAggregatorPadPrivate", "GstAggregatorFallbackPadPrivate");
+    build.define("GST_BASE_API", "G_GNUC_INTERNAL");
+
+    build.compile("libgstaggregator-c.a");
+}

gst-plugin-fallbackswitch/src/base/aggregator.rs

@@ -0,0 +1,95 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use super::gst_base_sys;
+use super::Aggregator;
+use glib::prelude::*;
+use glib::signal::{connect_raw, SignalHandlerId};
+use glib::translate::*;
+use glib::IsA;
+use glib::Value;
+use gst;
+use std::boxed::Box as Box_;
+use std::mem::transmute;
+
+pub trait AggregatorExtManual: 'static {
+    fn finish_buffer(&self, buffer: gst::Buffer) -> Result<gst::FlowSuccess, gst::FlowError>;
+    fn get_property_min_upstream_latency(&self) -> gst::ClockTime;
+
+    fn set_property_min_upstream_latency(&self, min_upstream_latency: gst::ClockTime);
+
+    fn connect_property_min_upstream_latency_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId;
+}
+
+impl<O: IsA<Aggregator>> AggregatorExtManual for O {
+    fn finish_buffer(&self, buffer: gst::Buffer) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let ret: gst::FlowReturn = unsafe {
+            from_glib(gst_base_sys::gst_aggregator_finish_buffer(
+                self.as_ref().to_glib_none().0,
+                buffer.into_ptr(),
+            ))
+        };
+        ret.into_result()
+    }
+
+    fn get_property_min_upstream_latency(&self) -> gst::ClockTime {
+        unsafe {
+            let mut value = Value::from_type(<gst::ClockTime as StaticType>::static_type());
+            gobject_sys::g_object_get_property(
+                self.to_glib_none().0 as *mut gobject_sys::GObject,
+                b"min-upstream-latency\0".as_ptr() as *const _,
+                value.to_glib_none_mut().0,
+            );
+            value
+                .get()
+                .expect("AggregatorExtManual::get_property_min_upstream_latency")
+                .unwrap()
+        }
+    }
+
+    fn set_property_min_upstream_latency(&self, min_upstream_latency: gst::ClockTime) {
+        unsafe {
+            gobject_sys::g_object_set_property(
+                self.to_glib_none().0 as *mut gobject_sys::GObject,
+                b"min-upstream-latency\0".as_ptr() as *const _,
+                Value::from(&min_upstream_latency).to_glib_none().0,
+            );
+        }
+    }
+
+    fn connect_property_min_upstream_latency_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId {
+        unsafe {
+            let f: Box_<F> = Box_::new(f);
+            connect_raw(
+                self.as_ptr() as *mut _,
+                b"notify::min-upstream-latency\0".as_ptr() as *const _,
+                Some(transmute(
+                    notify_min_upstream_latency_trampoline::<Self, F> as usize,
+                )),
+                Box_::into_raw(f),
+            )
+        }
+    }
+}
+
+unsafe extern "C" fn notify_min_upstream_latency_trampoline<P, F: Fn(&P) + Send + Sync + 'static>(
+    this: *mut gst_base_sys::GstAggregator,
+    _param_spec: glib_sys::gpointer,
+    f: glib_sys::gpointer,
+) where
+    P: IsA<Aggregator>,
+{
+    let f: &F = &*(f as *const F);
+    f(&Aggregator::from_glib_borrow(this).unsafe_cast())
+}

gst-plugin-fallbackswitch/src/base/aggregator_pad.rs

@@ -0,0 +1,28 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use super::gst_base_sys;
+use super::AggregatorPad;
+use glib::object::IsA;
+use glib::translate::*;
+use gst;
+use gst_sys;
+
+pub trait AggregatorPadExtManual: 'static {
+    fn get_segment(&self) -> gst::Segment;
+}
+
+impl<O: IsA<AggregatorPad>> AggregatorPadExtManual for O {
+    fn get_segment(&self) -> gst::Segment {
+        unsafe {
+            let ptr: &gst_base_sys::GstAggregatorPad = &*(self.as_ptr() as *const _);
+            super::utils::MutexGuard::lock(&ptr.parent.object.lock);
+            from_glib_none(&ptr.segment as *const gst_sys::GstSegment)
+        }
+    }
+}

gst-plugin-fallbackswitch/src/base/auto/aggregator.rs

@@ -0,0 +1,190 @@
+// This file was generated by gir (https://github.com/gtk-rs/gir)
+// from gir-files (https://github.com/gtk-rs/gir-files)
+// DO NOT EDIT
+
+use super::super::gst_base_sys;
+use glib::object::Cast;
+use glib::object::IsA;
+use glib::signal::connect_raw;
+use glib::signal::SignalHandlerId;
+use glib::translate::*;
+use glib::StaticType;
+use glib::Value;
+use glib_sys;
+use gobject_sys;
+use gst;
+use std::boxed::Box as Box_;
+use std::mem::transmute;
+
+glib_wrapper! {
+    pub struct Aggregator(Object<gst_base_sys::GstAggregator, gst_base_sys::GstAggregatorClass, AggregatorClass>) @extends gst::Element, gst::Object;
+
+    match fn {
+        get_type => || gst_base_sys::gst_aggregator_get_type(),
+    }
+}
+
+unsafe impl Send for Aggregator {}
+unsafe impl Sync for Aggregator {}
+
+pub const NONE_AGGREGATOR: Option<&Aggregator> = None;
+
+pub trait AggregatorExt: 'static {
+    fn get_buffer_pool(&self) -> Option<gst::BufferPool>;
+
+    fn get_latency(&self) -> gst::ClockTime;
+
+    fn set_latency(&self, min_latency: gst::ClockTime, max_latency: gst::ClockTime);
+
+    fn set_src_caps(&self, caps: &gst::Caps);
+
+    fn simple_get_next_time(&self) -> gst::ClockTime;
+
+    fn get_property_start_time(&self) -> u64;
+
+    fn set_property_start_time(&self, start_time: u64);
+
+    fn connect_property_latency_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId;
+
+    fn connect_property_start_time_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId;
+
+    fn negotiate(&self) -> bool;
+}
+
+impl<O: IsA<Aggregator>> AggregatorExt for O {
+    //fn get_allocator(&self, allocator: /*Ignored*/gst::Allocator, params: /*Ignored*/gst::AllocationParams) {
+    //    unsafe { TODO: call gst_base_sys:gst_aggregator_get_allocator() }
+    //}
+
+    fn get_buffer_pool(&self) -> Option<gst::BufferPool> {
+        unsafe {
+            from_glib_full(gst_base_sys::gst_aggregator_get_buffer_pool(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn get_latency(&self) -> gst::ClockTime {
+        unsafe {
+            from_glib(gst_base_sys::gst_aggregator_get_latency(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn set_latency(&self, min_latency: gst::ClockTime, max_latency: gst::ClockTime) {
+        unsafe {
+            gst_base_sys::gst_aggregator_set_latency(
+                self.as_ref().to_glib_none().0,
+                min_latency.to_glib(),
+                max_latency.to_glib(),
+            );
+        }
+    }
+
+    fn set_src_caps(&self, caps: &gst::Caps) {
+        unsafe {
+            gst_base_sys::gst_aggregator_set_src_caps(
+                self.as_ref().to_glib_none().0,
+                caps.to_glib_none().0,
+            );
+        }
+    }
+
+    fn simple_get_next_time(&self) -> gst::ClockTime {
+        unsafe {
+            from_glib(gst_base_sys::gst_aggregator_simple_get_next_time(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn get_property_start_time(&self) -> u64 {
+        unsafe {
+            let mut value = Value::from_type(<u64 as StaticType>::static_type());
+            gobject_sys::g_object_get_property(
+                self.to_glib_none().0 as *mut gobject_sys::GObject,
+                b"start-time\0".as_ptr() as *const _,
+                value.to_glib_none_mut().0,
+            );
+            value
+                .get()
+                .expect("Return Value for property `start-time` getter")
+                .unwrap()
+        }
+    }
+
+    fn set_property_start_time(&self, start_time: u64) {
+        unsafe {
+            gobject_sys::g_object_set_property(
+                self.to_glib_none().0 as *mut gobject_sys::GObject,
+                b"start-time\0".as_ptr() as *const _,
+                Value::from(&start_time).to_glib_none().0,
+            );
+        }
+    }
+
+    fn connect_property_latency_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId {
+        unsafe extern "C" fn notify_latency_trampoline<P, F: Fn(&P) + Send + Sync + 'static>(
+            this: *mut gst_base_sys::GstAggregator,
+            _param_spec: glib_sys::gpointer,
+            f: glib_sys::gpointer,
+        ) where
+            P: IsA<Aggregator>,
+        {
+            let f: &F = &*(f as *const F);
+            f(&Aggregator::from_glib_borrow(this).unsafe_cast())
+        }
+        unsafe {
+            let f: Box_<F> = Box_::new(f);
+            connect_raw(
+                self.as_ptr() as *mut _,
+                b"notify::latency\0".as_ptr() as *const _,
+                Some(transmute(notify_latency_trampoline::<Self, F> as usize)),
+                Box_::into_raw(f),
+            )
+        }
+    }
+
+    fn connect_property_start_time_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId {
+        unsafe extern "C" fn notify_start_time_trampoline<P, F: Fn(&P) + Send + Sync + 'static>(
+            this: *mut gst_base_sys::GstAggregator,
+            _param_spec: glib_sys::gpointer,
+            f: glib_sys::gpointer,
+        ) where
+            P: IsA<Aggregator>,
+        {
+            let f: &F = &*(f as *const F);
+            f(&Aggregator::from_glib_borrow(this).unsafe_cast())
+        }
+        unsafe {
+            let f: Box_<F> = Box_::new(f);
+            connect_raw(
+                self.as_ptr() as *mut _,
+                b"notify::start-time\0".as_ptr() as *const _,
+                Some(transmute(notify_start_time_trampoline::<Self, F> as usize)),
+                Box_::into_raw(f),
+            )
+        }
+    }
+
+    fn negotiate(&self) -> bool {
+        unsafe {
+            from_glib(gst_base_sys::gst_aggregator_negotiate(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+}

gst-plugin-fallbackswitch/src/base/auto/aggregator_pad.rs

@@ -0,0 +1,182 @@
+// This file was generated by gir (https://github.com/gtk-rs/gir)
+// from gir-files (https://github.com/gtk-rs/gir-files)
+// DO NOT EDIT
+
+use super::super::gst_base_sys;
+use glib::object::Cast;
+use glib::object::IsA;
+use glib::signal::connect_raw;
+use glib::signal::SignalHandlerId;
+use glib::translate::*;
+use glib::StaticType;
+use glib::Value;
+use glib_sys;
+use gobject_sys;
+use gst;
+use gst_sys;
+use std::boxed::Box as Box_;
+use std::mem::transmute;
+
+glib_wrapper! {
+    pub struct AggregatorPad(Object<gst_base_sys::GstAggregatorPad, gst_base_sys::GstAggregatorPadClass, AggregatorPadClass>) @extends gst::Pad, gst::Object;
+
+    match fn {
+        get_type => || gst_base_sys::gst_aggregator_pad_get_type(),
+    }
+}
+
+unsafe impl Send for AggregatorPad {}
+unsafe impl Sync for AggregatorPad {}
+
+pub const NONE_AGGREGATOR_PAD: Option<&AggregatorPad> = None;
+
+pub trait AggregatorPadExt: 'static {
+    fn drop_buffer(&self) -> bool;
+
+    fn has_buffer(&self) -> bool;
+
+    fn is_eos(&self) -> bool;
+
+    fn peek_buffer(&self) -> Option<gst::Buffer>;
+
+    fn pop_buffer(&self) -> Option<gst::Buffer>;
+
+    fn get_property_emit_signals(&self) -> bool;
+
+    fn set_property_emit_signals(&self, emit_signals: bool);
+
+    fn connect_buffer_consumed<F: Fn(&Self, &gst::Buffer) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId;
+
+    fn connect_property_emit_signals_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId;
+}
+
+impl<O: IsA<AggregatorPad>> AggregatorPadExt for O {
+    fn drop_buffer(&self) -> bool {
+        unsafe {
+            from_glib(gst_base_sys::gst_aggregator_pad_drop_buffer(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn has_buffer(&self) -> bool {
+        unsafe {
+            from_glib(gst_base_sys::gst_aggregator_pad_has_buffer(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn is_eos(&self) -> bool {
+        unsafe {
+            from_glib(gst_base_sys::gst_aggregator_pad_is_eos(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn peek_buffer(&self) -> Option<gst::Buffer> {
+        unsafe {
+            from_glib_full(gst_base_sys::gst_aggregator_pad_peek_buffer(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn pop_buffer(&self) -> Option<gst::Buffer> {
+        unsafe {
+            from_glib_full(gst_base_sys::gst_aggregator_pad_pop_buffer(
+                self.as_ref().to_glib_none().0,
+            ))
+        }
+    }
+
+    fn get_property_emit_signals(&self) -> bool {
+        unsafe {
+            let mut value = Value::from_type(<bool as StaticType>::static_type());
+            gobject_sys::g_object_get_property(
+                self.to_glib_none().0 as *mut gobject_sys::GObject,
+                b"emit-signals\0".as_ptr() as *const _,
+                value.to_glib_none_mut().0,
+            );
+            value
+                .get()
+                .expect("Return Value for property `emit-signals` getter")
+                .unwrap()
+        }
+    }
+
+    fn set_property_emit_signals(&self, emit_signals: bool) {
+        unsafe {
+            gobject_sys::g_object_set_property(
+                self.to_glib_none().0 as *mut gobject_sys::GObject,
+                b"emit-signals\0".as_ptr() as *const _,
+                Value::from(&emit_signals).to_glib_none().0,
+            );
+        }
+    }
+
+    fn connect_buffer_consumed<F: Fn(&Self, &gst::Buffer) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId {
+        unsafe extern "C" fn buffer_consumed_trampoline<
+            P,
+            F: Fn(&P, &gst::Buffer) + Send + Sync + 'static,
+        >(
+            this: *mut gst_base_sys::GstAggregatorPad,
+            object: *mut gst_sys::GstBuffer,
+            f: glib_sys::gpointer,
+        ) where
+            P: IsA<AggregatorPad>,
+        {
+            let f: &F = &*(f as *const F);
+            f(
+                &AggregatorPad::from_glib_borrow(this).unsafe_cast(),
+                &from_glib_borrow(object),
+            )
+        }
+        unsafe {
+            let f: Box_<F> = Box_::new(f);
+            connect_raw(
+                self.as_ptr() as *mut _,
+                b"buffer-consumed\0".as_ptr() as *const _,
+                Some(transmute(buffer_consumed_trampoline::<Self, F> as usize)),
+                Box_::into_raw(f),
+            )
+        }
+    }
+
+    fn connect_property_emit_signals_notify<F: Fn(&Self) + Send + Sync + 'static>(
+        &self,
+        f: F,
+    ) -> SignalHandlerId {
+        unsafe extern "C" fn notify_emit_signals_trampoline<P, F: Fn(&P) + Send + Sync + 'static>(
+            this: *mut gst_base_sys::GstAggregatorPad,
+            _param_spec: glib_sys::gpointer,
+            f: glib_sys::gpointer,
+        ) where
+            P: IsA<AggregatorPad>,
+        {
+            let f: &F = &*(f as *const F);
+            f(&AggregatorPad::from_glib_borrow(this).unsafe_cast())
+        }
+        unsafe {
+            let f: Box_<F> = Box_::new(f);
+            connect_raw(
+                self.as_ptr() as *mut _,
+                b"notify::emit-signals\0".as_ptr() as *const _,
+                Some(transmute(
+                    notify_emit_signals_trampoline::<Self, F> as usize,
+                )),
+                Box_::into_raw(f),
+            )
+        }
+    }
+}

gst-plugin-fallbackswitch/src/base/auto/mod.rs

@@ -0,0 +1,13 @@
+mod aggregator;
+pub use self::aggregator::AggregatorExt;
+pub use self::aggregator::{Aggregator, AggregatorClass, NONE_AGGREGATOR};
+
+mod aggregator_pad;
+pub use self::aggregator_pad::AggregatorPadExt;
+pub use self::aggregator_pad::{AggregatorPad, AggregatorPadClass, NONE_AGGREGATOR_PAD};
+
+#[doc(hidden)]
+pub mod traits {
+    pub use super::AggregatorExt;
+    pub use super::AggregatorPadExt;
+}

gst-plugin-fallbackswitch/src/base/gstaggregator.h

@@ -0,0 +1,394 @@
+/* GStreamer aggregator base class
+ * Copyright (C) 2014 Mathieu Duponchelle <mathieu.duponchelle@oencreed.com>
+ * Copyright (C) 2014 Thibault Saunier <tsaunier@gnome.org>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __GST_AGGREGATOR_H__
+#define __GST_AGGREGATOR_H__
+
+#include <gst/gst.h>
+#include <gst/base/base-prelude.h>
+
+G_BEGIN_DECLS
+
+/**************************
+ * GstAggregator Structs  *
+ *************************/
+
+typedef struct _GstAggregator GstAggregator;
+typedef struct _GstAggregatorPrivate GstAggregatorPrivate;
+typedef struct _GstAggregatorClass GstAggregatorClass;
+
+/************************
+ * GstAggregatorPad API *
+ ***********************/
+
+#define GST_TYPE_AGGREGATOR_PAD            (gst_aggregator_pad_get_type())
+#define GST_AGGREGATOR_PAD(obj)            (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_AGGREGATOR_PAD, GstAggregatorPad))
+#define GST_AGGREGATOR_PAD_CAST(obj)       ((GstAggregatorPad *)(obj))
+#define GST_AGGREGATOR_PAD_CLASS(klass)    (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_AGGREGATOR_PAD, GstAggregatorPadClass))
+#define GST_AGGREGATOR_PAD_GET_CLASS(obj)  (G_TYPE_INSTANCE_GET_CLASS ((obj),GST_TYPE_AGGREGATOR_PAD, GstAggregatorPadClass))
+#define GST_IS_AGGREGATOR_PAD(obj)         (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_AGGREGATOR_PAD))
+#define GST_IS_AGGREGATOR_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_AGGREGATOR_PAD))
+
+/****************************
+ * GstAggregatorPad Structs *
+ ***************************/
+
+typedef struct _GstAggregatorPad GstAggregatorPad;
+typedef struct _GstAggregatorPadClass GstAggregatorPadClass;
+typedef struct _GstAggregatorPadPrivate GstAggregatorPadPrivate;
+
+/**
+ * GstAggregatorPad:
+ * @segment: last segment received.
+ *
+ * The implementation the GstPad to use with #GstAggregator
+ *
+ * Since: 1.14
+ */
+struct _GstAggregatorPad
+{
+  GstPad                       parent;
+
+  /*< public >*/
+  /* Protected by the OBJECT_LOCK */
+  GstSegment segment;
+
+  /* < private > */
+  GstAggregatorPadPrivate   *  priv;
+
+  gpointer _gst_reserved[GST_PADDING];
+};
+
+/**
+ * GstAggregatorPadClass:
+ * @flush:       Optional
+ *               Called when the pad has received a flush stop, this is the place
+ *               to flush any information specific to the pad, it allows for individual
+ *               pads to be flushed while others might not be.
+ * @skip_buffer: Optional
+ *               Called before input buffers are queued in the pad, return %TRUE
+ *               if the buffer should be skipped.
+ *
+ * Since: 1.14
+ */
+struct _GstAggregatorPadClass
+{
+  GstPadClass   parent_class;
+
+  GstFlowReturn (*flush)       (GstAggregatorPad * aggpad, GstAggregator * aggregator);
+  gboolean      (*skip_buffer) (GstAggregatorPad * aggpad, GstAggregator * aggregator, GstBuffer * buffer);
+
+  /*< private >*/
+  gpointer      _gst_reserved[GST_PADDING_LARGE];
+};
+
+GST_BASE_API
+GType gst_aggregator_pad_get_type           (void);
+
+/****************************
+ * GstAggregatorPad methods *
+ ***************************/
+
+GST_BASE_API
+GstBuffer * gst_aggregator_pad_pop_buffer   (GstAggregatorPad *  pad);
+
+GST_BASE_API
+GstBuffer * gst_aggregator_pad_peek_buffer  (GstAggregatorPad *  pad);
+
+GST_BASE_API
+gboolean    gst_aggregator_pad_drop_buffer  (GstAggregatorPad *  pad);
+
+GST_BASE_API
+gboolean    gst_aggregator_pad_has_buffer   (GstAggregatorPad * pad);
+
+GST_BASE_API
+gboolean    gst_aggregator_pad_is_eos       (GstAggregatorPad *  pad);
+
+/*********************
+ * GstAggregator API *
+ ********************/
+
+#define GST_TYPE_AGGREGATOR            (gst_aggregator_get_type())
+#define GST_AGGREGATOR(obj)            (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_AGGREGATOR,GstAggregator))
+#define GST_AGGREGATOR_CAST(obj)       ((GstAggregator *)(obj))
+#define GST_AGGREGATOR_CLASS(klass)    (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_AGGREGATOR,GstAggregatorClass))
+#define GST_AGGREGATOR_GET_CLASS(obj)  (G_TYPE_INSTANCE_GET_CLASS ((obj),GST_TYPE_AGGREGATOR,GstAggregatorClass))
+#define GST_IS_AGGREGATOR(obj)         (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_AGGREGATOR))
+#define GST_IS_AGGREGATOR_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_AGGREGATOR))
+
+#define GST_AGGREGATOR_FLOW_NEED_DATA             GST_FLOW_CUSTOM_ERROR
+
+/**
+ * GstAggregator:
+ * @srcpad: the aggregator's source pad
+ *
+ * Aggregator base class object structure.
+ *
+ * Since: 1.14
+ */
+struct _GstAggregator
+{
+  GstElement               parent;
+
+  /*< public >*/
+  GstPad                *  srcpad;
+
+  /*< private >*/
+  GstAggregatorPrivate  *  priv;
+
+  gpointer                 _gst_reserved[GST_PADDING_LARGE];
+};
+
+/**
+ * GstAggregatorClass:
+ * @flush:          Optional.
+ *                  Called after a successful flushing seek, once all the flush
+ *                  stops have been received. Flush pad-specific data in
+ *                  #GstAggregatorPad->flush.
+ * @clip:           Optional.
+ *                  Called when a buffer is received on a sink pad, the task of
+ *                  clipping it and translating it to the current segment falls
+ *                  on the subclass. The function should use the segment of data
+ *                  and the negotiated media type on the pad to perform
+ *                  clipping of input buffer. This function takes ownership of
+ *                  buf and should output a buffer or return NULL in
+ *                  if the buffer should be dropped.
+ * @finish_buffer:  Optional.
+ *                  Called when a subclass calls gst_aggregator_finish_buffer()
+ *                  from their aggregate function to push out a buffer.
+ *                  Subclasses can override this to modify or decorate buffers
+ *                  before they get pushed out. This function takes ownership
+ *                  of the buffer passed. Subclasses that override this method
+ *                  should always chain up to the parent class virtual method.
+ * @sink_event:     Optional.
+ *                  Called when an event is received on a sink pad, the subclass
+ *                  should always chain up.
+ * @sink_query:     Optional.
+ *                  Called when a query is received on a sink pad, the subclass
+ *                  should always chain up.
+ * @src_event:      Optional.
+ *                  Called when an event is received on the src pad, the subclass
+ *                  should always chain up.
+ * @src_query:      Optional.
+ *                  Called when a query is received on the src pad, the subclass
+ *                  should always chain up.
+ * @src_activate:   Optional.
+ *                  Called when the src pad is activated, it will start/stop its
+ *                  pad task right after that call.
+ * @aggregate:      Mandatory.
+ *                  Called when buffers are queued on all sinkpads. Classes
+ *                  should iterate the GstElement->sinkpads and peek or steal
+ *                  buffers from the #GstAggregatorPads. If the subclass returns
+ *                  GST_FLOW_EOS, sending of the eos event will be taken care
+ *                  of. Once / if a buffer has been constructed from the
+ *                  aggregated buffers, the subclass should call _finish_buffer.
+ * @stop:           Optional.
+ *                  Called when the element goes from PAUSED to READY.
+ *                  The subclass should free all resources and reset its state.
+ * @start:          Optional.
+ *                  Called when the element goes from READY to PAUSED.
+ *                  The subclass should get ready to process
+ *                  aggregated buffers.
+ * @get_next_time:  Optional.
+ *                  Called when the element needs to know the running time of the next
+ *                  rendered buffer for live pipelines. This causes deadline
+ *                  based aggregation to occur. Defaults to returning
+ *                  GST_CLOCK_TIME_NONE causing the element to wait for buffers
+ *                  on all sink pads before aggregating.
+ * @create_new_pad: Optional.
+ *                  Called when a new pad needs to be created. Allows subclass that
+ *                  don't have a single sink pad template to provide a pad based
+ *                  on the provided information.
+ * @update_src_caps: Lets subclasses update the #GstCaps representing
+ *                   the src pad caps before usage.  The result should end up
+ *                   in @ret. Return %GST_AGGREGATOR_FLOW_NEED_DATA to indicate that the
+ *                   element needs more information (caps, a buffer, etc) to
+ *                   choose the correct caps. Should return ANY caps if the
+ *                   stream has not caps at all.
+ * @fixate_src_caps: Optional.
+ *                   Fixate and return the src pad caps provided.  The function takes
+ *                   ownership of @caps and returns a fixated version of
+ *                   @caps. @caps is not guaranteed to be writable.
+ * @negotiated_src_caps: Optional.
+ *                       Notifies subclasses what caps format has been negotiated
+ * @decide_allocation: Optional.
+ *                     Allows the subclass to influence the allocation choices.
+ *                     Setup the allocation parameters for allocating output
+ *                     buffers. The passed in query contains the result of the
+ *                     downstream allocation query.
+ * @propose_allocation: Optional.
+ *                     Allows the subclass to handle the allocation query from upstream.
+ * @negotiate: Optional.
+ *             Negotiate the caps with the peer (Since: 1.18).
+ * @sink_event_pre_queue: Optional.
+ *                        Called when an event is received on a sink pad before queueing up
+ *                        serialized events. The subclass should always chain up (Since: 1.18).
+ * @sink_query_pre_queue: Optional.
+ *                        Called when a query is received on a sink pad before queueing up
+ *                        serialized queries. The subclass should always chain up (Since: 1.18).
+ *
+ * The aggregator base class will handle in a thread-safe way all manners of
+ * concurrent flushes, seeks, pad additions and removals, leaving to the
+ * subclass the responsibility of clipping buffers, and aggregating buffers in
+ * the way the implementor sees fit.
+ *
+ * It will also take care of event ordering (stream-start, segment, eos).
+ *
+ * Basically, a simple implementation will override @aggregate, and call
+ * _finish_buffer from inside that function.
+ *
+ * Since: 1.14
+ */
+struct _GstAggregatorClass {
+  GstElementClass   parent_class;
+
+  GstFlowReturn     (*flush)          (GstAggregator    *  aggregator);
+
+  GstBuffer *       (*clip)           (GstAggregator    *  aggregator,
+                                       GstAggregatorPad *  aggregator_pad,
+                                       GstBuffer        *  buf);
+
+  GstFlowReturn     (*finish_buffer)  (GstAggregator    * aggregator,
+                                       GstBuffer        * buffer);
+
+  /* sinkpads virtual methods */
+  gboolean          (*sink_event)     (GstAggregator    *  aggregator,
+                                       GstAggregatorPad *  aggregator_pad,
+                                       GstEvent         *  event);
+
+  gboolean          (*sink_query)     (GstAggregator    *  aggregator,
+                                       GstAggregatorPad *  aggregator_pad,
+                                       GstQuery         *  query);
+
+  /* srcpad virtual methods */
+  gboolean          (*src_event)      (GstAggregator    *  aggregator,
+                                       GstEvent         *  event);
+
+  gboolean          (*src_query)      (GstAggregator    *  aggregator,
+                                       GstQuery         *  query);
+
+  gboolean          (*src_activate)   (GstAggregator    *  aggregator,
+                                       GstPadMode          mode,
+                                       gboolean            active);
+
+  GstFlowReturn     (*aggregate)      (GstAggregator    *  aggregator,
+                                       gboolean            timeout);
+
+  gboolean          (*stop)           (GstAggregator    *  aggregator);
+
+  gboolean          (*start)          (GstAggregator    *  aggregator);
+
+  GstClockTime      (*get_next_time)  (GstAggregator    *  aggregator);
+
+  GstAggregatorPad * (*create_new_pad) (GstAggregator  * self,
+                                        GstPadTemplate * templ,
+                                        const gchar    * req_name,
+                                        const GstCaps  * caps);
+
+  /**
+   * GstAggregatorClass::update_src_caps:
+   * @ret: (out) (allow-none):
+   */
+  GstFlowReturn     (*update_src_caps) (GstAggregator *  self,
+                                        GstCaps       *  caps,
+                                        GstCaps       ** ret);
+  GstCaps *         (*fixate_src_caps) (GstAggregator *  self,
+                                        GstCaps       *  caps);
+  gboolean          (*negotiated_src_caps) (GstAggregator *  self,
+                                            GstCaps      *  caps);
+  gboolean          (*decide_allocation) (GstAggregator * self,
+                                          GstQuery * query);
+  gboolean          (*propose_allocation) (GstAggregator * self,
+                                           GstAggregatorPad * pad,
+                                           GstQuery * decide_query,
+                                           GstQuery * query);
+
+  gboolean          (*negotiate) (GstAggregator * self);
+
+  gboolean          (*sink_event_pre_queue)     (GstAggregator    *  aggregator,
+                                                 GstAggregatorPad *  aggregator_pad,
+                                                 GstEvent         *  event);
+
+  gboolean          (*sink_query_pre_queue)     (GstAggregator    *  aggregator,
+                                                 GstAggregatorPad *  aggregator_pad,
+                                                 GstQuery         *  query);
+
+  /*< private >*/
+  gpointer          _gst_reserved[GST_PADDING_LARGE-3];
+};
+
+/************************************
+ * GstAggregator convenience macros *
+ ***********************************/
+
+/**
+ * GST_AGGREGATOR_SRC_PAD:
+ * @agg: a #GstAggregator
+ *
+ * Convenience macro to access the source pad of #GstAggregator
+ *
+ * Since: 1.6
+ */
+#define GST_AGGREGATOR_SRC_PAD(agg) (((GstAggregator *)(agg))->srcpad)
+
+/*************************
+ * GstAggregator methods *
+ ************************/
+
+GST_BASE_API
+GstFlowReturn  gst_aggregator_finish_buffer         (GstAggregator                *  aggregator,
+                                                     GstBuffer                    *  buffer);
+
+GST_BASE_API
+void           gst_aggregator_set_src_caps          (GstAggregator                *  self,
+                                                     GstCaps                      *  caps);
+
+GST_BASE_API
+gboolean        gst_aggregator_negotiate            (GstAggregator                * self);
+
+GST_BASE_API
+void           gst_aggregator_set_latency           (GstAggregator                *  self,
+                                                     GstClockTime                    min_latency,
+                                                     GstClockTime                    max_latency);
+
+GST_BASE_API
+GType gst_aggregator_get_type(void);
+
+GST_BASE_API
+GstClockTime  gst_aggregator_get_latency           (GstAggregator                 *  self);
+
+GST_BASE_API
+GstBufferPool * gst_aggregator_get_buffer_pool     (GstAggregator                 * self);
+
+GST_BASE_API
+void            gst_aggregator_get_allocator       (GstAggregator                 * self,
+                                                    GstAllocator                 ** allocator,
+                                                    GstAllocationParams           * params);
+
+GST_BASE_API
+GstClockTime    gst_aggregator_simple_get_next_time (GstAggregator                * self);
+
+
+G_END_DECLS
+
+G_DEFINE_AUTOPTR_CLEANUP_FUNC(GstAggregator, gst_object_unref)
+G_DEFINE_AUTOPTR_CLEANUP_FUNC(GstAggregatorPad, gst_object_unref)
+
+#endif /* __GST_AGGREGATOR_H__ */

gst-plugin-fallbackswitch/src/base/mod.rs

@@ -0,0 +1,27 @@
+#[allow(clippy::unreadable_literal)]
+#[allow(clippy::too_many_arguments)]
+#[allow(clippy::match_same_arms)]
+#[allow(clippy::type_complexity)]
+mod auto;
+pub use auto::*;
+
+mod utils;
+
+mod aggregator;
+mod aggregator_pad;
+
+pub mod prelude {
+    pub use glib::prelude::*;
+    pub use gst::prelude::*;
+
+    pub use super::aggregator::AggregatorExtManual;
+    pub use super::aggregator_pad::AggregatorPadExtManual;
+    pub use super::auto::traits::*;
+}
+
+pub mod subclass;
+
+mod sys;
+use sys as gst_base_sys;
+
+pub const AGGREGATOR_FLOW_NEED_DATA: gst::FlowError = gst::FlowError::CustomError;

gst-plugin-fallbackswitch/src/base/subclass/aggregator_pad.rs

@@ -0,0 +1,149 @@
+// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use super::super::gst_base_sys;
+use glib_sys;
+use gst_sys;
+
+use glib::translate::*;
+use gst;
+
+use glib::subclass::prelude::*;
+use gst::subclass::prelude::*;
+
+use super::super::Aggregator;
+use super::super::AggregatorPad;
+use super::super::AggregatorPadClass;
+
+pub trait AggregatorPadImpl: AggregatorPadImplExt + PadImpl + Send + Sync + 'static {
+    fn flush(
+        &self,
+        aggregator_pad: &AggregatorPad,
+        aggregator: &Aggregator,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        self.parent_flush(aggregator_pad, aggregator)
+    }
+
+    fn skip_buffer(
+        &self,
+        aggregator_pad: &AggregatorPad,
+        aggregator: &Aggregator,
+        buffer: &gst::Buffer,
+    ) -> bool {
+        self.parent_skip_buffer(aggregator_pad, aggregator, buffer)
+    }
+}
+
+pub trait AggregatorPadImplExt {
+    fn parent_flush(
+        &self,
+        aggregator_pad: &AggregatorPad,
+        aggregator: &Aggregator,
+    ) -> Result<gst::FlowSuccess, gst::FlowError>;
+
+    fn parent_skip_buffer(
+        &self,
+        aggregator_pad: &AggregatorPad,
+        aggregator: &Aggregator,
+        buffer: &gst::Buffer,
+    ) -> bool;
+}
+
+impl<T: AggregatorPadImpl + ObjectImpl> AggregatorPadImplExt for T {
+    fn parent_flush(
+        &self,
+        aggregator_pad: &AggregatorPad,
+        aggregator: &Aggregator,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        unsafe {
+            let data = self.get_type_data();
+            let parent_class =
+                data.as_ref().get_parent_class() as *mut gst_base_sys::GstAggregatorPadClass;
+            (*parent_class)
+                .flush
+                .map(|f| {
+                    from_glib(f(
+                        aggregator_pad.to_glib_none().0,
+                        aggregator.to_glib_none().0,
+                    ))
+                })
+                .unwrap_or(gst::FlowReturn::Ok)
+                .into_result()
+        }
+    }
+
+    fn parent_skip_buffer(
+        &self,
+        aggregator_pad: &AggregatorPad,
+        aggregator: &Aggregator,
+        buffer: &gst::Buffer,
+    ) -> bool {
+        unsafe {
+            let data = self.get_type_data();
+            let parent_class =
+                data.as_ref().get_parent_class() as *mut gst_base_sys::GstAggregatorPadClass;
+            (*parent_class)
+                .skip_buffer
+                .map(|f| {
+                    from_glib(f(
+                        aggregator_pad.to_glib_none().0,
+                        aggregator.to_glib_none().0,
+                        buffer.to_glib_none().0,
+                    ))
+                })
+                .unwrap_or(false)
+        }
+    }
+}
+unsafe impl<T: ObjectSubclass + AggregatorPadImpl> IsSubclassable<T> for AggregatorPadClass {
+    fn override_vfuncs(&mut self) {
+        <gst::PadClass as IsSubclassable<T>>::override_vfuncs(self);
+        unsafe {
+            let klass = &mut *(self as *mut Self as *mut gst_base_sys::GstAggregatorPadClass);
+            klass.flush = Some(aggregator_pad_flush::<T>);
+            klass.skip_buffer = Some(aggregator_pad_skip_buffer::<T>);
+        }
+    }
+}
+
+unsafe extern "C" fn aggregator_pad_flush<T: ObjectSubclass>(
+    ptr: *mut gst_base_sys::GstAggregatorPad,
+    aggregator: *mut gst_base_sys::GstAggregator,
+) -> gst_sys::GstFlowReturn
+where
+    T: AggregatorPadImpl,
+{
+    glib_floating_reference_guard!(ptr);
+    let instance = &*(ptr as *mut T::Instance);
+    let imp = instance.get_impl();
+    let wrap: AggregatorPad = from_glib_borrow(ptr);
+
+    let res: gst::FlowReturn = imp.flush(&wrap, &from_glib_borrow(aggregator)).into();
+    res.to_glib()
+}
+
+unsafe extern "C" fn aggregator_pad_skip_buffer<T: ObjectSubclass>(
+    ptr: *mut gst_base_sys::GstAggregatorPad,
+    aggregator: *mut gst_base_sys::GstAggregator,
+    buffer: *mut gst_sys::GstBuffer,
+) -> glib_sys::gboolean
+where
+    T: AggregatorPadImpl,
+{
+    glib_floating_reference_guard!(ptr);
+    let instance = &*(ptr as *mut T::Instance);
+    let imp = instance.get_impl();
+    let wrap: AggregatorPad = from_glib_borrow(ptr);
+
+    imp.skip_buffer(
+        &wrap,
+        &from_glib_borrow(aggregator),
+        &from_glib_borrow(buffer),
+    )
+    .to_glib()
+}

gst-plugin-fallbackswitch/src/base/subclass/mod.rs

@@ -0,0 +1,17 @@
+// Copyright (C) 2016-2018 Sebastian Dröge <sebastian@centricular.com>
+//               2016 Luis de Bethencourt <luisbg@osg.samsung.com>
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+#![allow(clippy::cast_ptr_alignment)]
+
+pub mod aggregator;
+pub mod aggregator_pad;
+
+pub mod prelude {
+    pub use super::aggregator::{AggregatorImpl, AggregatorImplExt};
+    pub use super::aggregator_pad::{AggregatorPadImpl, AggregatorPadImplExt};
+}

gst-plugin-fallbackswitch/src/base/sys.rs

@@ -0,0 +1,235 @@
+#![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)]
+#![allow(
+    clippy::approx_constant,
+    clippy::type_complexity,
+    clippy::unreadable_literal
+)]
+
+extern crate gstreamer_sys as gst;
+
+#[allow(unused_imports)]
+use libc::{
+    c_char, c_double, c_float, c_int, c_long, c_short, c_uchar, c_uint, c_ulong, c_ushort, c_void,
+    intptr_t, size_t, ssize_t, time_t, uintptr_t, FILE,
+};
+
+#[allow(unused_imports)]
+use glib_sys::{gboolean, gconstpointer, gpointer, GType};
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct GstAggregatorClass {
+    pub parent_class: gst::GstElementClass,
+    pub flush: Option<unsafe extern "C" fn(*mut GstAggregator) -> gst::GstFlowReturn>,
+    pub clip: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut GstAggregatorPad,
+            *mut gst::GstBuffer,
+        ) -> *mut gst::GstBuffer,
+    >,
+    pub finish_buffer:
+        Option<unsafe extern "C" fn(*mut GstAggregator, *mut gst::GstBuffer) -> gst::GstFlowReturn>,
+    pub sink_event: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut GstAggregatorPad,
+            *mut gst::GstEvent,
+        ) -> gboolean,
+    >,
+    pub sink_query: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut GstAggregatorPad,
+            *mut gst::GstQuery,
+        ) -> gboolean,
+    >,
+    pub src_event: Option<unsafe extern "C" fn(*mut GstAggregator, *mut gst::GstEvent) -> gboolean>,
+    pub src_query: Option<unsafe extern "C" fn(*mut GstAggregator, *mut gst::GstQuery) -> gboolean>,
+    pub src_activate:
+        Option<unsafe extern "C" fn(*mut GstAggregator, gst::GstPadMode, gboolean) -> gboolean>,
+    pub aggregate: Option<unsafe extern "C" fn(*mut GstAggregator, gboolean) -> gst::GstFlowReturn>,
+    pub stop: Option<unsafe extern "C" fn(*mut GstAggregator) -> gboolean>,
+    pub start: Option<unsafe extern "C" fn(*mut GstAggregator) -> gboolean>,
+    pub get_next_time: Option<unsafe extern "C" fn(*mut GstAggregator) -> gst::GstClockTime>,
+    pub create_new_pad: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut gst::GstPadTemplate,
+            *const c_char,
+            *const gst::GstCaps,
+        ) -> *mut GstAggregatorPad,
+    >,
+    pub update_src_caps: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut gst::GstCaps,
+            *mut *mut gst::GstCaps,
+        ) -> gst::GstFlowReturn,
+    >,
+    pub fixate_src_caps:
+        Option<unsafe extern "C" fn(*mut GstAggregator, *mut gst::GstCaps) -> *mut gst::GstCaps>,
+    pub negotiated_src_caps:
+        Option<unsafe extern "C" fn(*mut GstAggregator, *mut gst::GstCaps) -> gboolean>,
+    pub decide_allocation:
+        Option<unsafe extern "C" fn(*mut GstAggregator, *mut gst::GstQuery) -> gboolean>,
+    pub propose_allocation: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut GstAggregatorPad,
+            *mut gst::GstQuery,
+            *mut gst::GstQuery,
+        ) -> gboolean,
+    >,
+    pub negotiate: Option<unsafe extern "C" fn(*mut GstAggregator) -> gboolean>,
+    pub sink_event_pre_queue: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut GstAggregatorPad,
+            *mut gst::GstEvent,
+        ) -> gboolean,
+    >,
+    pub sink_query_pre_queue: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregator,
+            *mut GstAggregatorPad,
+            *mut gst::GstQuery,
+        ) -> gboolean,
+    >,
+    pub _gst_reserved: [gpointer; 17],
+}
+
+impl ::std::fmt::Debug for GstAggregatorClass {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        f.debug_struct(&format!("GstAggregatorClass @ {:?}", self as *const _))
+            .field("parent_class", &self.parent_class)
+            .field("flush", &self.flush)
+            .field("clip", &self.clip)
+            .field("finish_buffer", &self.finish_buffer)
+            .field("sink_event", &self.sink_event)
+            .field("sink_query", &self.sink_query)
+            .field("src_event", &self.src_event)
+            .field("src_query", &self.src_query)
+            .field("src_activate", &self.src_activate)
+            .field("aggregate", &self.aggregate)
+            .field("stop", &self.stop)
+            .field("start", &self.start)
+            .field("get_next_time", &self.get_next_time)
+            .field("create_new_pad", &self.create_new_pad)
+            .field("update_src_caps", &self.update_src_caps)
+            .field("fixate_src_caps", &self.fixate_src_caps)
+            .field("negotiated_src_caps", &self.negotiated_src_caps)
+            .field("decide_allocation", &self.decide_allocation)
+            .field("propose_allocation", &self.propose_allocation)
+            .finish()
+    }
+}
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct GstAggregatorPadClass {
+    pub parent_class: gst::GstPadClass,
+    pub flush: Option<
+        unsafe extern "C" fn(*mut GstAggregatorPad, *mut GstAggregator) -> gst::GstFlowReturn,
+    >,
+    pub skip_buffer: Option<
+        unsafe extern "C" fn(
+            *mut GstAggregatorPad,
+            *mut GstAggregator,
+            *mut gst::GstBuffer,
+        ) -> gboolean,
+    >,
+    pub _gst_reserved: [gpointer; 20],
+}
+
+impl ::std::fmt::Debug for GstAggregatorPadClass {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        f.debug_struct(&format!("GstAggregatorPadClass @ {:?}", self as *const _))
+            .field("parent_class", &self.parent_class)
+            .field("flush", &self.flush)
+            .field("skip_buffer", &self.skip_buffer)
+            .finish()
+    }
+}
+
+#[repr(C)]
+pub struct _GstAggregatorPadPrivate(c_void);
+
+pub type GstAggregatorPadPrivate = *mut _GstAggregatorPadPrivate;
+
+#[repr(C)]
+pub struct _GstAggregatorPrivate(c_void);
+
+pub type GstAggregatorPrivate = *mut _GstAggregatorPrivate;
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct GstAggregator {
+    pub parent: gst::GstElement,
+    pub srcpad: *mut gst::GstPad,
+    pub priv_: *mut GstAggregatorPrivate,
+    pub _gst_reserved: [gpointer; 20],
+}
+
+impl ::std::fmt::Debug for GstAggregator {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        f.debug_struct(&format!("GstAggregator @ {:?}", self as *const _))
+            .field("parent", &self.parent)
+            .field("srcpad", &self.srcpad)
+            .finish()
+    }
+}
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct GstAggregatorPad {
+    pub parent: gst::GstPad,
+    pub segment: gst::GstSegment,
+    pub priv_: *mut GstAggregatorPadPrivate,
+    pub _gst_reserved: [gpointer; 4],
+}
+
+impl ::std::fmt::Debug for GstAggregatorPad {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        f.debug_struct(&format!("GstAggregatorPad @ {:?}", self as *const _))
+            .field("parent", &self.parent)
+            .field("segment", &self.segment)
+            .finish()
+    }
+}
+
+extern "C" {
+    //=========================================================================
+    // GstAggregator
+    //=========================================================================
+    pub fn gst_aggregator_get_type() -> GType;
+    pub fn gst_aggregator_finish_buffer(
+        aggregator: *mut GstAggregator,
+        buffer: *mut gst::GstBuffer,
+    ) -> gst::GstFlowReturn;
+    pub fn gst_aggregator_negotiate(aggregator: *mut GstAggregator) -> gboolean;
+    pub fn gst_aggregator_get_allocator(
+        self_: *mut GstAggregator,
+        allocator: *mut *mut gst::GstAllocator,
+        params: *mut gst::GstAllocationParams,
+    );
+    pub fn gst_aggregator_get_buffer_pool(self_: *mut GstAggregator) -> *mut gst::GstBufferPool;
+    pub fn gst_aggregator_get_latency(self_: *mut GstAggregator) -> gst::GstClockTime;
+    pub fn gst_aggregator_set_latency(
+        self_: *mut GstAggregator,
+        min_latency: gst::GstClockTime,
+        max_latency: gst::GstClockTime,
+    );
+    pub fn gst_aggregator_set_src_caps(self_: *mut GstAggregator, caps: *mut gst::GstCaps);
+    pub fn gst_aggregator_simple_get_next_time(self_: *mut GstAggregator) -> gst::GstClockTime;
+
+    //=========================================================================
+    // GstAggregatorPad
+    //=========================================================================
+    pub fn gst_aggregator_pad_get_type() -> GType;
+    pub fn gst_aggregator_pad_drop_buffer(pad: *mut GstAggregatorPad) -> gboolean;
+    pub fn gst_aggregator_pad_has_buffer(pad: *mut GstAggregatorPad) -> gboolean;
+    pub fn gst_aggregator_pad_is_eos(pad: *mut GstAggregatorPad) -> gboolean;
+    pub fn gst_aggregator_pad_peek_buffer(pad: *mut GstAggregatorPad) -> *mut gst::GstBuffer;
+    pub fn gst_aggregator_pad_pop_buffer(pad: *mut GstAggregatorPad) -> *mut gst::GstBuffer;
+}

gst-plugin-fallbackswitch/src/base/utils.rs

@@ -0,0 +1,30 @@
+// Copyright (C) 2017 Sebastian Dröge <sebastian@centricular.com>
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use glib::translate::mut_override;
+use glib_sys;
+
+pub struct MutexGuard<'a>(&'a glib_sys::GMutex);
+
+impl<'a> MutexGuard<'a> {
+    #[allow(clippy::trivially_copy_pass_by_ref)]
+    pub fn lock(mutex: &'a glib_sys::GMutex) -> Self {
+        unsafe {
+            glib_sys::g_mutex_lock(mut_override(mutex));
+        }
+        MutexGuard(mutex)
+    }
+}
+
+impl<'a> Drop for MutexGuard<'a> {
+    fn drop(&mut self) {
+        unsafe {
+            glib_sys::g_mutex_unlock(mut_override(self.0));
+        }
+    }
+}

gst-plugin-fallbackswitch/src/fallbackswitch.rs

@@ -0,0 +1,759 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+#[cfg(not(feature = "v1_18"))]
+use super::base as gst_base;
+use glib;
+use glib::prelude::*;
+use glib::subclass;
+use glib::subclass::prelude::*;
+use gst;
+use gst::prelude::*;
+use gst::subclass::prelude::*;
+use gst_audio;
+#[cfg(feature = "v1_18")]
+use gst_base;
+use gst_base::prelude::*;
+use gst_base::subclass::prelude::*;
+use gst_video;
+
+use std::sync::{Mutex, RwLock};
+
+struct FallbackSwitch {
+    cat: gst::DebugCategory,
+    sinkpad: gst_base::AggregatorPad,
+    fallback_sinkpad: RwLock<Option<gst_base::AggregatorPad>>,
+    active_sinkpad: Mutex<gst::Pad>,
+    output_state: Mutex<OutputState>,
+    pad_states: RwLock<PadStates>,
+    settings: Mutex<Settings>,
+}
+
+#[derive(Debug)]
+struct OutputState {
+    last_sinkpad_time: gst::ClockTime,
+}
+
+#[derive(Debug, Default)]
+struct PadStates {
+    sinkpad: PadState,
+    fallback_sinkpad: Option<PadState>,
+}
+
+#[derive(Debug, Default)]
+struct PadState {
+    caps: Option<gst::Caps>,
+    audio_info: Option<gst_audio::AudioInfo>,
+    video_info: Option<gst_video::VideoInfo>,
+}
+
+const DEFAULT_TIMEOUT: u64 = 5 * gst::SECOND_VAL;
+
+#[derive(Debug, Clone)]
+struct Settings {
+    timeout: gst::ClockTime,
+}
+
+impl Default for OutputState {
+    fn default() -> Self {
+        OutputState {
+            last_sinkpad_time: gst::CLOCK_TIME_NONE,
+        }
+    }
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Settings {
+            timeout: DEFAULT_TIMEOUT.into(),
+        }
+    }
+}
+
+static PROPERTIES: [subclass::Property; 2] = [
+    subclass::Property("timeout", |name| {
+        glib::ParamSpec::uint64(
+            name,
+            "Timeout",
+            "Timeout in nanoseconds",
+            0,
+            std::u64::MAX,
+            DEFAULT_TIMEOUT,
+            glib::ParamFlags::READWRITE,
+        )
+    }),
+    subclass::Property("active-pad", |name| {
+        glib::ParamSpec::object(
+            name,
+            "Active Pad",
+            "Currently active pad",
+            gst::Pad::static_type(),
+            glib::ParamFlags::READABLE,
+        )
+    }),
+];
+
+impl FallbackSwitch {
+    fn handle_main_buffer(
+        &self,
+        agg: &gst_base::Aggregator,
+        state: &mut OutputState,
+        buffer: gst::Buffer,
+        fallback_sinkpad: Option<&gst_base::AggregatorPad>,
+    ) -> Result<(gst::Buffer, gst::Caps, bool), gst::FlowError> {
+        // If we got a buffer on the sinkpad just handle it
+        gst_debug!(self.cat, obj: agg, "Got buffer on sinkpad {:?}", buffer);
+
+        if buffer.get_pts().is_none() {
+            gst_error!(self.cat, obj: agg, "Only buffers with PTS supported");
+            return Err(gst::FlowError::Error);
+        }
+
+        let segment = match self.sinkpad.get_segment().downcast::<gst::ClockTime>() {
+            Ok(segment) => segment,
+            Err(_) => {
+                gst_error!(self.cat, obj: agg, "Only TIME segments supported");
+                return Err(gst::FlowError::Error);
+            }
+        };
+
+        let mut active_sinkpad = self.active_sinkpad.lock().unwrap();
+        let pad_change = &*active_sinkpad != self.sinkpad.upcast_ref::<gst::Pad>();
+        if pad_change {
+            gst_info!(self.cat, obj: agg, "Active pad changed to sinkpad");
+            *active_sinkpad = self.sinkpad.clone().upcast();
+        }
+        drop(active_sinkpad);
+
+        state.last_sinkpad_time = segment.to_running_time(buffer.get_pts());
+
+        // Drop all older buffers from the fallback sinkpad
+        if let Some(fallback_sinkpad) = fallback_sinkpad {
+            let fallback_segment = match self.sinkpad.get_segment().downcast::<gst::ClockTime>() {
+                Ok(segment) => segment,
+                Err(_) => {
+                    gst_error!(self.cat, obj: agg, "Only TIME segments supported");
+                    return Err(gst::FlowError::Error);
+                }
+            };
+
+            while let Some(fallback_buffer) = fallback_sinkpad.peek_buffer() {
+                let fallback_pts = fallback_buffer.get_pts();
+                if fallback_pts.is_none()
+                    || fallback_segment.to_running_time(fallback_pts) <= state.last_sinkpad_time
+                {
+                    gst_debug!(
+                        self.cat,
+                        obj: agg,
+                        "Dropping fallback buffer {:?}",
+                        fallback_buffer
+                    );
+                    fallback_sinkpad.drop_buffer();
+                } else {
+                    break;
+                }
+            }
+        }
+
+        let pad_states = self.pad_states.read().unwrap();
+        let active_caps = pad_states.sinkpad.caps.as_ref().unwrap().clone();
+        drop(pad_states);
+
+        Ok((buffer, active_caps, pad_change))
+    }
+
+    fn get_fallback_buffer(
+        &self,
+        agg: &gst_base::Aggregator,
+        state: &mut OutputState,
+        settings: &Settings,
+        fallback_sinkpad: &gst_base::AggregatorPad,
+    ) -> Result<(gst::Buffer, gst::Caps, bool), gst::FlowError> {
+        // If we have a fallback sinkpad and timeout, try to get a fallback buffer from here
+        // and drop all too old buffers in the process
+        loop {
+            let buffer = if let Some(buffer) = fallback_sinkpad.pop_buffer() {
+                buffer
+            } else {
+                gst_debug!(
+                    self.cat,
+                    obj: agg,
+                    "Got no buffer on sinkpad and fallback sinkpad"
+                );
+                return Err(gst_base::AGGREGATOR_FLOW_NEED_DATA);
+            };
+
+            gst_debug!(
+                self.cat,
+                obj: agg,
+                "Got buffer on fallback sinkpad {:?}",
+                buffer
+            );
+
+            if buffer.get_pts().is_none() {
+                gst_error!(self.cat, obj: agg, "Only buffers with PTS supported");
+                return Err(gst::FlowError::Error);
+            }
+
+            let fallback_segment = match fallback_sinkpad.get_segment().downcast::<gst::ClockTime>()
+            {
+                Ok(segment) => segment,
+                Err(_) => {
+                    gst_error!(self.cat, obj: agg, "Only TIME segments supported");
+                    return Err(gst::FlowError::Error);
+                }
+            };
+            let running_time = fallback_segment.to_running_time(buffer.get_pts());
+
+            // If we never had a real buffer, initialize with the running time of the fallback
+            // sinkpad so that we still output fallback buffers after the timeout
+            if state.last_sinkpad_time.is_none() {
+                state.last_sinkpad_time = running_time;
+            }
+
+            // Get the next one if this one is before the timeout
+            if state.last_sinkpad_time + settings.timeout > running_time {
+                gst_debug!(
+                    self.cat,
+                    obj: agg,
+                    "Timeout not reached yet: {} + {} > {}",
+                    state.last_sinkpad_time,
+                    settings.timeout,
+                    running_time
+                );
+
+                continue;
+            }
+
+            gst_debug!(
+                self.cat,
+                obj: agg,
+                "Timeout reached: {} + {} <= {}",
+                state.last_sinkpad_time,
+                settings.timeout,
+                running_time
+            );
+
+            let mut active_sinkpad = self.active_sinkpad.lock().unwrap();
+            let pad_change = &*active_sinkpad != fallback_sinkpad.upcast_ref::<gst::Pad>();
+            if pad_change {
+                gst_info!(self.cat, obj: agg, "Active pad changed to fallback sinkpad");
+                *active_sinkpad = fallback_sinkpad.clone().upcast();
+            }
+            drop(active_sinkpad);
+
+            let pad_states = self.pad_states.read().unwrap();
+            let active_caps = match pad_states.fallback_sinkpad {
+                None => {
+                    // This can happen if the pad is removed in the meantime,
+                    // not a problem really
+                    return Err(gst_base::AGGREGATOR_FLOW_NEED_DATA);
+                }
+                Some(ref fallback_sinkpad) => fallback_sinkpad.caps.as_ref().unwrap().clone(),
+            };
+            drop(pad_states);
+
+            break Ok((buffer, active_caps, pad_change));
+        }
+    }
+
+    fn get_next_buffer(
+        &self,
+        agg: &gst_base::Aggregator,
+        timeout: bool,
+    ) -> Result<(gst::Buffer, gst::Caps, bool), gst::FlowError> {
+        let settings = self.settings.lock().unwrap().clone();
+        let mut state = self.output_state.lock().unwrap();
+        let fallback_sinkpad = self.fallback_sinkpad.read().unwrap();
+
+        gst_debug!(self.cat, obj: agg, "Aggregate called: timeout {}", timeout);
+
+        if let Some(buffer) = self.sinkpad.pop_buffer() {
+            self.handle_main_buffer(agg, &mut *state, buffer, fallback_sinkpad.as_ref())
+        } else if self.sinkpad.is_eos() {
+            gst_log!(self.cat, obj: agg, "Sinkpad is EOS");
+            Err(gst::FlowError::Eos)
+        } else if let (false, Some(_)) = (timeout, &*fallback_sinkpad) {
+            gst_debug!(
+                self.cat,
+                obj: agg,
+                "Have fallback sinkpad but no timeout yet"
+            );
+
+            Err(gst_base::AGGREGATOR_FLOW_NEED_DATA)
+        } else if let (true, Some(fallback_sinkpad)) = (timeout, &*fallback_sinkpad) {
+            self.get_fallback_buffer(agg, &mut *state, &settings, fallback_sinkpad)
+        } else {
+            // Otherwise there's not much we can do at this point
+            gst_debug!(
+                self.cat,
+                obj: agg,
+                "Got no buffer on sinkpad and have no fallback sinkpad"
+            );
+            Err(gst_base::AGGREGATOR_FLOW_NEED_DATA)
+        }
+    }
+}
+
+impl ObjectSubclass for FallbackSwitch {
+    const NAME: &'static str = "FallbackSwitch";
+    type ParentType = gst_base::Aggregator;
+    type Instance = gst::subclass::ElementInstanceStruct<Self>;
+    type Class = subclass::simple::ClassStruct<Self>;
+
+    glib_object_subclass!();
+
+    fn new_with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+        let templ = klass.get_pad_template("sink").unwrap();
+        let sinkpad: gst_base::AggregatorPad = glib::Object::new(
+            gst_base::AggregatorPad::static_type(),
+            &[
+                ("name", &"sink"),
+                ("direction", &gst::PadDirection::Sink),
+                ("template", &templ),
+            ],
+        )
+        .unwrap()
+        .downcast()
+        .unwrap();
+
+        Self {
+            cat: gst::DebugCategory::new(
+                "fallbackswitch",
+                gst::DebugColorFlags::empty(),
+                Some("Fallback switch Element"),
+            ),
+            sinkpad: sinkpad.clone(),
+            fallback_sinkpad: RwLock::new(None),
+            active_sinkpad: Mutex::new(sinkpad.upcast()),
+            output_state: Mutex::new(OutputState::default()),
+            pad_states: RwLock::new(PadStates::default()),
+            settings: Mutex::new(Settings::default()),
+        }
+    }
+
+    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+        klass.set_metadata(
+            "Fallback Switch",
+            "Generic",
+            "Allows switching to a fallback input after a given timeout",
+            "Sebastian Dröge <sebastian@centricular.com>",
+        );
+
+        let caps = {
+            let mut caps = gst::Caps::new_empty();
+            {
+                let caps = caps.get_mut().unwrap();
+                caps.append_structure(gst::Structure::new_empty("video/x-raw"));
+                caps.append_structure(gst::Structure::new_empty("audio/x-raw"));
+            }
+            caps
+        };
+        let src_pad_template = gst::PadTemplate::new_with_gtype(
+            "src",
+            gst::PadDirection::Src,
+            gst::PadPresence::Always,
+            &caps,
+            gst_base::AggregatorPad::static_type(),
+        )
+        .unwrap();
+        klass.add_pad_template(src_pad_template);
+
+        let sink_pad_template = gst::PadTemplate::new_with_gtype(
+            "sink",
+            gst::PadDirection::Sink,
+            gst::PadPresence::Always,
+            &caps,
+            gst_base::AggregatorPad::static_type(),
+        )
+        .unwrap();
+        klass.add_pad_template(sink_pad_template);
+
+        let fallbacksink_pad_template = gst::PadTemplate::new_with_gtype(
+            "fallback_sink",
+            gst::PadDirection::Sink,
+            gst::PadPresence::Request,
+            &caps,
+            gst_base::AggregatorPad::static_type(),
+        )
+        .unwrap();
+        klass.add_pad_template(fallbacksink_pad_template);
+
+        klass.install_properties(&PROPERTIES);
+    }
+}
+
+impl ObjectImpl for FallbackSwitch {
+    glib_object_impl!();
+
+    fn constructed(&self, obj: &glib::Object) {
+        self.parent_constructed(obj);
+
+        let agg = obj.downcast_ref::<gst_base::Aggregator>().unwrap();
+        agg.add_pad(&self.sinkpad).unwrap();
+    }
+
+    fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) {
+        let prop = &PROPERTIES[id];
+        let agg = obj.downcast_ref::<gst_base::Aggregator>().unwrap();
+
+        match *prop {
+            subclass::Property("timeout", ..) => {
+                let mut settings = self.settings.lock().unwrap();
+                let timeout = value.get_some().expect("type checked upstream");
+                gst_info!(
+                    self.cat,
+                    obj: agg,
+                    "Changing timeout from {} to {}",
+                    settings.timeout,
+                    timeout
+                );
+                settings.timeout = timeout;
+                drop(settings);
+            }
+            _ => unimplemented!(),
+        }
+    }
+
+    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+        let prop = &PROPERTIES[id];
+
+        match *prop {
+            subclass::Property("timeout", ..) => {
+                let settings = self.settings.lock().unwrap();
+                Ok(settings.timeout.to_value())
+            }
+            subclass::Property("active-pad", ..) => {
+                let active_pad = self.active_sinkpad.lock().unwrap().clone();
+                Ok(active_pad.to_value())
+            }
+            _ => unimplemented!(),
+        }
+    }
+}
+
+impl ElementImpl for FallbackSwitch {
+    fn request_new_pad(
+        &self,
+        element: &gst::Element,
+        templ: &gst::PadTemplate,
+        name: Option<String>,
+        _caps: Option<&gst::Caps>,
+    ) -> Option<gst::Pad> {
+        let agg = element.downcast_ref::<gst_base::Aggregator>().unwrap();
+        let fallback_sink_templ = agg.get_pad_template("fallback_sink").unwrap();
+        if templ != &fallback_sink_templ
+            || (name.is_some() && name.as_ref().map(String::as_str) != Some("fallback_sink"))
+        {
+            gst_error!(self.cat, obj: agg, "Wrong pad template or name");
+            return None;
+        }
+
+        let mut fallback_sinkpad = self.fallback_sinkpad.write().unwrap();
+        if fallback_sinkpad.is_some() {
+            gst_error!(self.cat, obj: agg, "Already have a fallback sinkpad");
+            return None;
+        }
+
+        let sinkpad: gst_base::AggregatorPad = glib::Object::new(
+            gst_base::AggregatorPad::static_type(),
+            &[
+                ("name", &"fallback_sink"),
+                ("direction", &gst::PadDirection::Sink),
+                ("template", templ),
+            ],
+        )
+        .unwrap()
+        .downcast()
+        .unwrap();
+        *fallback_sinkpad = Some(sinkpad.clone());
+        drop(fallback_sinkpad);
+
+        agg.add_pad(&sinkpad).unwrap();
+
+        Some(sinkpad.upcast())
+    }
+
+    fn release_pad(&self, element: &gst::Element, pad: &gst::Pad) {
+        let agg = element.downcast_ref::<gst_base::Aggregator>().unwrap();
+        let mut fallback_sinkpad = self.fallback_sinkpad.write().unwrap();
+        let mut pad_states = self.pad_states.write().unwrap();
+
+        if fallback_sinkpad.as_ref().map(|p| p.upcast_ref()) == Some(pad) {
+            *fallback_sinkpad = None;
+            pad_states.fallback_sinkpad = None;
+            drop(pad_states);
+            drop(fallback_sinkpad);
+            agg.remove_pad(pad).unwrap();
+            gst_debug!(self.cat, obj: agg, "Removed fallback sinkpad {:?}", pad);
+        }
+    }
+}
+
+impl AggregatorImpl for FallbackSwitch {
+    fn start(&self, _agg: &gst_base::Aggregator) -> Result<(), gst::ErrorMessage> {
+        *self.active_sinkpad.lock().unwrap() = self.sinkpad.clone().upcast();
+        *self.output_state.lock().unwrap() = OutputState::default();
+        *self.pad_states.write().unwrap() = PadStates::default();
+
+        Ok(())
+    }
+
+    fn sink_event_pre_queue(
+        &self,
+        agg: &gst_base::Aggregator,
+        agg_pad: &gst_base::AggregatorPad,
+        event: gst::Event,
+    ) -> bool {
+        use gst::EventView;
+
+        match event.view() {
+            EventView::Gap(_) => {
+                gst_debug!(self.cat, obj: agg_pad, "Dropping gap event");
+                true
+            }
+            _ => self.parent_sink_event_pre_queue(agg, agg_pad, event),
+        }
+    }
+
+    fn sink_event(
+        &self,
+        agg: &gst_base::Aggregator,
+        agg_pad: &gst_base::AggregatorPad,
+        event: gst::Event,
+    ) -> bool {
+        use gst::EventView;
+
+        match event.view() {
+            EventView::Caps(caps) => {
+                let caps = caps.get_caps_owned();
+                gst_debug!(self.cat, obj: agg_pad, "Received caps {}", caps);
+
+                let audio_info;
+                let video_info;
+                if caps.get_structure(0).unwrap().get_name() == "audio/x-raw" {
+                    audio_info = gst_audio::AudioInfo::from_caps(&caps);
+                    video_info = None;
+                } else if caps.get_structure(0).unwrap().get_name() == "video/x-raw" {
+                    audio_info = None;
+                    video_info = gst_video::VideoInfo::from_caps(&caps);
+                } else {
+                    unreachable!()
+                }
+
+                let new_pad_state = PadState {
+                    caps: Some(caps),
+                    audio_info,
+                    video_info,
+                };
+
+                let mut pad_states = self.pad_states.write().unwrap();
+                if agg_pad == &self.sinkpad {
+                    pad_states.sinkpad = new_pad_state;
+                } else if Some(agg_pad) == self.fallback_sinkpad.read().unwrap().as_ref() {
+                    pad_states.fallback_sinkpad = Some(new_pad_state);
+                }
+                drop(pad_states);
+
+                self.parent_sink_event(agg, agg_pad, event)
+            }
+            _ => self.parent_sink_event(agg, agg_pad, event),
+        }
+    }
+
+    fn get_next_time(&self, agg: &gst_base::Aggregator) -> gst::ClockTime {
+        // If we have a buffer on the sinkpad then the timeout is always going to be immediately,
+        // i.e. 0. We want to output that buffer immediately, no matter what.
+        //
+        // Otherwise if we have a fallback sinkpad and it has a buffer, then the timeout is going
+        // to be its running time. We will then either output the buffer or drop it, depending on
+        // its distance from the last sinkpad time
+        if let Some(_) = self.sinkpad.peek_buffer() {
+            gst_debug!(
+                self.cat,
+                obj: agg,
+                "Have buffer on sinkpad, immediate timeout"
+            );
+            0.into()
+        } else if self.sinkpad.is_eos() {
+            gst_debug!(self.cat, obj: agg, "Sinkpad is EOS, immediate timeout");
+            0.into()
+        } else if let Some((buffer, fallback_sinkpad)) = self
+            .fallback_sinkpad
+            .read()
+            .unwrap()
+            .as_ref()
+            .and_then(|p| p.peek_buffer().map(|buffer| (buffer, p)))
+        {
+            if buffer.get_pts().is_none() {
+                gst_error!(self.cat, obj: agg, "Only buffers with PTS supported");
+                // Trigger aggregate immediately to error out immediately
+                return 0.into();
+            }
+
+            let segment = match fallback_sinkpad.get_segment().downcast::<gst::ClockTime>() {
+                Ok(segment) => segment,
+                Err(_) => {
+                    gst_error!(self.cat, obj: agg, "Only TIME segments supported");
+                    // Trigger aggregate immediately to error out immediately
+                    return 0.into();
+                }
+            };
+
+            let running_time = segment.to_running_time(buffer.get_pts());
+            gst_debug!(
+                self.cat,
+                obj: agg,
+                "Have buffer on fallback sinkpad, timeout at {}",
+                running_time
+            );
+            running_time
+        } else {
+            gst_debug!(self.cat, obj: agg, "Have no buffer at all yet");
+            gst::CLOCK_TIME_NONE
+        }
+    }
+
+    // Clip the raw audio/video buffers we have to the segment boundaries to ensure that
+    // calculating the running times later works correctly
+    fn clip(
+        &self,
+        agg: &gst_base::Aggregator,
+        agg_pad: &gst_base::AggregatorPad,
+        mut buffer: gst::Buffer,
+    ) -> Option<gst::Buffer> {
+        let segment = match agg_pad.get_segment().downcast::<gst::ClockTime>() {
+            Ok(segment) => segment,
+            Err(_) => {
+                gst_error!(self.cat, obj: agg, "Only TIME segments supported");
+                return Some(buffer);
+            }
+        };
+
+        let pts = buffer.get_pts();
+        if pts.is_none() {
+            gst_error!(self.cat, obj: agg, "Only buffers with PTS supported");
+            return Some(buffer);
+        }
+
+        let pad_states = self.pad_states.read().unwrap();
+        let pad_state = if agg_pad == &self.sinkpad {
+            &pad_states.sinkpad
+        } else if Some(agg_pad) == self.fallback_sinkpad.read().unwrap().as_ref() {
+            if let Some(ref pad_state) = pad_states.fallback_sinkpad {
+                pad_state
+            } else {
+                return Some(buffer);
+            }
+        } else {
+            unreachable!()
+        };
+
+        let duration = if buffer.get_duration().is_some() {
+            buffer.get_duration()
+        } else if let Some(ref audio_info) = pad_state.audio_info {
+            gst::SECOND
+                .mul_div_floor(
+                    buffer.get_size() as u64,
+                    audio_info.rate() as u64 * audio_info.bpf() as u64,
+                )
+                .unwrap()
+        } else if let Some(ref video_info) = pad_state.video_info {
+            gst::SECOND
+                .mul_div_floor(
+                    *video_info.fps().denom() as u64,
+                    *video_info.fps().numer() as u64,
+                )
+                .unwrap()
+        } else {
+            unreachable!()
+        };
+
+        gst_debug!(
+            self.cat,
+            obj: agg_pad,
+            "Clipping buffer {:?} with PTS {} and duration {}",
+            buffer,
+            pts,
+            duration
+        );
+        if let Some(ref audio_info) = pad_state.audio_info {
+            gst_audio::audio_buffer_clip(
+                buffer,
+                segment.upcast_ref(),
+                audio_info.rate(),
+                audio_info.bpf(),
+            )
+        } else if let Some(_) = pad_state.video_info {
+            segment.clip(pts, pts + duration).map(|(start, stop)| {
+                {
+                    let buffer = buffer.make_mut();
+                    buffer.set_pts(start);
+                    buffer.set_dts(start);
+                    buffer.set_duration(stop - start);
+                }
+
+                buffer
+            })
+        } else {
+            unreachable!();
+        }
+    }
+
+    fn aggregate(
+        &self,
+        agg: &gst_base::Aggregator,
+        timeout: bool,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        gst_debug!(self.cat, obj: agg, "Aggregate called: timeout {}", timeout);
+
+        let (mut buffer, active_caps, pad_change) = self.get_next_buffer(agg, timeout)?;
+
+        let current_src_caps = agg.get_static_pad("src").unwrap().get_current_caps();
+        if Some(&active_caps) != current_src_caps.as_ref() {
+            gst_info!(
+                self.cat,
+                obj: agg,
+                "Caps change from {:?} to {:?}",
+                current_src_caps,
+                active_caps
+            );
+            agg.set_src_caps(&active_caps);
+        }
+
+        if pad_change {
+            agg.notify("active-pad");
+            buffer.make_mut().set_flags(gst::BufferFlags::DISCONT);
+        }
+
+        gst_debug!(self.cat, obj: agg, "Finishing buffer {:?}", buffer);
+        agg.finish_buffer(buffer)
+    }
+
+    fn negotiate(&self, _agg: &gst_base::Aggregator) -> bool {
+        true
+    }
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "fallbackswitch",
+        gst::Rank::None,
+        FallbackSwitch::get_type(),
+    )
+}

gst-plugin-fallbackswitch/src/lib.rs

@@ -0,0 +1,59 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+#[macro_use]
+extern crate glib;
+#[macro_use]
+extern crate gstreamer as gst;
+
+extern crate gstreamer_audio as gst_audio;
+extern crate gstreamer_video as gst_video;
+
+#[cfg(not(feature = "v1_18"))]
+extern crate glib_sys;
+#[cfg(not(feature = "v1_18"))]
+extern crate gobject_sys;
+#[cfg(feature = "v1_18")]
+extern crate gstreamer_base as gst_base;
+#[cfg(not(feature = "v1_18"))]
+extern crate gstreamer_sys as gst_sys;
+#[cfg(not(feature = "v1_18"))]
+#[allow(dead_code)]
+mod base;
+#[cfg(not(feature = "v1_18"))]
+mod gst_base {
+    pub use super::base::*;
+}
+
+mod fallbackswitch;
+
+fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    fallbackswitch::register(plugin)?;
+    Ok(())
+}
+
+gst_plugin_define!(
+    fallbackswitch,
+    env!("CARGO_PKG_DESCRIPTION"),
+    plugin_init,
+    concat!(env!("CARGO_PKG_VERSION"), "-", env!("COMMIT_ID")),
+    "MIT/X11",
+    env!("CARGO_PKG_NAME"),
+    env!("CARGO_PKG_NAME"),
+    env!("CARGO_PKG_REPOSITORY"),
+    env!("BUILD_REL_DATE")
+);