use std::path::{Path, PathBuf}; use std::sync::mpsc; use std::thread; use std::time::Duration; use std::{collections::HashSet, io}; use crossbeam_channel::Receiver; use notify::{watcher, DebouncedEvent, RecommendedWatcher, RecursiveMode, Watcher}; use crate::{DirEntry, Metadata, ReadDir, VfsBackend, VfsEvent}; /// `VfsBackend` that uses `std::fs` and the `notify` crate. pub struct StdBackend { watcher: RecommendedWatcher, watcher_receiver: Receiver, watches: HashSet, } impl StdBackend { pub fn new() -> io::Result { let (notify_tx, notify_rx) = mpsc::channel(); let watcher = watcher(notify_tx, Duration::from_millis(50)).map_err(io::Error::other)?; let (tx, rx) = crossbeam_channel::unbounded(); thread::spawn(move || { for event in notify_rx { match event { DebouncedEvent::Create(path) => { tx.send(VfsEvent::Create(path))?; } DebouncedEvent::Write(path) => { tx.send(VfsEvent::Write(path))?; } DebouncedEvent::Remove(path) => { tx.send(VfsEvent::Remove(path))?; } DebouncedEvent::Rename(from, to) => { tx.send(VfsEvent::Remove(from))?; tx.send(VfsEvent::Create(to))?; } _ => {} } } Result::<(), crossbeam_channel::SendError>::Ok(()) }); Ok(Self { watcher, watcher_receiver: rx, watches: HashSet::new(), }) } } impl VfsBackend for StdBackend { fn read(&mut self, path: &Path) -> io::Result> { fs_err::read(path) } fn write(&mut self, path: &Path, data: &[u8]) -> io::Result<()> { fs_err::write(path, data) } fn exists(&mut self, path: &Path) -> io::Result { std::fs::exists(path) } fn read_dir(&mut self, path: &Path) -> io::Result { let entries: Result, _> = fs_err::read_dir(path)?.collect(); let mut entries = entries?; entries.sort_by_cached_key(|entry| entry.file_name()); let inner = entries .into_iter() .map(|entry| Ok(DirEntry { path: entry.path() })); Ok(ReadDir { inner: Box::new(inner), }) } fn create_dir(&mut self, path: &Path) -> io::Result<()> { fs_err::create_dir(path) } fn create_dir_all(&mut self, path: &Path) -> io::Result<()> { fs_err::create_dir_all(path) } fn remove_file(&mut self, path: &Path) -> io::Result<()> { fs_err::remove_file(path) } fn remove_dir_all(&mut self, path: &Path) -> io::Result<()> { fs_err::remove_dir_all(path) } fn metadata(&mut self, path: &Path) -> io::Result { let inner = fs_err::metadata(path)?; Ok(Metadata { is_file: inner.is_file(), }) } fn canonicalize(&mut self, path: &Path) -> io::Result { dunce::canonicalize(path) } fn event_receiver(&self) -> crossbeam_channel::Receiver { self.watcher_receiver.clone() } fn watch(&mut self, path: &Path) -> io::Result<()> { if self.watches.contains(path) || path .ancestors() .any(|ancestor| self.watches.contains(ancestor)) { Ok(()) } else { self.watches.insert(path.to_path_buf()); self.watcher .watch(path, RecursiveMode::Recursive) .map_err(io::Error::other) } } fn unwatch(&mut self, path: &Path) -> io::Result<()> { // `Vfs::commit_event` calls this for every removed path, but only the roots passed to // `watch` are watched. Forwarding any other path to notify made its FSEvents backend on // macOS restart the event stream, and changes made during the restart were lost. if self.watches.remove(path) { self.watcher.unwatch(path).map_err(io::Error::other) } else { Ok(()) } } } #[cfg(test)] mod test { use super::*; use std::fs; use std::time::Instant; /// `Vfs::commit_event` calls `unwatch` for every removed path, though only the roots passed /// to `watch` are watched. Forwarding those paths to notify made its macOS (FSEvents) backend /// restart its event stream each time, and events that happened during a restart were never /// delivered: deleting files while others were being created lost some of the creations. #[test] fn unwatching_paths_that_were_never_watched_loses_no_events() { let dir = tempfile::tempdir().unwrap(); let root = dunce::canonicalize(dir.path()).unwrap(); let (old, new) = (root.join("old"), root.join("new")); fs::create_dir(&old).unwrap(); fs::create_dir(&new).unwrap(); for i in 0..100 { fs::write(old.join(format!("{i}.txt")), "").unwrap(); } let mut backend = StdBackend::new().unwrap(); backend.watch(&root).unwrap(); thread::sleep(Duration::from_millis(500)); let events = backend.event_receiver(); while events.try_recv().is_ok() {} for i in 0..100 { let removed = old.join(format!("{i}.txt")); fs::remove_file(&removed).unwrap(); let _ = backend.unwatch(&removed); fs::write(new.join(format!("{i}.txt")), "").unwrap(); // Paced so the OS does not coalesce the burst into a rescan, which notify 4 drops on // its own; this test is about the events unwatch loses. thread::sleep(Duration::from_millis(5)); } let mut created = HashSet::new(); let deadline = Instant::now() + Duration::from_secs(5); while created.len() < 100 && Instant::now() < deadline { if let Ok(VfsEvent::Create(path) | VfsEvent::Write(path)) = events.recv_timeout(Duration::from_millis(100)) { if path.parent() == Some(new.as_path()) { created.insert(path); } } } assert_eq!(created.len(), 100, "file creations were not reported"); } }