diff options
Diffstat (limited to 'generic/threadshare/src/proxy/imp.rs')
-rw-r--r-- | generic/threadshare/src/proxy/imp.rs | 24 |
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(()) } |