feat: Get iced-video working

gst/src/pipeline.rs

@@ -12,6 +12,7 @@ impl Drop for Pipeline {
 }
 
 impl Pipeline {
+    #[track_caller]
     pub fn bus(&self) -> Result<Bus> {
         let bus = self
             .inner
@@ -22,15 +23,17 @@ impl Pipeline {
     }
 
     /// Get the state
+    #[track_caller]
     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)?);
+        let (result, current, _pending) = self.inner.state(duration_to_clocktime(timeout)?);
         result.change_context(Error).attach("Failed to get state")?;
         Ok(current)
     }
 
+    #[track_caller]
     pub fn play(&self) -> Result<()> {
         self.inner
             .set_state(gstreamer::State::Playing)
@@ -39,6 +42,7 @@ impl Pipeline {
         Ok(())
     }
 
+    #[track_caller]
     pub fn pause(&self) -> Result<()> {
         self.inner
             .set_state(gstreamer::State::Paused)
@@ -47,6 +51,7 @@ impl Pipeline {
         Ok(())
     }
 
+    #[track_caller]
     pub fn ready(&self) -> Result<()> {
         self.inner
             .set_state(gstreamer::State::Ready)
@@ -55,6 +60,7 @@ impl Pipeline {
         Ok(())
     }
 
+    #[track_caller]
     pub fn set_state(&self, state: gstreamer::State) -> Result<gstreamer::StateChangeSuccess> {
         let result = self
             .inner
@@ -63,37 +69,135 @@ impl Pipeline {
             .attach("Failed to set pipeline state")?;
         Ok(result)
     }
-}
 
-pub trait PipelineExt {
-    fn bus(&self) -> Result<Bus>;
-    fn play(&self) -> Result<()>;
-    fn pause(&self) -> Result<()>;
-    fn ready(&self) -> Result<()>;
-    fn set_state(&self, state: gstreamer::State) -> Result<gstreamer::StateChangeSuccess>;
-    fn state(&self, timeout: impl Into<Option<core::time::Duration>>) -> Result<gstreamer::State>;
-}
-
-impl<T> PipelineExt for T
-where
-    T: ChildOf<Pipeline>,
-{
-    fn bus(&self) -> Result<Bus> {
-        self.upcast_ref().bus()
+    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 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 {}