refactor(workspace): move crates under crates/

crates/gst/.gitignore

@@ -0,0 +1,3 @@
+/result
+/target
+.direnv

crates/gst/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "gst"
+version = "0.1.0"
+edition = "2024"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+error-stack = "0.6"
+futures = "0.3.31"
+futures-lite = "2.6.1"
+glib = "0.21.5"
+glib-sys = "0.21.5"
+gstreamer = { version = "0.24.4", features = ["v1_26"] }
+gstreamer-app = { version = "0.24.4", features = ["v1_26"] }
+gstreamer-video = { version = "0.24.4", features = ["v1_26"] }
+gstreamer-base = { version = "0.24.4", features = ["v1_26"] }
+thiserror = "2.0"
+tracing = { version = "0.1", features = ["log"] }
+bitflags = "2.10.0"
+
+[dev-dependencies]
+smol = "2.0.2"
+tracing-subscriber = "0.3.22"

crates/gst/src/bin.rs

@@ -0,0 +1,38 @@
+use crate::priv_prelude::*;
+
+wrap_gst!(Bin);
+parent_child!(Element, Bin);
+
+impl Bin {
+    pub fn new(name: impl AsRef<str>) -> Self {
+        let bin = gstreamer::Bin::with_name(name.as_ref());
+        Bin { inner: bin }
+    }
+
+    pub fn add(&mut self, element: &impl ChildOf<Element>) -> Result<&mut Self> {
+        self.inner
+            .add(&element.upcast_ref().inner)
+            .change_context(Error)
+            .attach("Failed to add element to bin")?;
+        Ok(self)
+    }
+
+    pub fn add_many<'a, E: ChildOf<Element> + 'a>(
+        &mut self,
+        elements: impl IntoIterator<Item = &'a E>,
+    ) -> Result<&mut Self> {
+        self.inner
+            .add_many(elements.into_iter().map(|e| &e.upcast_ref().inner))
+            .change_context(Error)
+            .attach("Failed to add elements to bin")?;
+        Ok(self)
+    }
+
+    pub fn add_pad(&mut self, pad: &Pad) -> Result<&mut Self> {
+        self.inner
+            .add_pad(&pad.inner)
+            .change_context(Error)
+            .attach("Failed to add pad to bin")?;
+        Ok(self)
+    }
+}

crates/gst/src/bus.rs

@@ -0,0 +1,27 @@
+use crate::priv_prelude::*;
+
+wrap_gst!(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.inner.iter_timed(clocktime)
+    }
+
+    pub fn stream(&self) -> gstreamer::bus::BusStream {
+        self.inner.stream()
+    }
+
+    pub fn filtered_stream<'a>(
+        &self,
+        msg_types: &'a [gstreamer::MessageType],
+    ) -> impl futures::stream::FusedStream<Item = gstreamer::Message> + Unpin + Send + 'a {
+        self.inner.stream_filtered(msg_types)
+    }
+}

crates/gst/src/caps.rs

@@ -0,0 +1,78 @@
+use gstreamer::Fraction;
+
+#[derive(Debug, Clone)]
+#[repr(transparent)]
+pub struct Caps {
+    pub(crate) inner: gstreamer::caps::Caps,
+}
+
+impl Caps {
+    pub fn builder(cs: CapsType) -> CapsBuilder {
+        CapsBuilder::new(cs)
+    }
+}
+
+pub struct CapsBuilder {
+    inner: gstreamer::caps::Builder<gstreamer::caps::NoFeature>,
+}
+
+impl CapsBuilder {
+    pub fn field<V: Into<glib::Value> + Send>(mut self, name: impl AsRef<str>, value: V) -> Self {
+        self.inner = self.inner.field(name.as_ref(), value);
+        self
+    }
+
+    pub fn build(self) -> Caps {
+        Caps {
+            inner: self.inner.build(),
+        }
+    }
+}
+
+pub enum CapsType {
+    Video,
+    Audio,
+    Text,
+}
+
+impl CapsType {
+    pub fn as_str(&self) -> &str {
+        match self {
+            CapsType::Video => "video/x-raw",
+            CapsType::Audio => "audio/x-raw",
+            CapsType::Text => "text/x-raw",
+        }
+    }
+}
+
+impl CapsBuilder {
+    pub fn new(cs: CapsType) -> Self {
+        CapsBuilder {
+            inner: gstreamer::Caps::builder(cs.as_str()),
+        }
+    }
+}
+
+impl Caps {
+    pub fn format(&self) -> Option<gstreamer_video::VideoFormat> {
+        self.inner
+            .structure(0)
+            .and_then(|s| s.get::<&str>("format").ok())
+            .map(|s| gstreamer_video::VideoFormat::from_string(s))
+    }
+    pub fn width(&self) -> Option<i32> {
+        self.inner
+            .structure(0)
+            .and_then(|s| s.get::<i32>("width").ok())
+    }
+    pub fn height(&self) -> Option<i32> {
+        self.inner
+            .structure(0)
+            .and_then(|s| s.get::<i32>("height").ok())
+    }
+    pub fn framerate(&self) -> Option<gstreamer::Fraction> {
+        self.inner
+            .structure(0)
+            .and_then(|s| s.get::<Fraction>("framerate").ok())
+    }
+}

crates/gst/src/element.rs

