diff --git a/Cargo.lock b/Cargo.lock index f97344fc..df54568b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1911,7 +1911,6 @@ dependencies = [ "fs-err", "log", "notify", - "notify-debouncer-mini", "serde", "tempfile", ] @@ -1981,18 +1980,6 @@ dependencies = [ "windows-sys 0.60.2", ] -[[package]] -name = "notify-debouncer-mini" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17849edfaabd9a5fef1c606d99cfc615a8e99f7ac4366406d86c7942a3184cf2" -dependencies = [ - "log", - "notify", - "notify-types", - "tempfile", -] - [[package]] name = "notify-types" version = "2.1.0" diff --git a/crates/memofs/Cargo.toml b/crates/memofs/Cargo.toml index 7409e80f..34fdac8d 100644 --- a/crates/memofs/Cargo.toml +++ b/crates/memofs/Cargo.toml @@ -19,7 +19,6 @@ crossbeam-channel = "0.5.15" fs-err = "3" log = "0.4" notify = "8.2" -notify-debouncer-mini = "0.7" serde = { version = "1.0.228", features = ["derive"] } [dev-dependencies] diff --git a/crates/memofs/src/std_backend.rs b/crates/memofs/src/std_backend.rs index 0a669481..9c04899f 100644 --- a/crates/memofs/src/std_backend.rs +++ b/crates/memofs/src/std_backend.rs @@ -1,10 +1,11 @@ +use std::collections::HashMap; use std::path::{Path, PathBuf}; -use std::time::Duration; -use std::{collections::HashSet, io}; +use std::time::{Duration, Instant}; +use std::{collections::HashSet, io, thread}; -use crossbeam_channel::{Receiver, Sender}; -use notify::RecursiveMode; -use notify_debouncer_mini::{new_debouncer, DebounceEventResult, DebouncedEventKind, Debouncer}; +use crossbeam_channel::Receiver; +use notify::event::{CreateKind, EventKind, ModifyKind, RemoveKind, RenameMode}; +use notify::{RecursiveMode, Watcher}; use crate::{DirEntry, Metadata, ReadDir, VfsBackend, VfsEvent}; @@ -51,12 +52,10 @@ pub type CriticalErrorHandler = Box bool + Send /// `VfsBackend` that uses `std::fs` and the `notify` crate. pub struct StdBackend { - debouncer: Debouncer, + watcher: notify::RecommendedWatcher, watcher_receiver: Receiver, watches: HashSet, recursive_watches: HashSet, - /// Receiver for critical errors from the watcher thread. - /// Callers should poll this to detect when watching becomes unreliable. critical_error_receiver: Receiver, } @@ -103,41 +102,51 @@ impl StdBackend { pub fn new_with_error_handler(error_handler: CriticalErrorHandler) -> StdBackend { let (event_tx, event_rx) = crossbeam_channel::unbounded(); let (error_tx, error_rx) = crossbeam_channel::unbounded(); + let (raw_tx, raw_rx) = std::sync::mpsc::channel::>(); - let debouncer = Self::create_debouncer(event_tx, error_tx, error_handler); + let watcher = notify::RecommendedWatcher::new( + move |result| { + let _ = raw_tx.send(result); + }, + notify::Config::default(), + ) + .expect("Failed to create file watcher"); - Self { - debouncer, - watcher_receiver: event_rx, - watches: HashSet::new(), - recursive_watches: HashSet::new(), - critical_error_receiver: error_rx, - } - } + let debounce_ms = Duration::from_millis(50); - fn create_debouncer( - event_tx: Sender, - error_tx: Sender, - error_handler: CriticalErrorHandler, - ) -> Debouncer { - let debounce_timeout = Duration::from_millis(50); + thread::spawn(move || { + let mut pending: HashMap = HashMap::new(); + let mut last_event = Instant::now(); - new_debouncer( - debounce_timeout, - move |result: DebounceEventResult| match result { - Ok(events) => { - for event in events { - let vfs_event = match event.kind { - DebouncedEventKind::Any | DebouncedEventKind::AnyContinuous => { - if event.path.exists() { - VfsEvent::Write(event.path) - } else { - VfsEvent::Remove(event.path) + loop { + match raw_rx.recv_timeout(debounce_ms) { + Ok(Ok(event)) => { + last_event = Instant::now(); + for vfs_event in Self::convert_event(&event) { + let path = match &vfs_event { + VfsEvent::Create(p) | VfsEvent::Write(p) | VfsEvent::Remove(p) => { + p.clone() } - } - _ => continue, + }; + pending.insert(path, vfs_event); + } + } + Ok(Err(error)) => { + let critical_err = WatcherCriticalError::WatcherError { + error: format!("{:?}", error.kind), + path: error.paths.first().cloned(), }; - log::trace!("VFS debounced event: {:?}", &vfs_event); + let _ = error_tx.send(critical_err.clone()); + if error_handler(critical_err) { + return; + } + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => return, + } + + if !pending.is_empty() && last_event.elapsed() >= debounce_ms { + for (_, vfs_event) in pending.drain() { if let Err(err) = event_tx.send(vfs_event) { let critical_err = WatcherCriticalError::ChannelSendFailed(err.to_string()); @@ -148,17 +157,86 @@ impl StdBackend { } } } - Err(error) => { - let critical_err = WatcherCriticalError::WatcherError { - error: format!("{:?}", error.kind), - path: error.paths.first().cloned(), - }; - let _ = error_tx.send(critical_err.clone()); - error_handler(critical_err); + } + }); + + Self { + watcher, + watcher_receiver: event_rx, + watches: HashSet::new(), + recursive_watches: HashSet::new(), + critical_error_receiver: error_rx, + } + } + + fn convert_event(event: ¬ify::Event) -> Vec { + let mut vfs_events = Vec::new(); + + match &event.kind { + EventKind::Create(CreateKind::File) + | EventKind::Create(CreateKind::Folder) + | EventKind::Create(CreateKind::Any) + | EventKind::Create(CreateKind::Other) => { + for path in &event.paths { + vfs_events.push(VfsEvent::Create(path.clone())); } - }, - ) - .expect("Failed to create file watcher debouncer") + } + + EventKind::Modify(ModifyKind::Data(_)) + | EventKind::Modify(ModifyKind::Any) + | EventKind::Modify(ModifyKind::Other) + | EventKind::Modify(ModifyKind::Metadata(_)) => { + for path in &event.paths { + vfs_events.push(VfsEvent::Write(path.clone())); + } + } + + EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => { + if event.paths.len() >= 2 { + vfs_events.push(VfsEvent::Remove(event.paths[0].clone())); + vfs_events.push(VfsEvent::Create(event.paths[1].clone())); + } + } + EventKind::Modify(ModifyKind::Name(RenameMode::From)) => { + for path in &event.paths { + vfs_events.push(VfsEvent::Remove(path.clone())); + } + } + EventKind::Modify(ModifyKind::Name(RenameMode::To)) => { + for path in &event.paths { + vfs_events.push(VfsEvent::Create(path.clone())); + } + } + EventKind::Modify(ModifyKind::Name(RenameMode::Any)) + | EventKind::Modify(ModifyKind::Name(RenameMode::Other)) => { + for path in &event.paths { + vfs_events.push(VfsEvent::Write(path.clone())); + } + } + + EventKind::Remove(RemoveKind::File) + | EventKind::Remove(RemoveKind::Folder) + | EventKind::Remove(RemoveKind::Any) + | EventKind::Remove(RemoveKind::Other) => { + for path in &event.paths { + vfs_events.push(VfsEvent::Remove(path.clone())); + } + } + + EventKind::Access(_) => {} + + EventKind::Other | EventKind::Any => { + for path in &event.paths { + vfs_events.push(VfsEvent::Write(path.clone())); + } + } + } + + for event in &vfs_events { + log::trace!("VFS event: {:?}", event); + } + + vfs_events } /// Returns a receiver for critical errors from the watcher thread. @@ -255,7 +333,7 @@ impl VfsBackend for StdBackend { } else { RecursiveMode::NonRecursive }; - match self.debouncer.watcher().watch(path, mode) { + match self.watcher.watch(path, mode) { Ok(()) => { log::debug!( "Watching path ({}): {}", @@ -282,7 +360,7 @@ impl VfsBackend for StdBackend { fn unwatch(&mut self, path: &Path) -> io::Result<()> { let was_watched = self.watches.contains(path); - match self.debouncer.watcher().unwatch(path) { + match self.watcher.unwatch(path) { Ok(()) => { if was_watched { log::info!("Unwatched path: {}", path.display());