Remove debounced dependency (#2138)

Author: bim9262

Cargo.lock

@@ -36,7 +36,6 @@ calibright = { version = "0.1.9", features = ["watch"] }
 chrono = { version = "0.4", default-features = false, features = ["clock", "unstable-locales"] }
 chrono-tz = { version = "0.10", features = ["serde"] }
 clap = { version = "4.0", default-features = false, features = ["std", "derive", "help", "usage"] }
-debounced = "0.2.0"
 dirs = "5.0"
 env_logger = "0.11"
 futures = { version = "0.3.31", default-features = false }

Cargo.toml

@@ -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/bluetooth.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/external_ip.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/prelude.rs

@@ -1,4 +1,3 @@
-use debounced::{Debounced, debounced};
 use inotify::{EventStream, Inotify, WatchDescriptor, WatchMask, Watches};
 use tokio::fs::{File, read_dir};
 use tokio::time::{Interval, interval};
@@ -20,27 +19,24 @@ pub(super) struct Monitor<'a> {
     devices: HashMap<PathBuf, WatchDescriptor>,
     interval: Interval,
     watches: Watches,
-    updates: Debounced<EventStream<[u8; 1024]>>,
+    stream: EventStream<[u8; 1024]>,
 }
 
 impl<'a> Monitor<'a> {
     pub(super) async fn new(config: &'a Config, duration: Duration) -> Result<Self> {
         let notify = Inotify::init().error("Failed to start inotify")?;
         let watches = notify.watches();
 
-        let updates = debounced(
-            notify
-                .into_event_stream([0; 1024])
-                .error("Failed to create event stream")?,
-            Duration::from_millis(100),
-        );
+        let stream = notify
+            .into_event_stream([0; 1024])
+            .error("Failed to create event stream")?;
 
         let mut s = Self {
             config,
             devices: HashMap::new(),
             interval: interval(duration),
             watches,
-            updates,
+            stream,
         };
         s.update_devices().await?;
 
@@ -149,7 +145,7 @@ impl PrivacyMonitor for Monitor<'_> {
                         break;
                     }
                 },
-                _ = self.updates.next() => break,
+                _ = self.stream.next_debounced() => break
             }
         }
         Ok(())

src/blocks/privacy/v4l.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::*;