domi_server/serve/
watcher.rs

1use std::collections::VecDeque;
2use std::io;
3use std::path::{Path, PathBuf};
4use std::sync::mpsc::{channel, RecvTimeoutError};
5
6use notify::{Event as NotifyEvent, EventKind, RecommendedWatcher, RecursiveMode, Watcher as NotifyWatcherTrait};
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct WatchEvent {
10    pub kind: WatchEventKind,
11    pub paths: Vec<PathBuf>,
12}
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum WatchEventKind {
16    Created,
17    Modified,
18    Removed,
19    Any,
20}
21
22pub trait Watcher: Send {
23    fn next_event(&mut self, timeout_ms: u32) -> io::Result<Option<WatchEvent>>;
24}
25
26pub struct MockWatcher {
27    evs: VecDeque<WatchEvent>,
28}
29
30impl MockWatcher {
31    pub fn new() -> Self {
32        Self { evs: VecDeque::new() }
33    }
34
35    pub fn push(&mut self, ev: WatchEvent) {
36        self.evs.push_back(ev);
37    }
38}
39
40impl Watcher for MockWatcher {
41    fn next_event(&mut self, _timeout_ms: u32) -> io::Result<Option<WatchEvent>> {
42        Ok(self.evs.pop_front())
43    }
44}
45
46pub struct NotifyWatcher {
47    _impl: RecommendedWatcher,
48    rx: std::sync::mpsc::Receiver<io::Result<NotifyEvent>>,
49    root: PathBuf,
50}
51
52impl NotifyWatcher {
53    pub fn new(root: &Path, coalesce_ms: u32) -> io::Result<Self> {
54        let (tx, rx) = channel();
55        let mut impl_ = RecommendedWatcher::new(
56            move |res: notify::Result<NotifyEvent>| {
57                let _ = tx.send(res.map_err(|e| io::Error::other(e.to_string())));
58            },
59            notify::Config::default()
60                .with_poll_interval(std::time::Duration::from_millis(coalesce_ms.max(10) as u64)),
61        )
62        .map_err(|e| io::Error::other(e.to_string()))?;
63        impl_
64            .watch(root, RecursiveMode::Recursive)
65            .map_err(|e| io::Error::other(e.to_string()))?;
66        Ok(Self {
67            _impl: impl_,
68            rx,
69            root: root.to_path_buf(),
70        })
71    }
72
73    pub fn root(&self) -> &Path {
74        &self.root
75    }
76}
77
78fn map_kind(k: EventKind) -> WatchEventKind {
79    match k {
80        EventKind::Create(_) => WatchEventKind::Created,
81        EventKind::Modify(_) => WatchEventKind::Modified,
82        EventKind::Remove(_) => WatchEventKind::Removed,
83        _ => WatchEventKind::Any,
84    }
85}
86
87impl Watcher for NotifyWatcher {
88    fn next_event(&mut self, timeout_ms: u32) -> io::Result<Option<WatchEvent>> {
89        let recv = self
90            .rx
91            .recv_timeout(std::time::Duration::from_millis(timeout_ms as u64));
92        match recv {
93            Ok(Ok(ev)) => Ok(Some(WatchEvent {
94                kind: map_kind(ev.kind),
95                paths: ev.paths,
96            })),
97            Ok(Err(e)) => Err(e),
98            Err(RecvTimeoutError::Timeout) => Ok(None),
99            Err(RecvTimeoutError::Disconnected) => Ok(None),
100        }
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107
108    #[test]
109    fn mock_yields_pushed_events_in_order() {
110        let mut w = MockWatcher::new();
111        let ev_a = WatchEvent { kind: WatchEventKind::Created, paths: vec![PathBuf::from("/a")] };
112        let ev_b = WatchEvent { kind: WatchEventKind::Removed, paths: vec![PathBuf::from("/b")] };
113        w.push(ev_a.clone());
114        w.push(ev_b.clone());
115        assert_eq!(w.next_event(0).unwrap(), Some(ev_a));
116        assert_eq!(w.next_event(0).unwrap(), Some(ev_b));
117        assert_eq!(w.next_event(0).unwrap(), None);
118    }
119
120    #[test]
121    fn mock_returns_none_when_empty_regardless_of_timeout() {
122        let mut w = MockWatcher::new();
123        assert_eq!(w.next_event(0).unwrap(), None);
124        assert_eq!(w.next_event(1000).unwrap(), None);
125    }
126
127    #[test]
128    fn watch_event_partial_eq() {
129        let ev1 = WatchEvent { kind: WatchEventKind::Modified, paths: vec![PathBuf::from("/x")] };
130        let ev2 = WatchEvent { kind: WatchEventKind::Modified, paths: vec![PathBuf::from("/x")] };
131        assert_eq!(ev1, ev2);
132    }
133
134    #[test]
135    #[ignore = "flaky in CI on macOS FSEvents; run with --ignored to verify manually"]
136    fn notify_watcher_emits_event_on_create() {
137        let dir = tempfile::tempdir().unwrap();
138        let root = dir.path();
139        let mut w = NotifyWatcher::new(root, 50).expect("watcher created");
140        std::thread::sleep(std::time::Duration::from_millis(50));
141        std::fs::write(root.join("new.html"), "<h1>x</h1>").unwrap();
142        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
143        let mut got = None;
144        while std::time::Instant::now() < deadline {
145            match w.next_event(100) {
146                Ok(Some(ev)) => {
147                    if ev.paths.iter().any(|p| p.ends_with("new.html")) {
148                        got = Some(ev);
149                        break;
150                    }
151                    // FSEvents on macOS emits a synthetic root-directory event first;
152                    // keep draining until we see the file we wrote.
153                }
154                Ok(None) => continue,
155                Err(e) => panic!("watcher errored: {e}"),
156            }
157        }
158        let ev = got.expect("watcher emitted a new.html event within 2s");
159        assert!(ev.paths.iter().any(|p| p.ends_with("new.html")));
160    }
161}
domi_server/serve/watcher.rs

domi_server/serve/
watcher.rs
