clankerdiff_watch/file_watcher/
notify_file_watcher.rs

1use super::FileWatcher;
2use notify::{Event, RecommendedWatcher, RecursiveMode, Watcher};
3use std::{future::Future, path::PathBuf, time::Duration};
4use tokio::{spawn, sync::mpsc, task::JoinHandle, time::sleep};
5
6#[derive(Debug)]
7pub struct NotifyFileWatcher {
8    _watcher: Option<RecommendedWatcher>,
9    rx: mpsc::Receiver<()>,
10    task: JoinHandle<()>,
11}
12
13#[derive(Debug, PartialEq, Eq)]
14enum FilesChangedEvent {
15    Paths(Vec<PathBuf>),
16    Rescan,
17}
18
19#[derive(Debug, thiserror::Error)]
20pub enum FileWatchError {
21    #[error("could not create filesystem watcher")]
22    Create(#[source] notify::Error),
23    #[error("could not watch {path}")]
24    Watch {
25        path: PathBuf,
26        #[source]
27        source: notify::Error,
28    },
29}
30
31impl NotifyFileWatcher {
32    /// Starts a recursive file watcher with a caller-supplied filter.
33    pub fn new<T>(
34        roots: impl IntoIterator<Item = PathBuf>,
35        debounce: Duration,
36        filter: impl FnMut(Vec<PathBuf>) -> T + Send + 'static,
37    ) -> Result<Self, FileWatchError>
38    where
39        T: Future<Output = bool> + Send + 'static,
40    {
41        let (files_changed_tx, files_changed_rx) = mpsc::unbounded_channel();
42        let mut watcher = notify::recommended_watcher(move |event: notify::Result<Event>| {
43            if let Some(change) = match event {
44                Ok(e) if e.need_rescan() => Some(FilesChangedEvent::Rescan),
45                Ok(e) if e.kind.is_access() => None,
46                Ok(e) if e.paths.is_empty() => Some(FilesChangedEvent::Rescan),
47                Ok(e) => Some(FilesChangedEvent::Paths(e.paths)),
48                Err(_) => Some(FilesChangedEvent::Rescan),
49            } {
50                let _ = files_changed_tx.send(change);
51            }
52        })
53        .map_err(FileWatchError::Create)?;
54
55        for path in watch_roots(roots) {
56            watcher
57                .watch(&path, RecursiveMode::Recursive)
58                .map_err(|source| FileWatchError::Watch { path, source })?;
59        }
60
61        let (tx, rx) = mpsc::channel(1);
62        Ok(Self {
63            _watcher: Some(watcher),
64            rx,
65            task: spawn(process_events(tx, files_changed_rx, debounce, filter)),
66        })
67    }
68}
69
70impl FileWatcher for NotifyFileWatcher {
71    async fn recv(&mut self) -> Option<()> {
72        self.rx.recv().await
73    }
74}
75
76async fn process_events<T>(
77    tx: mpsc::Sender<()>,
78    mut rx: mpsc::UnboundedReceiver<FilesChangedEvent>,
79    debounce: Duration,
80    mut filter: impl FnMut(Vec<PathBuf>) -> T,
81) where
82    T: Future<Output = bool>,
83{
84    while let Some(event) = rx.recv().await {
85        let mut paths = Vec::new();
86        let mut rescan = false;
87        collect_event(event, &mut paths, &mut rescan);
88
89        let deadline = sleep(debounce);
90        tokio::pin!(deadline);
91        let mut ended = false;
92        loop {
93            tokio::select! {
94                biased;
95                () = &mut deadline => break,
96                change = rx.recv() => {
97                    if let Some(change) = change { collect_event(change, &mut paths, &mut rescan); }
98                    else { ended = true; break; }
99                },
100            }
101        }
102        paths.sort();
103        paths.dedup();
104        if rescan || (!paths.is_empty() && filter(paths).await) {
105            let _ = tx.try_send(());
106        }
107        if ended {
108            break;
109        }
110    }
111}
112
113fn collect_event(event: FilesChangedEvent, paths: &mut Vec<PathBuf>, rescan: &mut bool) {
114    match event {
115        FilesChangedEvent::Paths(batch) => paths.extend(batch),
116        FilesChangedEvent::Rescan => *rescan = true,
117    }
118}
119
120fn watch_roots(roots: impl IntoIterator<Item = PathBuf>) -> Vec<PathBuf> {
121    let mut roots: Vec<_> = roots.into_iter().collect();
122    roots.sort();
123    roots.dedup();
124    roots
125        .iter()
126        .filter(|path| {
127            !roots
128                .iter()
129                .any(|other| *path != other && path.starts_with(other))
130        })
131        .cloned()
132        .collect()
133}
134
135impl Drop for NotifyFileWatcher {
136    fn drop(&mut self) {
137        self.task.abort();
138    }
139}
clankerdiff_watch/file_watcher/notify_file_watcher.rs

clankerdiff_watch/file_watcher/
notify_file_watcher.rs
