cea708mux: add support for force-live

This requires a bump of the gst-base dependency to 1.22 in order to
use the new set_force_live() API.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1876>

docs/plugins/gst_plugins_cache.json

@@ -6572,6 +6572,20 @@
                         "type": "GstAggregatorPad"
                     }
                 },
+                "properties": {
+                    "force-live": {
+                        "blurb": "Always operate in live mode and aggregate on timeout",
+                        "conditionally-available": false,
+                        "construct": false,
+                        "construct-only": true,
+                        "controllable": false,
+                        "default": "false",
+                        "mutable": "null",
+                        "readable": true,
+                        "type": "gboolean",
+                        "writable": true
+                    }
+                },
                 "rank": "none"
             },
             "cea708overlay": {

video/closedcaption/Cargo.toml

@@ -23,7 +23,7 @@ serde_json = { version = "1.0", features = ["raw_value"] }
 cea708-types = "0.3.2"
 cea608-types = "0.1.1"
 gst = { workspace = true, features = ["v1_20"]}
-gst-base = { workspace = true, features = ["v1_18"]}
+gst-base = { workspace = true, features = ["v1_22"]}
 gst-video = { workspace = true, features = ["v1_16"]}
 winnow = "0.6"
 smallvec = "1"

video/closedcaption/src/cea708mux/imp.rs

@@ -19,6 +19,8 @@ use gst_base::subclass::prelude::*;
 
 use std::sync::LazyLock;
 
+const DEFAULT_FORCE_LIVE: bool = false;
+
 #[derive(Default, Copy, Clone, PartialEq, Eq)]
 enum CeaFormat {
     S334_1a,
@@ -127,7 +129,8 @@ impl AggregatorImpl for Cea708Mux {
             .nseconds();
         let end_running_time = start_running_time + duration;
         let mut need_data = false;
-        let mut all_eos = true;
+        // In force-live mode, we never go EOS
+        let mut all_eos = !self.obj().is_force_live();
         gst::debug!(
             CAT,
             imp = self,
@@ -353,11 +356,33 @@ impl AggregatorImpl for Cea708Mux {
 
             self.finish_buffer(buf)
         } else {
-            self.obj().src_pad().push_event(
+            gst::trace!(
-                gst::event::Gap::builder(start_running_time)
+                CAT,
-                    .duration(duration)
+                imp = self,
-                    .build(),
+                "pushing {} gap at {}",
+                duration,
+                start_running_time
             );
+
+            #[cfg(feature = "v1_26")]
+            {
+                self.obj().push_src_event(
+                    gst::event::Gap::builder(start_running_time)
+                        .duration(duration)
+                        .build(),
+                );
+            }
+
+            #[cfg(not(feature = "v1_26"))]
+            {
+                self.obj().src_pad().push_event(
+                    gst::event::Gap::builder(start_running_time)
+                        .duration(duration)
+                        .build(),
+                );
+            }
+
+            gst::trace!(CAT, imp = self, "Pushed gap");
             Ok(gst::FlowSuccess::Ok)
         };
 
@@ -573,7 +598,37 @@ impl ElementImpl for Cea708Mux {
 
 impl GstObjectImpl for Cea708Mux {}
 
-impl ObjectImpl for Cea708Mux {}
+impl ObjectImpl for Cea708Mux {
+    fn properties() -> &'static [glib::ParamSpec] {
+        static PROPERTIES: LazyLock<Vec<glib::ParamSpec>> = LazyLock::new(|| {
+            vec![glib::ParamSpecBoolean::builder("force-live")
+                .nick("Force live")
+                .blurb("Always operate in live mode and aggregate on timeout")
+                .default_value(DEFAULT_FORCE_LIVE)
+                .construct_only()
+                .build()]
+        });
+
+        PROPERTIES.as_ref()
+    }
+
+    fn set_property(&self, _id: usize, value: &glib::Value, pspec: &glib::ParamSpec) {
+        match pspec.name() {
+            "force-live" => {
+                self.obj()
+                    .set_force_live(value.get().expect("type checked upstream"));
+            }
+            _ => unimplemented!(),
+        }
+    }
+
+    fn property(&self, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
+        match pspec.name() {
+            "force-live" => self.obj().is_force_live().to_value(),
+            _ => unimplemented!(),
+        }
+    }
+}
 
 #[glib::object_subclass]
 impl ObjectSubclass for Cea708Mux {