Playlist of subtitles enabled

src/main.rs

@@ -1,31 +1,40 @@
-use gst::prelude::*;
 use anyhow::Error;
+use gst::prelude::*;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
 
-fn main() -> Result<(), Error>  {
+fn main() -> Result<(), Error> {
     gst::init().unwrap();
     gstrsclosedcaption::plugin_register_static().expect("Failed to register closed caption plugin");
 
     // Create a video source with X buffers and burn the captions in the video feed using the captions tooling
     let pipeline = gst::parse_launch(
         "cccombiner name=ccc ! cea608overlay ! autovideosink \
-         videotestsrc num-buffers=1800 ! video/x-raw,width=1280,height=720,framerate=30/1 ! queue ! ccc.sink \
-         tttocea608 name=converter mode=1 ! queue ! ccc.caption",
+         videotestsrc num-buffers=3600 ! video/x-raw,width=1280,height=720,framerate=30/1 ! queue ! ccc.sink \
+         tttocea608 name=converter mode=0 ! queue ! ccc.caption",
     )?
         .downcast::<gst::Pipeline>()
         .expect("Expected a gst::Pipeline");
 
-    let text_src = gst::ElementFactory::make("appsrc", Some("text-src")).unwrap().downcast::<gst_app::AppSrc>().unwrap();
+    let text_src = gst::ElementFactory::make("appsrc", Some("text-src"))
+        .unwrap()
+        .downcast::<gst_app::AppSrc>()
+        .unwrap();
     text_src.set_is_live(true);
     text_src.set_format(gst::Format::Time);
     text_src.set_caps(Some(
         &gst::Caps::builder("text/x-raw")
-        .field("format", "utf8")
-        .build(),
+            .field("format", "utf8")
+            .build(),
     ));
-    pipeline.add(&text_src).expect("Failed to add elements to pipeline");
+    pipeline
+        .add(&text_src)
+        .expect("Failed to add elements to pipeline");
 
     let converter = pipeline.by_name("converter").unwrap();
-    text_src.link_pads(Some("src"), &converter, Some("sink")).expect("Failed to link elements");
+    text_src
+        .link_pads(Some("src"), &converter, Some("sink"))
+        .expect("Failed to link elements");
 
     println!("Starting pipeline..");
 
@@ -62,44 +71,63 @@ fn main_loop(pipeline: gst::Pipeline) -> Result<(), Error> {
             glib::Continue(true)
         }
     })
-        .expect("Failed to add bus watch");
+    .expect("Failed to add bus watch");
 
-    let pipeline_weak = pipeline.downgrade();
-    let timeout_id = glib::timeout_add_local(std::time::Duration::from_secs(20), move || {
-        let pipeline = match pipeline_weak.upgrade() {
-            Some(pipeline) => pipeline,
-            None => return glib::Continue(true),
-        };
-        push_buffer(&pipeline);
-        glib::Continue(true)
+    static RUNNING: AtomicBool = AtomicBool::new(true);
+    let subtitles_src_thread = std::thread::spawn({
+        let pipeline_weak = pipeline.downgrade();
+        move || {
+            std::thread::sleep(Duration::from_secs(2));
+
+            let subtitles = [
+                "Hello people!",
+                "This is just an example of \nsubtitle",
+                "We will learn more about \nGStreamer and buffers today",
+            ];
+            let mut sub_idx = (0..subtitles.len()).cycle().into_iter();
+
+            while RUNNING.load(Ordering::Relaxed) {
+                if let Some(pipeline) = pipeline_weak.upgrade() {
+                    push_buffer(&pipeline, subtitles[sub_idx.next().unwrap_or(0)]);
+                }
+                println!("pushed the buffer.. now waiting.");
+                std::thread::sleep(Duration::from_secs(5));
+            }
+            println!("Stopping subtitles source thread!");
+        }
     });
 
     main_loop.run();
 
     pipeline.set_state(gst::State::Null)?;
+    RUNNING.store(false, Ordering::Relaxed);
+    subtitles_src_thread.join().unwrap();
 
-    timeout_id.remove();
     bus.remove_watch().unwrap();
 
     Ok(())
 }
 
-fn push_buffer(pipeline: &gst::Pipeline) {
+fn push_buffer(pipeline: &gst::Pipeline, text: impl AsRef<str>) {
     let src = pipeline
         .by_name("text-src")
         .unwrap()
         .downcast::<gst_app::AppSrc>()
         .unwrap();
 
-    let timestamp = pipeline.query_position::<gst::ClockTime>().unwrap() + gst::ClockTime::from_seconds(1);
+    let present_caption_at =
+        pipeline.query_position::<gst::ClockTime>().unwrap() + gst::ClockTime::from_seconds(1);
 
-    println!("Trying to publish text buffer NOW! - {}", timestamp.display());
+    println!(
+        "Trying to publish text buffer NOW! - {} > {}",
+        present_caption_at.display(),
+        text.as_ref()
+    );
 
-    let text = format!("Such nice: {}!", timestamp.display());
-    let mut buffer = gst::Buffer::from_mut_slice(text.into_bytes());
+    let mut buffer = gst::Buffer::from_mut_slice(text.as_ref().to_string().into_bytes());
     {
         let buffer = buffer.get_mut().unwrap();
-        buffer.set_pts(timestamp);
+        buffer.set_pts(present_caption_at);
         buffer.set_duration(gst::ClockTime::from_seconds(1));
     }
     src.push_buffer(buffer).unwrap();