| 1 | //! `general.watch_dir`: watch the queue's directories and report files that appear, change or |
| 2 | //! vanish, so the queue follows the disk live. |
| 3 | |
| 4 | use std::collections::HashSet; |
| 5 | use std::path::{Path, PathBuf}; |
| 6 | use std::time::{Duration, SystemTime}; |
| 7 | |
| 8 | use crossbeam_channel::{Receiver, RecvTimeoutError, Sender}; |
| 9 | use notify::event::{AccessKind, AccessMode}; |
| 10 | use notify::{EventKind, RecommendedWatcher, RecursiveMode, Watcher}; |
| 11 | |
| 12 | use crate::decode; |
| 13 | |
| 14 | /// Quiet time before reporting, so a file being written arrives once, complete. |
| 15 | const DEBOUNCE: Duration = Duration::from_millis(300); |
| 16 | |
| 17 | /// What changed in the watched directories. |
| 18 | #[derive(Debug, Default)] |
| 19 | pub struct Changes { |
| 20 | /// Image files that exist now, with their mtime: new ones join the queue, known ones |
| 21 | /// reload if their mtime moved. |
| 22 | pub present: Vec<(PathBuf, SystemTime)>, |
| 23 | /// Paths that no longer exist. |
| 24 | pub gone: Vec<PathBuf>, |
| 25 | } |
| 26 | |
| 27 | enum Msg { |
| 28 | Fs(notify::Event), |
| 29 | Dirs(Vec<(PathBuf, bool)>), |
| 30 | } |
| 31 | |
| 32 | /// Handle to the watch thread. |
| 33 | pub struct DirWatch { |
| 34 | tx: Sender<Msg>, |
| 35 | } |
| 36 | |
| 37 | impl DirWatch { |
| 38 | /// Watch these directories instead (after `open`). `true` = recursively. |
| 39 | pub fn set(&self, dirs: Vec<(PathBuf, bool)>) { |
| 40 | let _ = self.tx.send(Msg::Dirs(dirs)); |
| 41 | } |
| 42 | } |
| 43 | |
| 44 | pub fn spawn(dirs: Vec<(PathBuf, bool)>, deliver: impl Fn(Changes) + Send + 'static) -> DirWatch { |
| 45 | let (tx, rx) = crossbeam_channel::unbounded(); |
| 46 | let fs_tx = tx.clone(); |
| 47 | let _ = tx.send(Msg::Dirs(dirs)); |
| 48 | std::thread::Builder::new() |
| 49 | .name("remage-watch".into()) |
| 50 | .spawn(move || { |
| 51 | let watcher = notify::recommended_watcher(move |res: notify::Result<notify::Event>| { |
| 52 | if let Ok(ev) = res { |
| 53 | let _ = fs_tx.send(Msg::Fs(ev)); |
| 54 | } |
| 55 | }); |
| 56 | match watcher { |
| 57 | Ok(w) => run(w, &rx, deliver), |
| 58 | Err(e) => log::warn!("directory watcher: {e}"), |
| 59 | } |
| 60 | }) |
| 61 | .ok(); |
| 62 | DirWatch { tx } |
| 63 | } |
| 64 | |
| 65 | fn run(mut w: RecommendedWatcher, rx: &Receiver<Msg>, deliver: impl Fn(Changes)) { |
| 66 | let mut watched: Vec<(PathBuf, bool)> = Vec::new(); |
| 67 | let mut touched: HashSet<PathBuf> = HashSet::new(); |
| 68 | loop { |
| 69 | let msg = if touched.is_empty() { |
| 70 | rx.recv().map_err(|_| RecvTimeoutError::Disconnected) |
| 71 | } else { |
| 72 | rx.recv_timeout(DEBOUNCE) |
| 73 | }; |
| 74 | match msg { |
| 75 | Ok(Msg::Dirs(dirs)) => { |
| 76 | for (d, _) in &watched { |
| 77 | let _ = w.unwatch(d); |
| 78 | } |
| 79 | for (d, rec) in &dirs { |
| 80 | let mode = if *rec { |
| 81 | RecursiveMode::Recursive |
| 82 | } else { |
| 83 | RecursiveMode::NonRecursive |
| 84 | }; |
| 85 | if let Err(e) = w.watch(d, mode) { |
| 86 | log::warn!("watch {}: {e}", d.display()); |
| 87 | } |
| 88 | } |
| 89 | log::debug!("watching {} directories", dirs.len()); |
| 90 | watched = dirs; |
| 91 | touched.clear(); |
| 92 | } |
| 93 | Ok(Msg::Fs(ev)) => { |
| 94 | if relevant(&ev.kind) { |
| 95 | touched.extend(ev.paths.into_iter().filter(|p| visible(p))); |
| 96 | } |
| 97 | } |
| 98 | Err(RecvTimeoutError::Timeout) => { |
| 99 | let changes = classify(touched.drain()); |
| 100 | if !changes.present.is_empty() || !changes.gone.is_empty() { |
| 101 | deliver(changes); |
| 102 | } |
| 103 | } |
| 104 | Err(RecvTimeoutError::Disconnected) => return, |
| 105 | } |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | fn relevant(kind: &EventKind) -> bool { |
| 110 | match kind { |
| 111 | EventKind::Access(AccessKind::Close(AccessMode::Write)) => true, |
| 112 | EventKind::Access(_) => false, |
| 113 | _ => true, |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | /// Hidden files stay out, as in the initial scan. |
| 118 | fn visible(p: &Path) -> bool { |
| 119 | p.file_name() |
| 120 | .is_some_and(|n| !n.as_encoded_bytes().starts_with(b".")) |
| 121 | } |
| 122 | |
| 123 | fn classify(paths: impl Iterator<Item = PathBuf>) -> Changes { |
| 124 | let mut c = Changes::default(); |
| 125 | for p in paths { |
| 126 | match std::fs::metadata(&p) { |
| 127 | Ok(m) if m.is_file() => { |
| 128 | if decode::is_known_extension(&p) || decode::sniff(&p) { |
| 129 | let t = m.modified().unwrap_or(SystemTime::UNIX_EPOCH); |
| 130 | c.present.push((p, t)); |
| 131 | } |
| 132 | } |
| 133 | Ok(_) => {} |
| 134 | Err(_) => c.gone.push(p), |
| 135 | } |
| 136 | } |
| 137 | c |
| 138 | } |
| 139 | |
| 140 | #[cfg(test)] |
| 141 | mod tests { |
| 142 | use super::*; |
| 143 | |
| 144 | #[test] |
| 145 | fn reports_new_and_removed_files() { |
| 146 | let dir = std::env::temp_dir().join(format!("remage-watch-{}", std::process::id())); |
| 147 | std::fs::create_dir_all(&dir).unwrap(); |
| 148 | let (tx, rx) = crossbeam_channel::unbounded(); |
| 149 | let _w = spawn(vec![(dir.clone(), false)], move |c| { |
| 150 | let _ = tx.send(c); |
| 151 | }); |
| 152 | std::thread::sleep(Duration::from_millis(200)); |
| 153 | std::fs::write(dir.join("a.png"), b"x").unwrap(); |
| 154 | std::fs::write(dir.join(".hidden.png"), b"x").unwrap(); |
| 155 | std::fs::write(dir.join("notes.txt"), b"x").unwrap(); |
| 156 | let c = rx.recv_timeout(Duration::from_secs(5)).unwrap(); |
| 157 | let names: Vec<_> = c |
| 158 | .present |
| 159 | .iter() |
| 160 | .map(|(p, _)| p.file_name().unwrap()) |
| 161 | .collect(); |
| 162 | assert_eq!(names, ["a.png"]); |
| 163 | std::fs::remove_file(dir.join("a.png")).unwrap(); |
| 164 | let c = rx.recv_timeout(Duration::from_secs(5)).unwrap(); |
| 165 | assert_eq!(c.gone, [dir.join("a.png")]); |
| 166 | std::fs::remove_dir_all(&dir).unwrap(); |
| 167 | } |
| 168 | } |