@@ -0,0 +1,133 @@
+use crate::priv_prelude::*;
+use crate::wrap_gst;
+
+wrap_gst!(Element, gstreamer::Element);
+
+// pub trait IsElement {
+//     fn upcast_ref(&self) -> ∈
+//     fn into_element(self) -> Element;
+//     fn pad(&self, name: &str) -> Option<Pad> {
+//         use gstreamer::prelude::*;
+//         self.upcast_ref().inner.static_pad(name).map(Pad::from)
+//     }
+// }
+
+// impl IsElement for Element {
+//     fn upcast_ref(&self) -> &Element {
+//         self
+//     }
+//
+//     fn into_element(self) -> Element {
+//         self
+//     }
+// }
+
+impl Element {
+    pub fn pad(&self, name: impl AsRef<str>) -> Option<Pad> {
+        use gstreamer::prelude::*;
+        self.inner.static_pad(name.as_ref()).map(Pad::from)
+    }
+
+    pub fn bus(&self) -> Result<Bus> {
+        use gstreamer::prelude::*;
+        self.inner
+            .bus()
+            .map(Bus::from)
+            .ok_or(Error)
+            .attach_with(|| format!("Failed to get bus from Element: {}", self.inner.name()))
+    }
+}
+
+pub trait Sink: ChildOf<Element> {
+    fn sink(&self, name: impl AsRef<str>) -> Pad {
+        self.upcast_ref()
+            .pad(name)
+            .expect("Sink element has no sink pad")
+    }
+}
+pub trait Source: ChildOf<Element> {
+    fn source(&self, name: impl AsRef<str>) -> Pad {
+        self.upcast_ref()
+            .pad(name)
+            .expect("Source element has no src pad")
+    }
+
+    fn link<S: Sink>(&self, sink: &S) -> Result<Bin>
+    where
+        Self: Sized,
+    {
+        use gstreamer::prelude::*;
+        if let Ok(bin) = self.upcast_ref().inner.clone().downcast::<gstreamer::Bin>() {
+            bin.add(&sink.upcast_ref().inner)
+                .change_context(Error)
+                .attach("Failed to add sink to bin")?;
+            self.upcast_ref()
+                .inner
+                .link(&sink.upcast_ref().inner)
+                .change_context(Error)
+                .attach("Failed to link elements")?;
+            Ok(Bin::from(bin))
+        } else {
+            let bin = gstreamer::Bin::builder()
+                .name(format!(
+                    "{}-link-{}",
+                    self.upcast_ref().inner.name(),
+                    sink.upcast_ref().inner.name()
+                ))
+                .build();
+            bin.add(&self.upcast_ref().inner)
+                .change_context(Error)
+                .attach("Failed to add source to bin")?;
+            bin.add(&sink.upcast_ref().inner)
+                .change_context(Error)
+                .attach("Failed to add sink to bin")?;
+            self.upcast_ref()
+                .inner
+                .link(&sink.upcast_ref().inner)
+                .change_context(Error)
+                .attach("Failed to link elements")?;
+            if let Some(sink_pad) = self.upcast_ref().pad("sink") {
+                let ghost_pad = Pad::ghost(&sink_pad)?;
+                bin.add_pad(&ghost_pad.inner)
+                    .change_context(Error)
+                    .attach("Failed to add src pad to bin")?;
+                ghost_pad.activate(true)?;
+            }
+            Ok(From::from(bin))
+        }
+    }
+
+    // fn link_pad<S: Sink>(&self, sink: &S, src_pad_name: &str, sink_pad_name: &str) -> Result<()> {
+    //     use gstreamer::prelude::*;
+    //     let src_pad = self
+    //         .upcast_ref()
+    //         .pad(src_pad_name)
+    //         .ok_or(Error)
+    //         .attach("Source pad not found")?;
+    //     let sink_pad = sink
+    //         .upcast_ref()
+    //         .pad(sink_pad_name)
+    //         .ok_or(Error)
+    //         .attach("Sink pad not found")?;
+    //     src_pad
+    //         .inner
+    //         .link(&sink_pad.inner)
+    //         .change_context(Error)
+    //         .attach("Failed to link source pad to sink pad")?;
+    //     Ok(())
+    // }
+}
+
+pub trait ElementExt: ChildOf<Element> + Sync {
+    #[track_caller]
+    fn bus(&self) -> Result<Bus> {
+        self.upcast_ref().bus()
+    }
+
+    #[track_caller]
+    fn pad(&self, name: impl AsRef<str>) -> Option<Pad> {
+        self.upcast_ref().pad(name)
+    }
+}
+
+impl<T: ChildOf<Element> + Sync> ElementExt for T {}

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

crates/gst/src/lib.rs

@@ -0,0 +1,64 @@
+pub mod bin;
+pub mod bus;
+pub mod caps;
+pub mod element;
+pub mod errors;
+pub mod pad;
+pub mod pipeline;
+pub mod plugins;
+#[macro_use]
+pub mod wrapper;
+pub mod sample;
+
+pub use bin::*;
+pub use bus::*;
+pub use caps::*;
+pub use element::*;
+pub use gstreamer;
+#[doc(inline)]
+pub use gstreamer::{Message, MessageType, MessageView, State};
+pub use gstreamer_video::VideoFormat;
+pub use pad::*;
+pub use pipeline::*;
+pub use plugins::*;
+pub use sample::*;
+
+pub(crate) mod priv_prelude {
+    pub use crate::errors::*;
+    pub use crate::wrapper::*;
+    pub use crate::*;
+    pub use gstreamer::prelude::ElementExt as _;
+    pub use gstreamer::prelude::*;
+    #[track_caller]
+    pub fn duration_to_clocktime(
+        timeout: impl Into<Option<core::time::Duration>>,
+    ) -> Result<Option<gstreamer::ClockTime>> {
+        match timeout.into() {
+            Some(dur) => {
+                let clocktime = gstreamer::ClockTime::try_from(dur)
+                    .change_context(Error)
+                    .attach("Failed to convert duration to ClockTime")?;
+                Ok(Some(clocktime))
+            }
+            None => Ok(gstreamer::ClockTime::NONE),
+        }
+    }
+}
+
+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,
+    })
+});
+
+pub struct Gst {
+    __private: core::marker::PhantomData<()>,
+}
+
+impl Gst {
+    pub fn new() -> Arc<Self> {
+        Arc::clone(&GST)
+    }
+}

crates/gst/src/pad.rs

