feat(gst): Added gst a high level wrapper over gstreamer

chore(example): Added hdr-gstreamer-wgpu example
chore(license): Added MIT license to all crates

gst/src/errors.rs

@@ -0,0 +1,7 @@
+pub use error_stack::{Report, ResultExt};
+
+#[derive(Debug, thiserror::Error)]
+#[error("An error occurred")]
+pub struct Error;
+
+pub type Result<T, E = error_stack::Report<Error>> = core::result::Result<T, E>;

gst/src/lib.rs

@@ -0,0 +1,245 @@
+pub mod errors;
+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 Sink {
+    element: gstreamer::Element,
+}
+
+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<gstreamer::ClockTime>>,
+    ) -> gstreamer::bus::Iter<'_> {
+        self.bus.iter_timed(timeout)
+    }
+}
+
+/// Pads are link points between elements
+pub struct Pad {
+    pad: gstreamer::Pad,
+}
+
+pub struct Element {
+    element: gstreamer::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(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)
+        .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(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;
+            }
+            _ => (),
+        }
+    }
+}