clankerdiff_watch/file_watcher/
notify_file_watcher.rs

1use super::FileWatcher;
2use ignore::WalkBuilder;
3use notify::{
4    Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher,
5    event::{Flag, ModifyKind},
6};
7use std::{
8    collections::BTreeSet,
9    future::Future,
10    path::{Path, PathBuf},
11    time::Duration,
12};
13use tokio::{
14    spawn,
15    sync::{broadcast, broadcast::error::RecvError, mpsc},
16    task::{JoinHandle, spawn_blocking},
17    time::sleep,
18};
19
20const MAX_EVENTS: usize = 256;
21const MAX_PATHS: usize = 1024;
22
23#[derive(Debug)]
24pub struct NotifyFileWatcher {
25    rx: mpsc::Receiver<()>,
26    task: JoinHandle<()>,
27}
28
29#[derive(Debug, thiserror::Error)]
30pub enum FileWatchError {
31    #[error("could not create filesystem watcher")]
32    Create(#[source] notify::Error),
33    #[error("could not watch {path}")]
34    Watch {
35        path: PathBuf,
36        #[source]
37        source: notify::Error,
38    },
39    #[error(transparent)]
40    Walk(#[from] ignore::Error),
41}
42
43pub(crate) fn worktree_walk(root: impl AsRef<Path>) -> WalkBuilder {
44    let mut walk = WalkBuilder::new(root);
45    walk.hidden(false).ignore(false);
46    walk
47}
48
49impl NotifyFileWatcher {
50    pub fn new<T>(
51        roots: impl IntoIterator<Item = PathBuf>,
52        debounce: Duration,
53        filter: impl FnMut(Vec<PathBuf>) -> T + Send + 'static,
54    ) -> Result<Self, FileWatchError>
55    where
56        T: Future<Output = bool> + Send + 'static,
57    {
58        let walks = roots.into_iter().map(worktree_walk).collect();
59        Self::with_walks(walks, debounce, filter)
60    }
61
62    pub(crate) fn with_walks<T>(
63        walks: Vec<WalkBuilder>,
64        debounce: Duration,
65        filter: impl FnMut(Vec<PathBuf>) -> T + Send + 'static,
66    ) -> Result<Self, FileWatchError>
67    where
68        T: Future<Output = bool> + Send + 'static,
69    {
70        let (events_tx, events_rx) = broadcast::channel(MAX_EVENTS);
71        let watches = Watches::new(walks, events_tx)?;
72        let (tx, rx) = mpsc::channel(1);
73        Ok(Self {
74            rx,
75            task: spawn(process_events(tx, events_rx, watches, debounce, filter)),
76        })
77    }
78}
79
80impl FileWatcher for NotifyFileWatcher {
81    async fn recv(&mut self) -> Option<()> {
82        self.rx.recv().await
83    }
84}
85
86impl Drop for NotifyFileWatcher {
87    fn drop(&mut self) {
88        self.task.abort();
89    }
90}
91
92struct Watches {
93    walks: Vec<WalkBuilder>,
94    events_tx: broadcast::Sender<Event>,
95    watcher: RecommendedWatcher,
96}
97
98impl Watches {
99    fn new(
100        walks: Vec<WalkBuilder>,
101        events_tx: broadcast::Sender<Event>,
102    ) -> Result<Self, FileWatchError> {
103        let watcher = watch_all(&walks, &events_tx)?;
104        Ok(Self {
105            walks,
106            events_tx,
107            watcher,
108        })
109    }
110
111    fn rebuild(mut self) -> Result<Self, FileWatchError> {
112        self.watcher = watch_all(&self.walks, &self.events_tx)?;
113        Ok(self)
114    }
115}
116
117fn watch_all(
118    walks: &[WalkBuilder],
119    events_tx: &broadcast::Sender<Event>,
120) -> Result<RecommendedWatcher, FileWatchError> {
121    let events_tx = events_tx.clone();
122    let mut watcher = notify::recommended_watcher(move |event: notify::Result<Event>| {
123        let event = event.unwrap_or_else(|_| Event::new(EventKind::Any).set_flag(Flag::Rescan));
124        let _ = events_tx.send(event);
125    })
126    .map_err(FileWatchError::Create)?;
127    for entry in walks.iter().flat_map(WalkBuilder::build) {
128        let entry = entry?;
129        if entry.depth() == 0 || entry.file_type().is_some_and(|kind| kind.is_dir()) {
130            let path = entry.into_path();
131            watcher
132                .watch(&path, RecursiveMode::NonRecursive)
133                .map_err(|source| FileWatchError::Watch { path, source })?;
134        }
135    }
136    Ok(watcher)
137}
138
139#[derive(Default)]
140struct Batch {
141    paths: BTreeSet<PathBuf>,
142    rescan: bool,
143    rewatch: bool,
144}
145
146impl Batch {
147    fn push(&mut self, received: Result<Event, RecvError>) {
148        let Ok(event) = received else {
149            self.rescan = true;
150            return;
151        };
152        if self.rescan || event.kind.is_access() {
153            return;
154        }
155        if event.need_rescan() || event.paths.is_empty() {
156            self.rescan = true;
157            return;
158        }
159        self.rewatch |= needs_rewatch(&event);
160        self.paths.extend(event.paths);
161        self.rescan = self.paths.len() > MAX_PATHS;
162    }
163}
164
165fn needs_rewatch(event: &Event) -> bool {
166    !event.kind.is_modify()
167        || matches!(event.kind, EventKind::Modify(ModifyKind::Name(_)))
168        || event.paths.iter().any(|path| {
169            path.file_name()
170                .is_some_and(|name| name == ".gitignore" || name == "exclude" || name == "config")
171        })
172}
173
174async fn process_events<T>(
175    tx: mpsc::Sender<()>,
176    mut events_rx: broadcast::Receiver<Event>,
177    mut watches: Watches,
178    debounce: Duration,
179    mut filter: impl FnMut(Vec<PathBuf>) -> T,
180) where
181    T: Future<Output = bool>,
182{
183    loop {
184        let mut batch = Batch::default();
185        batch.push(events_rx.recv().await);
186        let deadline = sleep(debounce);
187        tokio::pin!(deadline);
188        loop {
189            tokio::select! {
190                biased;
191                () = &mut deadline => break,
192                received = events_rx.recv() => batch.push(received),
193            }
194        }
195        if batch.rescan || batch.rewatch {
196            watches = match spawn_blocking(move || watches.rebuild()).await {
197                Ok(Ok(watches)) => watches,
198                _ => return,
199            };
200        }
201        if batch.rescan
202            || (!batch.paths.is_empty() && filter(batch.paths.into_iter().collect()).await)
203        {
204            let _ = tx.try_send(());
205        }
206    }
207}
clankerdiff_watch/file_watcher/notify_file_watcher.rs

clankerdiff_watch/file_watcher/
notify_file_watcher.rs