@@ -0,0 +1,45 @@
+use crate::priv_prelude::*;
+
+wrap_gst!(Pad, gstreamer::Pad);
+
+impl Pad {
+    #[track_caller]
+    pub fn ghost(target: &Pad) -> Result<Pad> {
+        let ghost_pad = gstreamer::GhostPad::with_target(&target.inner)
+            .change_context(Error)
+            .attach("Failed to create ghost pad")?;
+        Ok(Pad {
+            inner: ghost_pad.upcast(),
+        })
+    }
+
+    #[track_caller]
+    pub fn link(&self, peer: &Pad) -> Result<()> {
+        use gstreamer::prelude::*;
+        self.inner
+            .link(&peer.inner)
+            .change_context(Error)
+            .attach("Failed to link pads")?;
+        Ok(())
+    }
+
+    #[track_caller]
+    pub fn current_caps(&self) -> Result<Caps> {
+        let caps = self
+            .inner
+            .current_caps()
+            .ok_or(Error)
+            .attach("Failed to get pad caps")?;
+        Ok(Caps { inner: caps })
+    }
+
+    #[track_caller]
+    pub fn activate(&self, activate: bool) -> Result<()> {
+        use gstreamer::prelude::*;
+        self.inner
+            .set_active(activate)
+            .change_context(Error)
+            .attach("Failed to set_active pad")?;
+        Ok(())
+    }
+}

crates/gst/src/pipeline.rs

@@ -0,0 +1,211 @@
+use crate::priv_prelude::*;
+
+wrap_gst!(Pipeline);
+parent_child!(Element, Pipeline);
+parent_child!(Bin, Pipeline);
+
+impl Drop for Pipeline {
+    fn drop(&mut self) {
+        let _ = self.inner.set_state(gstreamer::State::Null);
+    }
+}
+
+impl Pipeline {
+    #[track_caller]
+    pub fn bus(&self) -> Result<Bus> {
+        let bus = self
+            .inner
+            .bus()
+            .ok_or(Error)
+            .attach("Failed to get bus from pipeline")?;
+        Ok(Bus::from_gst(bus))
+    }
+
+    /// Get the state
+    pub fn state(
+        &self,
+        timeout: impl Into<Option<core::time::Duration>>,
+    ) -> Result<gstreamer::State> {
+        let (result, current, _pending) = self.inner.state(duration_to_clocktime(timeout)?);
+        result.change_context(Error).attach("Failed to get state")?;
+        Ok(current)
+    }
+
+    pub fn play(&self) -> Result<()> {
+        self.inner
+            .set_state(gstreamer::State::Playing)
+            .change_context(Error)
+            .attach("Failed to set pipeline to Playing state")?;
+        Ok(())
+    }
+
+    pub fn pause(&self) -> Result<()> {
+        self.inner
+            .set_state(gstreamer::State::Paused)
+            .change_context(Error)
+            .attach("Failed to set pipeline to Paused state")?;
+        Ok(())
+    }
+
+    pub fn ready(&self) -> Result<()> {
+        self.inner
+            .set_state(gstreamer::State::Ready)
+            .change_context(Error)
+            .attach("Failed to set pipeline to Ready state")?;
+        Ok(())
+    }
+
+    pub fn stop(&self) -> Result<()> {
+        self.inner
+            .set_state(gstreamer::State::Null)
+            .change_context(Error)
+            .attach("Failed to set pipeline to Null state")?;
+        Ok(())
+    }
+
+    pub fn set_state(&self, state: gstreamer::State) -> Result<gstreamer::StateChangeSuccess> {
+        let result = self
+            .inner
+            .set_state(state)
+            .change_context(Error)
+            .attach("Failed to set pipeline state")?;
+        Ok(result)
+    }
+
+    pub async fn wait_for(&self, state: gstreamer::State) -> Result<()> {
+        let current_state = self.state(core::time::Duration::ZERO)?;
+        if current_state == state {
+            Ok(())
+        } else {
+            // use futures::stream::StreamExt;
+            use futures_lite::stream::StreamExt as _;
+            self.bus()?
+                .filtered_stream(&[MessageType::StateChanged])
+                .find(|message: &gstreamer::Message| {
+                    let view = message.view();
+                    if let gstreamer::MessageView::StateChanged(changed) = view {
+                        changed.current() == state
+                            && changed.src().is_some_and(|s| s == &self.inner)
+                    } else {
+                        false
+                    }
+                })
+                .await;
+            Ok(())
+        }
+    }
+
+    pub async fn wait_for_states(&self, states: impl AsRef<[gstreamer::State]>) -> Result<()> {
+        let current_state = self.state(core::time::Duration::ZERO)?;
+        let states = states.as_ref();
+        if states.contains(&current_state) {
+            Ok(())
+        } else {
+            use futures_lite::stream::StreamExt as _;
+            self.bus()?
+                .filtered_stream(&[MessageType::StateChanged])
+                .find(|message: &gstreamer::Message| {
+                    let view = message.view();
+                    if let gstreamer::MessageView::StateChanged(changed) = view {
+                        states.contains(&changed.current())
+                            && changed.src().is_some_and(|s| s == &self.inner)
+                    } else {
+                        false
+                    }
+                })
+                .await;
+            Ok(())
+        }
+    }
+
+    pub async fn wait_for_message<'a, F2>(
+        &self,
+        filter: Option<&'a [gstreamer::MessageType]>,
+        filter_fn: F2,
+    ) -> Result<gstreamer::Message>
+    where
+        F2: Fn(&gstreamer::Message) -> bool + Send + 'a,
+    {
+        use futures_lite::stream::StreamExt as _;
+        match filter {
+            Some(filter) => {
+                let message = self.bus()?.filtered_stream(filter).find(filter_fn).await;
+                match message {
+                    Some(msg) => Ok(msg),
+                    None => {
+                        Err(Error).attach("Failed to find message matching the provided filter")
+                    }
+                }
+            }
+            None => {
+                let message = self.bus()?.stream().find(filter_fn).await;
+                match message {
+                    Some(msg) => Ok(msg),
+                    None => {
+                        Err(Error).attach("Failed to find message matching the provided filter")
+                    }
+                }
+            }
+        }
+    }
+}
+
+pub trait PipelineExt: ChildOf<Pipeline> + Sync {
+    // #[track_caller]
+    // fn bus(&self) -> Result<Bus> {
+    //     self.upcast_ref().bus()
+    // }
+    #[track_caller]
+    fn play(&self) -> Result<()> {
+        self.upcast_ref().play()
+    }
+    #[track_caller]
+    fn pause(&self) -> Result<()> {
+        self.upcast_ref().pause()
+    }
+    #[track_caller]
+    fn ready(&self) -> Result<()> {
+        self.upcast_ref().ready()
+    }
+
+    #[track_caller]
+    fn stop(&self) -> Result<()> {
+        self.upcast_ref().stop()
+    }
+
+    #[track_caller]
+    fn set_state(&self, state: gstreamer::State) -> Result<gstreamer::StateChangeSuccess> {
+        self.upcast_ref().set_state(state)
+    }
+    #[track_caller]
+    fn state(&self, timeout: impl Into<Option<core::time::Duration>>) -> Result<gstreamer::State> {
+        self.upcast_ref().state(timeout)
+    }
+
+    fn wait_for(
+        &self,
+        state: gstreamer::State,
+    ) -> impl std::future::Future<Output = Result<()>> + Send {
+        self.upcast_ref().wait_for(state)
+    }
+
+    fn wait_for_states(
+        &self,
+        states: impl AsRef<[gstreamer::State]> + Send,
+    ) -> impl std::future::Future<Output = Result<()>> + Send {
+        self.upcast_ref().wait_for_states(states)
+    }
+
+    fn wait_for_message<'a, F2>(
+        &self,
+        filter: Option<&'a [gstreamer::MessageType]>,
+        filter_fn: F2,
+    ) -> impl std::future::Future<Output = Result<gstreamer::Message>> + Send
+    where
+        F2: Fn(&gstreamer::Message) -> bool + Send + 'a,
+    {
+        self.upcast_ref().wait_for_message(filter, filter_fn)
+    }
+}
+
+impl<T: ChildOf<Pipeline> + Sync> PipelineExt for T {}

