Add StreamExtDebounced

src/blocks/bluetooth.rs

@@ -260,11 +260,7 @@ impl DeviceMonitor {
 
                 loop {
                     select! {
-                        _ = updates.next() => {
-                            // avoid too frequent updates
-                            let _ = tokio::time::timeout(Duration::from_millis(100), async {
-                                loop { let _ = updates.next().await; }
-                            }).await;
+                        _ = updates.next_debounced() => {
                             debug!("Got update for device");
                             return Ok(());
                         }

src/blocks/external_ip.rs

@@ -180,12 +180,7 @@ pub async fn run(config: &Config, api: &CommonApi) -> Result<()> {
         select! {
             _ = sleep(config.interval.0) => (),
             _ = api.wait_for_update_request() => (),
-            _ = stream.next() => {
-                // avoid too frequent updates
-                let _ = tokio::time::timeout(Duration::from_millis(100), async {
-                    loop { let _ = stream.next().await; }
-                }).await;
-            }
+            _ = stream.next_debounced() => ()
         }
     }
 }

src/blocks/prelude.rs

@@ -29,3 +29,5 @@ pub use futures::{Stream, StreamExt};
 pub use smart_default::SmartDefault;
 
 pub use async_trait::async_trait;
+
+pub use crate::util::StreamExtDebounced as _;

src/blocks/privacy/v4l.rs

@@ -145,13 +145,7 @@ impl PrivacyMonitor for Monitor<'_> {
                         break;
                     }
                 },
-                _ = self.stream.next() => {
-                // avoid too frequent updates
-                let _ = tokio::time::timeout(Duration::from_millis(100), async {
-                    loop { let _ = self.stream.next().await; }
-                }).await;
-                break;
-            }
+                _ = self.stream.next_debounced() => break
             }
         }
         Ok(())

src/util.rs

@@ -252,6 +252,23 @@ pub fn default<T: Default>() -> T {
     Default::default()
 }
 
+pub trait StreamExtDebounced: futures::StreamExt {
+    fn next_debounced(&mut self) -> impl Future<Output = Option<Self::Item>>;
+}
+
+impl<T: futures::StreamExt + Unpin> StreamExtDebounced for T {
+    async fn next_debounced(&mut self) -> Option<Self::Item> {
+        let mut result = self.next().await?;
+        let mut noop_ctx = std::task::Context::from_waker(std::task::Waker::noop());
+        loop {
+            match self.poll_next_unpin(&mut noop_ctx) {
+                std::task::Poll::Ready(Some(x)) => result = x,
+                std::task::Poll::Ready(None) | std::task::Poll::Pending => return Some(result),
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;