From c01f7072fbb8151489fd7b15857bc67a0b03c95b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sebastian=20Dr=C3=B6ge?= <sebastian@centricular.com>
Date: Thu, 1 Oct 2020 10:13:52 +0300
Subject: [PATCH] gstreamer/clock: Add ClockID::wait_async_stream()

This provides an async stream with the values of each timeout.
---
 gstreamer/src/clock.rs | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/gstreamer/src/clock.rs b/gstreamer/src/clock.rs
index c643a5284..9182ba202 100644
--- a/gstreamer/src/clock.rs
+++ b/gstreamer/src/clock.rs
@@ -25,6 +25,9 @@ use ClockSuccess;
 use ClockTime;
 use ClockTimeDiff;
 
+use futures_core::Stream;
+use std::marker::Unpin;
+use std::pin::Pin;
 use std::sync::atomic;
 use std::sync::atomic::AtomicI32;
 
@@ -102,6 +105,24 @@ impl ClockId {
         ret.into_result()
     }
 
+    #[allow(clippy::type_complexity)]
+    pub fn wait_async_stream(
+        &self,
+    ) -> Result<
+        Pin<Box<dyn Stream<Item = (ClockTime, ClockId)> + Unpin + Send + 'static>>,
+        ClockError,
+    > {
+        use futures_channel::mpsc;
+
+        let (sender, receiver) = mpsc::unbounded();
+
+        self.wait_async(move |_clock, jitter, id| {
+            let _ = sender.unbounded_send((jitter, id.clone()));
+        })?;
+
+        Ok(Box::pin(receiver))
+    }
+
     pub fn compare_by_time(&self, other: &Self) -> cmp::Ordering {
         unsafe {
             let res =