crates/gst/src/plugins.rs

@@ -0,0 +1,4 @@
+pub mod app;
+pub mod autodetect;
+pub mod playback;
+pub mod videoconvertscale;

crates/gst/src/plugins/app.rs

@@ -0,0 +1,2 @@
+pub mod appsink;
+pub use appsink::*;

crates/gst/src/plugins/app/appsink.rs

@@ -0,0 +1,278 @@
+use crate::priv_prelude::*;
+
+#[doc(inline)]
+pub use gstreamer_app::AppSinkCallbacks;
+
+wrap_gst!(AppSink, gstreamer::Element);
+parent_child!(Element, AppSink);
+
+pub struct AppSinkBuilder {
+    inner: AppSink,
+    callbacks: Option<gstreamer_app::app_sink::AppSinkCallbacksBuilder>,
+}
+
+impl AppSinkBuilder {
+    pub fn on_new_sample<F>(mut self, mut f: F) -> Self
+    where
+        F: FnMut(&AppSink) -> Result<(), gstreamer::FlowError> + Send + 'static,
+    {
+        let mut callbacks_builder = self
+            .callbacks
+            .take()
+            .unwrap_or_else(gstreamer_app::app_sink::AppSinkCallbacks::builder);
+        callbacks_builder = callbacks_builder.new_sample(move |appsink| {
+            use glib::object::Cast;
+            let element = appsink.upcast_ref::<gstreamer::Element>();
+            let appsink = AppSink::from_gst_ref(element);
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(appsink)))
+                .unwrap_or(Err(gstreamer::FlowError::Error))
+                .map(|_| gstreamer::FlowSuccess::Ok)
+        });
+        self.callbacks = Some(callbacks_builder);
+        self
+    }
+
+    pub fn on_new_preroll<F>(mut self, mut f: F) -> Self
+    where
+        F: FnMut(&AppSink) -> Result<(), gstreamer::FlowError> + Send + 'static,
+    {
+        let mut callbacks_builder = self
+            .callbacks
+            .take()
+            .unwrap_or_else(gstreamer_app::app_sink::AppSinkCallbacks::builder);
+        callbacks_builder = callbacks_builder.new_preroll(move |appsink| {
+            use glib::object::Cast;
+            let element = appsink.upcast_ref::<gstreamer::Element>();
+            let appsink = AppSink::from_gst_ref(element);
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(appsink)))
+                .unwrap_or(Err(gstreamer::FlowError::Error))
+                .map(|_| gstreamer::FlowSuccess::Ok)
+        });
+        self.callbacks = Some(callbacks_builder);
+        self
+    }
+
+    pub fn build(self) -> AppSink {
+        let AppSinkBuilder { inner, callbacks } = self;
+        if let Some(callbacks) = callbacks {
+            inner.appsink().set_callbacks(callbacks.build());
+        }
+        inner
+    }
+}
+
+impl Sink for AppSink {}
+
+impl AppSink {
+    pub fn builder(name: impl AsRef<str>) -> AppSinkBuilder {
+        let inner = AppSink::new(name).expect("Failed to create AppSink");
+        AppSinkBuilder {
+            inner,
+            callbacks: None,
+        }
+    }
+    fn appsink(&self) -> &gstreamer_app::AppSink {
+        self.inner
+            .downcast_ref::<gstreamer_app::AppSink>()
+            .expect("Failed to downcast to AppSink")
+    }
+
+    pub fn new(name: impl AsRef<str>) -> Result<Self> {
+        let inner = gstreamer::ElementFactory::make("appsink")
+            .name(name.as_ref())
+            .build()
+            .change_context(Error)
+            .attach("Failed to create appsink element")?;
+        Ok(AppSink { inner })
+    }
+
+    pub fn emit_signals(&mut self, emit: bool) -> &mut Self {
+        self.inner.set_property("emit-signals", emit);
+        self
+    }
+
+    pub fn async_(&mut self, async_: bool) -> &mut Self {
+        self.inner.set_property("async", async_);
+        self
+    }
+
+    pub fn sync(&mut self, sync: bool) -> &mut Self {
+        self.inner.set_property("sync", sync);
+        self
+    }
+
+    pub fn drop(&mut self, drop: bool) -> &mut Self {
+        self.inner.set_property("drop", drop);
+        self
+    }
+
+    pub fn caps(&mut self, caps: Caps) -> &mut Self {
+        self.inner.set_property("caps", caps.inner);
+        self
+    }
+
+    pub fn callbacks(&mut self, callbacks: gstreamer_app::AppSinkCallbacks) -> &mut Self {
+        self.appsink().set_callbacks(callbacks);
+        self
+    }
+
+    pub fn on_new_sample<F>(&mut self, mut f: F) -> &mut Self
+    where
+        F: FnMut(&AppSink) -> Result<(), gstreamer::FlowError> + Send + 'static,
+    {
+        self.emit_signals(true).callbacks(
+            AppSinkCallbacks::builder()
+                .new_sample(move |appsink| {
+                    use glib::object::Cast;
+                    let element = appsink.upcast_ref::<gstreamer::Element>();
+                    let appsink = AppSink::from_gst_ref(element);
+                    std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(appsink)))
+                        .unwrap_or(Err(gstreamer::FlowError::Error))
+                        .map(|_| gstreamer::FlowSuccess::Ok)
+                })
+                .build(),
+        )
+    }
+
+    pub fn pull_sample(&self) -> Result<Sample> {
+        self.appsink()
+            .pull_sample()
+            .change_context(Error)
+            .attach("Failed to pull sample from AppSink")
+            .map(Sample::from)
+    }
+    pub fn try_pull_sample(
+        &self,
+        timeout: impl Into<Option<core::time::Duration>>,
+    ) -> Result<Option<Sample>> {
+        Ok(self
+            .appsink()
+            .try_pull_sample(duration_to_clocktime(timeout)?)
+            .map(From::from))
+    }
+
+    pub fn pull_preroll(&self) -> Result<Sample> {
+        self.appsink()
+            .pull_preroll()
+            .change_context(Error)
+            .attach("Failed to pull preroll sample from AppSink")
+            .map(Sample::from)
+    }
+
+    pub fn try_pull_preroll(
+        &self,
+        timeout: impl Into<Option<core::time::Duration>>,
+    ) -> Result<Option<Sample>> {
+        Ok(self
+            .appsink()
+            .try_pull_preroll(duration_to_clocktime(timeout)?)
+            .map(From::from))
+    }
+}
+
+#[test]
+fn test_appsink() {
+    use gstreamer::prelude::*;
+    use tracing_subscriber::prelude::*;
+    tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::fmt::layer()
+                .with_thread_ids(true)
+                .with_file(true),
+        )
+        .init();
+    tracing::info!("Linking videoconvert to appsink");
+    Gst::new();
+    let playbin3 = playback::Playbin3::new("pppppppppppppppppppppppppppppp").unwrap().with_uri("https://jellyfin.tsuba.darksailor.dev/Items/6010382cf25273e624d305907010d773/Download?api_key=036c140222464878862231ef66a2bc9c");
+
+    let video_convert = plugins::videoconvertscale::VideoConvert::new("vcvcvcvcvcvcvcvcvcvcvcvcvc")
+        .expect("Create videoconvert");
+    let mut appsink = app::AppSink::new("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa").expect("Create appsink");
+    appsink.caps(
+        Caps::builder(CapsType::Video)
+            .field("format", "RGB")
+            .build(),
+    );
+
+    let video_sink = video_convert
+        .link(&appsink)
+        .expect("Link videoconvert to appsink");
+
+    let playbin3 = playbin3.with_video_sink(&video_sink);
+    playbin3.play().expect("Play video");
+    let bus = playbin3.bus().unwrap();
+    for msg in bus.iter_timed(None) {
+        match msg.view() {
+            gstreamer::MessageView::Eos(..) => {
+                tracing::info!("End of stream reached");
+                break;
+            }
+            gstreamer::MessageView::Error(err) => {
+                tracing::error!(
+                    "Error from {:?}: {} ({:?})",
+                    err.src().map(|s| s.path_string()),
+                    err.error(),
+                    err.debug()
+                );
+                break;
+            }
+            gstreamer::MessageView::StateChanged(state) => {
+                eprintln!(
+                    "State changed from {:?} to \x1b[33m{:?}\x1b[0m for {:?}",
+                    state.old(),
+                    state.current(),
+                    state.src().map(|s| s.path_string())
+                );
+            }
+            _ => {}
+        }
+        // tracing::info!("{:#?}", &msg.view());
+    }
+    // std::thread::sleep(std::time::Duration::from_secs(5));
+}
+
+#[test]
+fn test_appsink_metadata() {
+    use tracing_subscriber::prelude::*;
+    tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::fmt::layer()
+                .with_thread_ids(true)
+                .with_file(true),
+        )
+        .init();
+
+    crate::Gst::new();
+
+    let url = "https://jellyfin.tsuba.darksailor.dev/Items/6010382cf25273e624d305907010d773/Download?api_key=036c140222464878862231ef66a2bc9c";
+
+    let videoconvert = crate::plugins::videoconvertscale::VideoConvert::new("iced-video-convert")
+        // .unwrap();
+        // .with_output_format(gst::plugins::videoconvertscale::VideoFormat::Rgba)
+        .unwrap();
+    let appsink = crate::plugins::app::AppSink::new("iced-video-sink")
+        .unwrap()
+        .with_async(true)
+        .with_sync(true);
+
+    let video_sink = videoconvert.link(&appsink).unwrap();
+    let playbin = crate::plugins::playback::Playbin3::new("iced-video")
+        .unwrap()
+        .with_uri(url)
+        .with_video_sink(&video_sink);
+
+    playbin.pause().unwrap();
+
+    smol::block_on(async {
+        playbin.wait_for(gstreamer::State::Paused).await.unwrap();
+    });
+    // std::thread::sleep(core::time::Duration::from_secs(1));
+    let pad = appsink.pad("sink").unwrap();
+    let caps = pad.current_caps().unwrap();
+    let format = caps.format();
+    let height = caps.height();
+    let width = caps.width();
+    let framerate = caps.framerate();
+    dbg!(&format, height, width, framerate);
+    dbg!(&caps);
+}

