pfx_load/scene/
watch.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::path::{Path, PathBuf};
3use std::sync::Arc;
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
6use std::time::{Duration, Instant};
7
8use notify::{EventKind, RecommendedWatcher, RecursiveMode, Watcher};
9
10use super::{Scene, SceneDiff, SceneError};
11use crate::sha256;
12
13pub const DEBOUNCE: Duration = Duration::from_millis(100);
14
15#[derive(Clone, Debug)]
16pub struct Reload {
17    pub scene: Scene,
18    pub diff: SceneDiff,
19    pub recovered: bool,
20}
21
22pub struct SceneWatch {
23    path: PathBuf,
24    scene: Scene,
25    watcher: RecommendedWatcher,
26    events: Receiver<Instant>,
27    folders: BTreeSet<PathBuf>,
28    hashes: BTreeMap<PathBuf, Option<[u8; 32]>>,
29    pending: Option<Instant>,
30    debounce: Duration,
31    shared_debounce: Arc<AtomicU64>,
32    broken: Option<SceneError>,
33}
34
35fn counts(event: &notify::Result<notify::Event>) -> bool {
36    matches!(event, Ok(event) if !matches!(event.kind, EventKind::Access(_)))
37}
38
39fn nanos(debounce: Duration) -> u64 {
40    u64::try_from(debounce.as_nanos()).unwrap_or(u64::MAX)
41}
42
43fn waker(
44    signals: Receiver<()>,
45    debounce: Arc<AtomicU64>,
46    wake: Box<dyn Fn() + Send>,
47) -> std::io::Result<()> {
48    std::thread::Builder::new()
49        .name("scene watch waker".into())
50        .spawn(move || {
51            while signals.recv().is_ok() {
52                loop {
53                    let quiet = Duration::from_nanos(debounce.load(Ordering::Relaxed));
54                    match signals.recv_timeout(quiet) {
55                        Ok(()) => {}
56                        Err(RecvTimeoutError::Timeout) => break,
57                        Err(RecvTimeoutError::Disconnected) => return,
58                    }
59                }
60                wake();
61            }
62        })
63        .map(|_| ())
64}
65
66fn folder(file: &Path) -> PathBuf {
67    match file.parent() {
68        Some(parent) if !parent.as_os_str().is_empty() => parent.to_path_buf(),
69        _ => PathBuf::from("."),
70    }
71}
72
73fn hashes(files: impl IntoIterator<Item = PathBuf>) -> BTreeMap<PathBuf, Option<[u8; 32]>> {
74    files
75        .into_iter()
76        .map(|file| {
77            let hash = std::fs::read(&file).ok().map(|bytes| sha256(&bytes));
78            (file, hash)
79        })
80        .collect()
81}
82
83impl SceneWatch {
84    pub fn open(path: impl AsRef<Path>) -> Result<Self, SceneError> {
85        Self::start(path.as_ref(), None)
86    }
87
88    pub fn open_with(
89        path: impl AsRef<Path>,
90        wake: impl Fn() + Send + 'static,
91    ) -> Result<Self, SceneError> {
92        Self::start(path.as_ref(), Some(Box::new(wake)))
93    }
94
95    fn start(path: &Path, wake: Option<Box<dyn Fn() + Send>>) -> Result<Self, SceneError> {
96        let path = path.to_path_buf();
97        let scene = Scene::open(&path)?;
98        let (sender, events) = channel();
99        let shared_debounce = Arc::new(AtomicU64::new(nanos(DEBOUNCE)));
100        let signals: Option<Sender<()>> = match wake {
101            Some(wake) => {
102                let (signal, signals) = channel();
103                waker(signals, shared_debounce.clone(), wake).map_err(|error| {
104                    SceneError::new(&path, None, format!("cannot start the waker: {error}"))
105                })?;
106                Some(signal)
107            }
108            None => None,
109        };
110        let watcher = notify::recommended_watcher(move |event| {
111            if counts(&event) {
112                let _ = sender.send(Instant::now());
113                if let Some(signal) = &signals {
114                    let _ = signal.send(());
115                }
116            }
117        })
118        .map_err(|error| SceneError::new(&path, None, format!("cannot watch: {error}")))?;
119        let mut watch = Self {
120            hashes: scene
121                .files
122                .iter()
123                .map(|(file, hash)| (file.clone(), Some(*hash)))
124                .collect(),
125            path,
126            scene,
127            watcher,
128            events,
129            folders: BTreeSet::new(),
130            pending: None,
131            debounce: DEBOUNCE,
132            shared_debounce,
133            broken: None,
134        };
135        watch.follow()?;
136        Ok(watch)
137    }
138
139    pub fn path(&self) -> &Path {
140        &self.path
141    }
142
143    pub fn scene(&self) -> &Scene {
144        &self.scene
145    }
146
147    pub fn error(&self) -> Option<&SceneError> {
148        self.broken.as_ref()
149    }
150
151    pub fn set_debounce(&mut self, debounce: Duration) {
152        self.debounce = debounce;
153        self.shared_debounce
154            .store(nanos(debounce), Ordering::Relaxed);
155    }
156
157    fn drain(&mut self) {
158        while let Ok(at) = self.events.try_recv() {
159            self.pending = Some(self.pending.map_or(at, |seen| seen.max(at)));
160        }
161    }
162
163    pub fn pending(&mut self) -> Option<Instant> {
164        self.drain();
165        self.pending.map(|seen| seen + self.debounce)
166    }
167
168    pub fn files(&self) -> impl Iterator<Item = &Path> {
169        self.hashes.keys().map(PathBuf::as_path)
170    }
171
172    fn follow(&mut self) -> Result<(), SceneError> {
173        let wanted: BTreeSet<PathBuf> = self.hashes.keys().map(|file| folder(file)).collect();
174        for gone in self.folders.difference(&wanted) {
175            let _ = self.watcher.unwatch(gone);
176        }
177        for new in wanted.difference(&self.folders) {
178            self.watcher
179                .watch(new, RecursiveMode::NonRecursive)
180                .map_err(|error| {
181                    SceneError::new(
182                        &self.path,
183                        None,
184                        format!("cannot watch {}: {error}", new.display()),
185                    )
186                })?;
187        }
188        self.folders = wanted;
189        Ok(())
190    }
191
192    pub fn poll(&mut self) -> Option<Result<Reload, SceneError>> {
193        self.drain();
194        let since = self.pending?;
195        if since.elapsed() < self.debounce {
196            return None;
197        }
198        self.pending = None;
199        self.check()
200    }
201
202    pub fn check(&mut self) -> Option<Result<Reload, SceneError>> {
203        let now = hashes(self.hashes.keys().cloned());
204        if now == self.hashes {
205            return None;
206        }
207        match Scene::open(&self.path) {
208            Ok(scene) => {
209                let diff = self.scene.diff(&scene);
210                let recovered = self.broken.take().is_some();
211                self.hashes = scene
212                    .files
213                    .iter()
214                    .map(|(file, hash)| (file.clone(), Some(*hash)))
215                    .collect();
216                self.scene = scene.clone();
217                if let Err(error) = self.follow() {
218                    return Some(Err(error));
219                }
220                if diff.is_empty() && !recovered {
221                    return None;
222                }
223                Some(Ok(Reload {
224                    scene,
225                    diff,
226                    recovered,
227                }))
228            }
229            Err(error) => {
230                self.hashes = now;
231                self.broken = Some(error.clone());
232                Some(Err(error))
233            }
234        }
235    }
236}
pfx_load/scene/watch.rs

pfx_load/scene/
watch.rs
