diff --git a/src-tauri/src/core/notify/announce.rs b/src-tauri/src/core/notify/announce.rs index 785b8b9c7..b6fb11234 100644 --- a/src-tauri/src/core/notify/announce.rs +++ b/src-tauri/src/core/notify/announce.rs @@ -1,6 +1,7 @@ use super::{handle::Handle, notification::FrontendEvent}; use anyhow::Result; -use std::{cell::RefCell, future::Future}; +use parking_lot::Mutex; +use std::{cell::RefCell, future::Future, time::Duration}; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Refresh { @@ -14,9 +15,35 @@ tokio::task_local! { static PENDING: RefCell>; } +const REFRESH_WINDOW: Duration = Duration::from_millis(20); +static BATCH: RefreshBatch = RefreshBatch(Mutex::new(Vec::new())); + +#[derive(Default)] +struct RefreshBatch(Mutex>); + +impl RefreshBatch { + fn enqueue(&self, refresh: Refresh) -> bool { + let mut pending = self.0.lock(); + let start_window = pending.is_empty(); + if !pending.contains(&refresh) { + pending.push(refresh); + } + start_window + } + + async fn flush(&self, emit: impl Fn(Refresh)) { + // A fixed window bounds latency even under a continuous stream of refreshes. + tokio::time::sleep(REFRESH_WINDOW).await; + let refreshes = std::mem::take(&mut *self.0.lock()); + for refresh in refreshes { + emit(refresh); + } + } +} + pub fn announce(refresh: Refresh) { - if PENDING.try_with(|pending| pending.borrow_mut().push(refresh)).is_err() { - emit(refresh); + if PENDING.try_with(|pending| pending.borrow_mut().push(refresh)).is_err() && BATCH.enqueue(refresh) { + crate::AsyncHandler::spawn(|| BATCH.flush(emit)); } } @@ -74,4 +101,39 @@ mod tests { assert_eq!(refreshes, [Refresh::Clash, Refresh::Verge, Refresh::Clash]); Ok(()) } + + #[tokio::test] + async fn bursts_keep_each_refresh_kind_and_rearm_during_delivery() -> Result<()> { + let batch = RefreshBatch::default(); + let delivered = Mutex::new(Vec::new()); + assert!(batch.enqueue(Refresh::Clash)); + for refresh in [Refresh::Verge, Refresh::Clash, Refresh::Profiles, Refresh::Proxies] { + assert!(!batch.enqueue(refresh)); + } + tokio::time::timeout(Duration::from_secs(1), async { + batch + .flush(|refresh| { + delivered.lock().push(refresh); + if refresh == Refresh::Clash { + assert!(batch.enqueue(Refresh::Clash)); + assert!(!batch.enqueue(Refresh::Clash)); + } + }) + .await; + batch.flush(|refresh| delivered.lock().push(refresh)).await; + }) + .await?; + assert_eq!( + *delivered.lock(), + [ + Refresh::Clash, + Refresh::Verge, + Refresh::Profiles, + Refresh::Proxies, + Refresh::Clash + ] + ); + assert!(batch.enqueue(Refresh::Verge)); + Ok(()) + } }