crates/gst/src/plugins/autodetect.rs

@@ -0,0 +1,2 @@
+pub mod autovideosink;
+pub use autovideosink::*;

crates/gst/src/plugins/autodetect/autovideosink.rs

@@ -0,0 +1,18 @@
+use crate::priv_prelude::*;
+
+wrap_gst!(AutoVideoSink, gstreamer::Element);
+parent_child!(Element, AutoVideoSink);
+parent_child!(Bin, AutoVideoSink, downcast);
+
+impl Sink for AutoVideoSink {}
+
+impl AutoVideoSink {
+    pub fn new(name: impl AsRef<str>) -> Result<Self> {
+        let element = gstreamer::ElementFactory::make("autovideosink")
+            .name(name.as_ref())
+            .build()
+            .change_context(Error)
+            .attach("Failed to create autovideosink element")?;
+        Ok(AutoVideoSink { inner: element })
+    }
+}

crates/gst/src/plugins/playback.rs

@@ -0,0 +1,71 @@
+pub mod playbin3;
+pub use playbin3::*;
+pub mod playbin;
+pub use playbin::*;
+
+bitflags::bitflags! {
+    /// Extra flags to configure the behaviour of the sinks.
+    #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+    pub struct PlayFlags: u32 {
+        /// Render the video stream
+        const VIDEO = (1 << 0);
+        /// Render the audio stream
+        const AUDIO = (1 << 1);
+        /// Render subtitles
+        const TEXT = (1 << 2);
+        /// Render visualisation when no video is present
+        const VIS = (1 << 3);
+        /// Use software volume
+        const SOFT_VOLUME = (1 << 4);
+        /// Only use native audio formats
+        const NATIVE_AUDIO = (1 << 5);
+        /// Only use native video formats
+        const NATIVE_VIDEO = (1 << 6);
+        /// Attempt progressive download buffering
+        const DOWNLOAD = (1 << 7);
+        /// Buffer demuxed/parsed data
+        const BUFFERING = (1 << 8);
+        /// Deinterlace video if necessary
+        const DEINTERLACE = (1 << 9);
+        /// Use software color balance
+        const SOFT_COLORBALANCE = (1 << 10);
+        /// Force audio/video filter(s) to be applied
+        const FORCE_FILTERS = (1 << 11);
+        /// Force only software-based decoders (no effect for playbin3)
+        const FORCE_SW_DECODERS = (1 << 12);
+    }
+}
+
+const _: () = {
+    use glib::types::StaticType;
+    impl glib::types::StaticType for PlayFlags {
+        #[inline]
+        #[doc(alias = "gst_play_flags_get_type")]
+        fn static_type() -> glib::Type {
+            glib::Type::from_name("GstPlayFlags").expect("GstPlayFlags type not found")
+        }
+    }
+
+    impl glib::value::ToValue for PlayFlags {
+        #[inline]
+        fn to_value(&self) -> glib::Value {
+            let value = self.bits().to_value();
+            value
+                .transform_with_type(Self::static_type())
+                .expect("Failed to transform PlayFlags(u32) to GstPlayFlags")
+        }
+
+        #[inline]
+        fn value_type(&self) -> glib::Type {
+            Self::static_type()
+        }
+    }
+
+    impl From<PlayFlags> for glib::Value {
+        #[inline]
+        fn from(v: PlayFlags) -> Self {
+            // skip_assert_initialized!();
+            glib::value::ToValue::to_value(&v)
+        }
+    }
+};

