webvtt-playground/src/main.rs

use anyhow::Error;
use gst::prelude::*;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

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=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();
    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(),
    ));
    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");

    println!("Starting pipeline..");

    main_loop(pipeline)
}

fn main_loop(pipeline: gst::Pipeline) -> Result<(), Error> {
    let context = glib::MainContext::default();
    let main_loop = glib::MainLoop::new(Some(&context), false);

    pipeline.set_state(gst::State::Playing)?;

    let bus = pipeline.bus().unwrap();
    bus.add_watch({
        let main_loop = main_loop.clone();
        move |_, msg| {
            use gst::MessageView;

            let main_loop = &main_loop;
            match msg.view() {
                MessageView::Eos(..) => main_loop.quit(),
                MessageView::Error(err) => {
                    println!(
                        "Error from {:?}: {} ({:?})",
                        err.src().map(|s| s.path_string()),
                        err.error(),
                        err.debug()
                    );
                    main_loop.quit();
                }
                _ => (),
            };

            glib::Continue(true)
        }
    })
    .expect("Failed to add bus watch");

    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();

            while RUNNING.load(Ordering::Relaxed) {
                if let Some(pipeline) = pipeline_weak.upgrade() {
                    push_buffer(&pipeline, subtitles[sub_idx.next().unwrap_or(0)]);
                } else {
                    break;
                }
                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();

    bus.remove_watch().unwrap();

    Ok(())
}

fn push_buffer(pipeline: &gst::Pipeline, text: impl AsRef<str>) {
    let src = pipeline
        .by_name("text-src")
        .unwrap()
        .downcast::<gst_app::AppSrc>()
        .unwrap();

    let present_caption_at =
        pipeline.query_position::<gst::ClockTime>().unwrap() + gst::ClockTime::from_seconds(1);

    println!(
        "Trying to publish text buffer NOW! - {} > {}",
        present_caption_at.display(),
        text.as_ref()
    );

    let mut buffer = gst::Buffer::from_mut_slice(text.as_ref().to_string().into_bytes());
    {
        let buffer = buffer.get_mut().unwrap();
        buffer.set_pts(present_caption_at);
        buffer.set_duration(gst::ClockTime::from_seconds(1));
    }
    src.push_buffer(buffer).unwrap();
}