jello/gst/src/lib.rs

pub mod errors;
pub mod playbin3;
use errors::*;
use gstreamer::prelude::*;
use std::sync::Arc;

static GST: std::sync::LazyLock<std::sync::Arc<Gst>> = std::sync::LazyLock::new(|| {
    gstreamer::init().expect("Failed to initialize GStreamer");
    std::sync::Arc::new(Gst {
        __private: core::marker::PhantomData,
    })
});

/// This should be a global singleton
pub struct Gst {
    __private: core::marker::PhantomData<()>,
}

impl Gst {
    pub fn new() -> Arc<Self> {
        Arc::clone(&GST)
    }

    pub fn pipeline_from_str(&self, s: &str) -> Result<Pipeline> {
        let pipeline = gstreamer::parse::launch(s).change_context(Error)?;
        let pipeline = pipeline.downcast::<gstreamer::Pipeline>();
        let pipeline = match pipeline {
            Err(_e) => return Err(Error).attach("Failed to downcast to Pipeline"),
            Ok(p) => p,
        };
        Ok(Pipeline { pipeline })
    }
}

pub struct Bin {
    bin: gstreamer::Bin,
}
pub struct Sample {
    sample: gstreamer::Sample,
}

pub struct Pipeline {
    pipeline: gstreamer::Pipeline,
}

impl core::fmt::Debug for Pipeline {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Pipeline")
            .field("pipeline", &self.pipeline)
            // .field("state", &self.pipeline.state(gstreamer::ClockTime::NONE))
            .finish()
    }
}

impl Drop for Pipeline {
    fn drop(&mut self) {
        let _ = self.pipeline.set_state(gstreamer::State::Null);
    }
}

impl Pipeline {
    pub fn bus(&self) -> Result<Bus> {
        let bus = self
            .pipeline
            .bus()
            .ok_or(Error)
            .attach("Failed to get bus from pipeline")?;
        Ok(Bus { bus })
    }

    pub fn set_state(&self, state: gstreamer::State) -> Result<gstreamer::StateChangeSuccess> {
        let result = self
            .pipeline
            .set_state(state)
            .change_context(Error)
            .attach("Failed to set pipeline state")?;
        Ok(result)
    }
}

pub struct Bus {
    bus: gstreamer::Bus,
}

impl Bus {
    pub fn iter_timed(
        &self,
        timeout: impl Into<Option<core::time::Duration>>,
    ) -> gstreamer::bus::Iter<'_> {
        let clocktime = match timeout.into() {
            Some(dur) => gstreamer::ClockTime::try_from(dur).ok(),
            None => gstreamer::ClockTime::NONE,
        };
        self.bus.iter_timed(clocktime)
    }
}

/// Pads are link points between elements
pub struct Pad {
    pad: gstreamer::Pad,
}

pub struct Element {
    element: gstreamer::Element,
}

pub struct AppSink {
    appsink: gstreamer_app::AppSink,
}

pub struct Playbin3Builder {
    uri: Option<String>,
    video_sink: Option<Element>,
    audio_sink: Option<Element>,
    text_sink: Option<Element>,
}

#[test]
fn gst_parse_pipeline() {
    let gst = Gst::new();
    let pipeline = gst
        .pipeline_from_str("videotestsrc ! autovideosink")
        .expect("Failed to create pipeline");
    println!("{:?}", pipeline);
}

#[test]
fn gst_parse_invalid_pipeline() {
    let gst = Gst::new();
    let result = gst.pipeline_from_str("invalidpipeline");
    assert!(result.is_err());
}

#[test]
fn gst_play_pipeline() {
    let gst = Gst::new();
    let pipeline = gst
        .pipeline_from_str("videotestsrc ! autovideosink")
        .expect("Failed to create pipeline");
    let bus = pipeline.bus().expect("Failed to get bus from pipeline");

    pipeline
        .set_state(gstreamer::State::Playing)
        .expect("Unable to set the pipeline to the `Playing` state");

    for msg in bus.iter_timed(None) {
        use gstreamer::MessageView;

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

    pipeline
        .set_state(gstreamer::State::Null)
        .expect("Unable to set the pipeline to the `Null` state");
}

#[test]
#[ignore]
fn gstreamer_unwrapped() {
    gstreamer::init();
    let uri = "https://gstreamer.freedesktop.org/data/media/sintel_trailer-480p.webm";
    let pipeline = gstreamer::parse::launch(&format!("playbin uri={}", uri)).unwrap();
    use gstreamer::prelude::*;

    pipeline.set_state(gstreamer::State::Playing).unwrap();

    let bus = pipeline.bus().unwrap();
    for msg in bus.iter_timed(gstreamer::ClockTime::NONE) {
        use gstreamer::MessageView;

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

    pipeline.set_state(gstreamer::State::Null).unwrap();
}

#[test]
fn test_appsink() {
    let gst = Gst::new();
    let pipeline = gst
        .pipeline_from_str(
            "videotestsrc ! videoconvert | capsfilter name=video-filter ! appsink name=video-sink",
        )
        .expect("Failed to create pipeline");

    // let video_sink = pipeline.

    let bus = pipeline.bus().expect("Failed to get bus from pipeline");

    let sink = pipeline
        .pipeline
        .by_name("video-sink")
        .expect("Sink not found")
        .downcast::<gstreamer_app::AppSink>()
        .expect("Failed to downcast to AppSink");
    let capsfilter = pipeline
        .pipeline
        .by_name("video-filter")
        .expect("Capsfilter not found");

    let caps = gstreamer::Caps::builder("video/x-raw")
        .field("format", "RGBA")
        .build();
    capsfilter.set_property("caps", &caps);

    sink.set_callbacks(
        gstreamer_app::AppSinkCallbacks::builder()
            .new_sample(|sink| {
                // foo
                Ok(gstreamer::FlowSuccess::Ok)
            })
            .build(),
    );

    // let appsink = sink
    //     .dynamic_cast::<gstreamer_app::AppSink>()
    //     .expect("Failed to cast to AppSink");

    pipeline
        .set_state(gstreamer::State::Playing)
        .expect("Unable to set the pipeline to the `Playing` state");

    for msg in bus.iter_timed(None) {
        use gstreamer::MessageView;

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

#[test]
fn gst_test_manual_pipeline() {
    use gstreamer as gst;
    use gstreamer::prelude::*;
    // Initialize GStreamer
    gst::init().unwrap();

    // Create a new pipeline
    let pipeline = gst::Pipeline::new();

    // Create elements for the pipeline
    let src = gst::ElementFactory::make("videotestsrc").build().unwrap();
    let sink = gst::ElementFactory::make("autovideosink").build().unwrap();

    // Add elements to the pipeline
    pipeline.add_many(&[&src, &sink]).unwrap();

    // Link elements together
    src.link(&sink).unwrap();

    // Set the pipeline to the playing state
    pipeline.set_state(gst::State::Playing).unwrap();

    // Start the main event loop
    // let main_loop = glib::MainLoop::new(None, false);
    // main_loop.run();
    // Shut down the pipeline and GStreamer
    let bus = pipeline.bus().unwrap();
    let messages = bus.iter_timed(gst::ClockTime::NONE);
    for msg in messages {
        dbg!(msg);
    }
    pipeline.set_state(gst::State::Null).unwrap();
}