crates/gst/src/plugins/playback/playbin.rs

@@ -0,0 +1,82 @@
+use crate::priv_prelude::*;
+
+wrap_gst!(Playbin, gstreamer::Element);
+parent_child!(Element, Playbin);
+parent_child!(Pipeline, Playbin, downcast);
+parent_child!(Bin, Playbin, downcast);
+
+impl Drop for Playbin {
+    fn drop(&mut self) {
+        self.set_state(gstreamer::State::Null).ok();
+    }
+}
+
+impl Playbin {
+    pub fn new(name: impl AsRef<str>) -> Result<Self> {
+        gstreamer::ElementFactory::make("playbin3")
+            .name(name.as_ref())
+            .build()
+            .map(|element| Playbin { inner: element })
+            .change_context(Error)
+    }
+
+    pub fn with_uri(self, uri: impl AsRef<str>) -> Self {
+        self.inner.set_property("uri", uri.as_ref());
+        self
+    }
+
+    pub fn with_buffer_duration(self, duration: impl Into<Option<core::time::Duration>>) -> Self {
+        let duration = match duration.into() {
+            Some(dur) => dur.as_secs() as i64,
+            None => -1,
+        };
+        self.inner.set_property("buffer-duration", duration);
+        self
+    }
+
+    pub fn with_buffer_size(self, size: impl Into<Option<u32>>) -> Self {
+        let size = match size.into() {
+            Some(size) => size as i32,
+            None => -1,
+        };
+        self.inner.set_property("buffer-size", size);
+        self
+    }
+
+    /// Sets the maximum size of the ring buffer in bytes.
+    pub fn with_ring_buffer_max_size(self, size: u64) -> Self {
+        self.inner.set_property("ring-buffer-max-size", size);
+        self
+    }
+
+    pub fn with_video_sink(self, video_sink: &impl ChildOf<Element>) -> Self {
+        self.inner
+            .set_property("video-sink", &video_sink.upcast_ref().inner);
+        self
+    }
+
+    pub fn with_text_sink(self, text_sink: &impl ChildOf<Element>) -> Self {
+        self.inner
+            .set_property("text-sink", &text_sink.upcast_ref().inner);
+        self
+    }
+
+    pub fn with_audio_sink(self, audio_sink: &impl ChildOf<Element>) -> Self {
+        self.inner
+            .set_property("audio-sink", &audio_sink.upcast_ref().inner);
+        self
+    }
+
+    pub fn set_volume(&self, volume: f64) {
+        self.inner.set_property("volume", volume.clamp(1.0, 100.0))
+    }
+
+    pub fn get_volume(&self) -> f64 {
+        self.inner.property::<f64>("volume")
+    }
+
+    pub fn with_flags(self, flags: playback::PlayFlags) -> Self {
+        self.inner.set_property("flags", flags);
+        self
+    }
+}

