1 files changed, 13 insertions, 11 deletions

diff --git a/generic/threadshare/src/proxy/imp.rs b/generic/threadshare/src/proxy/imp.rs
index 3d47e3665..06e9e7d61 100644
--- a/generic/threadshare/src/proxy/imp.rs
+++ b/generic/threadshare/src/proxy/imp.rs
@@ -858,6 +858,8 @@ impl ProxySrcTask {
 }
 
 impl TaskImpl for ProxySrcTask {
+    type Item = DataQueueItem;
+
     fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
         async move {
             gst::log!(SRC_CAT, obj: &self.element, "Starting task");
@@ -880,18 +882,18 @@ impl TaskImpl for ProxySrcTask {
         .boxed()
     }
 
-    fn iterate(&mut self) -> BoxFuture<'_, Result<(), gst::FlowError>> {
+    fn try_next(&mut self) -> BoxFuture<'_, Result<DataQueueItem, gst::FlowError>> {
         async move {
-            let item = self.dataqueue.next().await;
-
-            let item = match item {
-                Some(item) => item,
-                None => {
-                    gst::log!(SRC_CAT, obj: &self.element, "DataQueue Stopped");
-                    return Err(gst::FlowError::Flushing);
-                }
-            };
+            self.dataqueue
+                .next()
+                .await
+                .ok_or_else(|| panic!("DataQueue stopped while Task is Started"))
+        }
+        .boxed()
+    }
 
+    fn handle_item(&mut self, item: DataQueueItem) -> BoxFuture<'_, Result<(), gst::FlowError>> {
+        async move {
             let res = self.push_item(item).await;
             let proxysrc = self.element.imp();
             match res {
@@ -1085,7 +1087,7 @@ impl ProxySrc {
 
     fn pause(&self, element: &super::ProxySrc) -> Result<(), gst::ErrorMessage> {
         gst::debug!(SRC_CAT, obj: element, "Pausing");
-        let _ = self.task.pause().check()?;
+        self.task.pause().block_on()?;
         gst::debug!(SRC_CAT, obj: element, "Paused");
         Ok(())
     }