crates/gst/src/plugins/playback/playbin3.rs

@@ -0,0 +1,95 @@
+use crate::priv_prelude::*;
+use playback::PlayFlags;
+
+wrap_gst!(Playbin3, gstreamer::Element);
+parent_child!(Element, Playbin3);
+parent_child!(Pipeline, Playbin3, downcast);
+parent_child!(Bin, Playbin3, downcast);
+
+impl Drop for Playbin3 {
+    fn drop(&mut self) {
+        self.set_state(gstreamer::State::Null).ok();
+    }
+}
+
+impl Playbin3 {
+    pub fn new(name: impl AsRef<str>) -> Result<Self> {
+        gstreamer::ElementFactory::make("playbin3")
+            .name(name.as_ref())
+            .build()
+            .map(|element| Playbin3 { inner: element })
+            .change_context(Error)
+    }
+
+    pub fn with_uri(self, uri: impl AsRef<str>) -> Self {
+        self.inner.set_property("uri", uri.as_ref());
+        self
+    }
+
+    pub fn with_buffer_duration(self, duration: impl Into<Option<core::time::Duration>>) -> Self {
+        let duration = match duration.into() {
+            Some(dur) => dur.as_secs() as i64,
+            None => -1,
+        };
+        self.inner.set_property("buffer-duration", duration);
+        self
+    }
+
+    pub fn with_buffer_size(self, size: impl Into<Option<u32>>) -> Self {
+        let size = match size.into() {
+            Some(size) => size as i32,
+            None => -1,
+        };
+        self.inner.set_property("buffer-size", size);
+        self
+    }
+
+    /// Sets the maximum size of the ring buffer in bytes.
+    pub fn with_ring_buffer_max_size(self, size: u64) -> Self {
+        self.inner.set_property("ring-buffer-max-size", size);
+        self
+    }
+
+    pub fn with_video_sink(self, video_sink: &impl ChildOf<Element>) -> Self {
+        self.inner
+            .set_property("video-sink", &video_sink.upcast_ref().inner);
+        self
+    }
+
+    pub fn with_text_sink(self, text_sink: &impl ChildOf<Element>) -> Self {
+        self.inner
+            .set_property("text-sink", &text_sink.upcast_ref().inner);
+        self
+    }
+
+    pub fn with_audio_sink(self, audio_sink: &impl ChildOf<Element>) -> Self {
+        self.inner
+            .set_property("audio-sink", &audio_sink.upcast_ref().inner);
+        self
+    }
+
+    pub fn set_volume(&self, volume: f64) {
+        self.inner.set_property("volume", volume.clamp(1.0, 100.0))
+    }
+
+    pub fn get_volume(&self) -> f64 {
+        self.inner.property::<f64>("volume")
+    }
+
+    pub fn with_flags(self, flags: playback::PlayFlags) -> Self {
+        self.inner.set_property("flags", flags);
+        self
+    }
+}
+
+impl Playbin3 {
+    pub fn default_flags() -> PlayFlags {
+        PlayFlags::SOFT_COLORBALANCE
+            | PlayFlags::DEINTERLACE
+            | PlayFlags::BUFFERING
+            | PlayFlags::SOFT_VOLUME
+            | PlayFlags::TEXT
+            | PlayFlags::AUDIO
+            | PlayFlags::VIDEO
+    }
+}

crates/gst/src/plugins/videoconvertscale.rs

@@ -0,0 +1,2 @@
+pub mod videoconvert;
+pub use videoconvert::*;

crates/gst/src/plugins/videoconvertscale/videoconvert.rs

@@ -0,0 +1,36 @@
+use crate::priv_prelude::*;
+#[doc(inline)]
+pub use gstreamer_video::VideoFormat;
+
+wrap_gst!(VideoConvert, gstreamer::Element);
+parent_child!(Element, VideoConvert);
+
+impl Sink for VideoConvert {}
+impl Source for VideoConvert {}
+
+impl VideoConvert {
+    pub fn new(name: impl AsRef<str>) -> Result<Self> {
+        let element = gstreamer::ElementFactory::make("videoconvert")
+            .name(name.as_ref())
+            .build()
+            .change_context(Error)
+            .attach("Failed to create videoconvert element")?;
+        Ok(VideoConvert { inner: element })
+    }
+
+    // pub fn with_caps(mut self, caps: &gstreamer::Caps) -> Self {
+    //     use gstreamer::prelude::*;
+    //     self.inner.set_property("caps", caps);
+    //     self
+    // }
+    pub fn with_output_format(self, format: VideoFormat) -> Result<Self> {
+        use gstreamer::prelude::*;
+        let caps = Caps::builder(CapsType::Video)
+            .field("format", format.to_str())
+            .build();
+        self.inner.set_property("caps", &caps.inner);
+        // .change_context(Error)
+        // .attach("Failed to set output format on videoconvert")?;
+        Ok(self)
+    }
+}

crates/gst/src/sample.rs

@@ -0,0 +1,37 @@
+impl From<gstreamer::Sample> for Sample {
+    fn from(inner: gstreamer::Sample) -> Self {
+        Sample { inner }
+    }
+}
+
+#[repr(transparent)]
+#[derive(Debug, Clone)]
+pub struct Sample {
+    pub inner: gstreamer::Sample,
+}
+
+use gstreamer::BufferRef;
+impl Sample {
+    #[doc(alias = "empty")]
+    pub fn new() -> Self {
+        Self {
+            inner: gstreamer::Sample::builder().build(),
+        }
+    }
+
+    pub fn buffer(&self) -> Option<&BufferRef> {
+        self.inner.buffer()
+    }
+
+    pub fn caps(&self) -> Option<&gstreamer::CapsRef> {
+        self.inner.caps()
+    }
+
+    pub fn info(&self) -> Option<&gstreamer::StructureRef> {
+        self.inner.info()
+    }
+
+    // pub fn set_buffer(&mut self) {
+    //     self.inner.set_buffer(None);
+    // }
+}

crates/gst/src/wgpu.rs

@@ -0,0 +1,2 @@
+
+

crates/gst/src/wrapper.rs

@@ -0,0 +1,145 @@
+pub trait GstWrapper {
+    type GstType: glib::prelude::ObjectType;
+    fn from_gst(gst: Self::GstType) -> Self;
+    // fn into_gst(self) -> Self::GstType;
+    fn as_gst_ref(&self) -> &Self::GstType;
+    fn from_gst_ref(gst: &Self::GstType) -> &Self;
+}
+
+#[macro_export]
+macro_rules! wrap_gst {
+    ($name:ident) => {
+        $crate::wrap_gst!($name, gstreamer::$name);
+    };
+    ($name:ident, $inner:ty) => {
+        $crate::wrap_gst!(core $name, $inner);
+        $crate::wrap_gst!($name, $inner, into_inner);
+    };
+    ($name:ident, $inner:ty, skip_inner) => {
+        $crate::wrap_gst!(core $name, $inner);
+    };
+
+    (core $name:ident, $inner:ty) => {
+        #[derive(Debug, Clone)]
+        #[repr(transparent)]
+        pub struct $name {
+            pub(crate) inner: $inner,
+        }
+
+        // impl From<$name> for $inner {
+        //     fn from(wrapper: $name) -> Self {
+        //         wrapper.into_inner()
+        //     }
+        // }
+
+        impl $name {
+            pub fn into_inner(self) -> $inner {
+                self.inner.clone()
+            }
+        }
+
+        impl $crate::wrapper::GstWrapper for $name {
+            type GstType = $inner;
+
+            fn from_gst(gst: Self::GstType) -> Self {
+                Self { inner: gst }
+            }
+
+            // fn into_gst(self) -> Self::GstType {
+            //     self.inner.clone()
+            // }
+
+            fn as_gst_ref(&self) -> &Self::GstType {
+                &self.inner
+            }
+
+            fn from_gst_ref(gst: &Self::GstType) -> &Self {
+                unsafe { &*(gst as *const Self::GstType as *const Self) }
+            }
+        }
+
+        impl ChildOf<$name> for $name {
+            fn upcast_ref(&self) -> &$name {
+                self
+            }
+        }
+    };
+    ($name:ident, $inner:ty, into_inner) => {
+        impl From<$inner> for $name {
+            fn from(inner: $inner) -> Self {
+                Self { inner }
+            }
+        }
+    };
+}
+
+/// A trait for types that can be upcasted to type T.
+pub trait ChildOf<T> {
+    fn upcast_ref(&self) -> &T;
+}
+
+#[macro_export]
+macro_rules! parent_child {
+    ($parent:ty, $child:ty) => {
+        impl ChildOf<$parent> for $child
+        where
+            $child: GstWrapper,
+            $parent: GstWrapper,
+        {
+            fn upcast_ref(&self) -> &$parent {
+                let upcasted = self.inner.upcast_ref::<<$parent as GstWrapper>::GstType>();
+                unsafe { &*(upcasted as *const <$parent as GstWrapper>::GstType as *const $parent) }
+            }
+        }
+    };
+    ($parent:ty, $child:ty, downcast) => {
+        impl ChildOf<$parent> for $child
+        where
+            $child: GstWrapper,
+            $parent: GstWrapper,
+        {
+            fn upcast_ref(&self) -> &$parent {
+                let downcasted = self
+                    .inner
+                    .downcast_ref::<<$parent as GstWrapper>::GstType>()
+                    .expect(
+                        format!(
+                            "BUG: Failed to downcast GStreamer type from child {} to parent {}",
+                            stringify!($child),
+                            stringify!($parent)
+                        )
+                        .as_str(),
+                    );
+                unsafe {
+                    &*(downcasted as *const <$parent as GstWrapper>::GstType as *const $parent)
+                }
+            }
+        }
+    }; // ($parent:ty, $child:ty, deref) => {
+       //     $crate::parent_child!($parent, $child);
+       //     $crate::parent_child!($parent, $child, __deref);
+       // };
+       //
+       // ($parent:ty, $child:ty, downcast, deref) => {
+       //     $crate::parent_child!($parent, $child, downcast);
+       //     $crate::parent_child!($parent, $child, __deref);
+       // };
+       // ($parent:ty, $child:ty, deref, downcast) => {
+       //     $crate::parent_child!($parent, $child, downcast);
+       //     $crate::parent_child!($parent, $child, __deref);
+       // };
+       //
+       // ($parent:ty, $child:ty, __deref) => {
+       //     impl core::ops::Deref for $child
+       //     where
+       //         $child: GstWrapper,
+       //         $parent: GstWrapper,
+       //     {
+       //         type Target = $parent;
+       //
+       //         fn deref(&self) -> &Self::Target {
+       //             self.upcast_ref()
+       //         }
+       //     }
+       // };
+}