diff --git a/Cargo.lock b/Cargo.lock index 27af675174..0d4f5e252a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2042,6 +2042,7 @@ dependencies = [ "percent-encoding", "regex", "rustyline", + "same-file", "serde", "signal-hook", "signal-hook-registry", diff --git a/runtime/Cargo.toml b/runtime/Cargo.toml index 14580a8a64..fbf5d95811 100644 --- a/runtime/Cargo.toml +++ b/runtime/Cargo.toml @@ -117,6 +117,7 @@ once_cell.workspace = true percent-encoding.workspace = true regex.workspace = true rustyline = { workspace = true, features = ["custom-bindings"] } +same-file = "1.0.6" serde.workspace = true signal-hook = "0.3.17" signal-hook-registry = "1.4.0" diff --git a/runtime/js/40_fs_events.js b/runtime/js/40_fs_events.js index ec2474c0ad..322ee6b3ca 100644 --- a/runtime/js/40_fs_events.js +++ b/runtime/js/40_fs_events.js @@ -21,7 +21,7 @@ class FsWatcher { constructor(paths, options) { const { recursive } = options; - this.#rid = op_fs_events_open({ recursive, paths }); + this.#rid = op_fs_events_open(recursive, paths); } unref() { diff --git a/runtime/ops/fs_events.rs b/runtime/ops/fs_events.rs index 89a0606dbb..648553376a 100644 --- a/runtime/ops/fs_events.rs +++ b/runtime/ops/fs_events.rs @@ -19,13 +19,14 @@ use notify::EventKind; use notify::RecommendedWatcher; use notify::RecursiveMode; use notify::Watcher; -use serde::Deserialize; use serde::Serialize; use std::borrow::Cow; use std::cell::RefCell; use std::convert::From; +use std::path::Path; use std::path::PathBuf; use std::rc::Rc; +use std::sync::Arc; use tokio::sync::mpsc; deno_core::extension!( @@ -34,8 +35,6 @@ deno_core::extension!( ); struct FsEventsResource { - #[allow(unused)] - watcher: RecommendedWatcher, receiver: AsyncRefCell>>, cancel: CancelHandle, } @@ -58,7 +57,7 @@ impl Resource for FsEventsResource { /// /// Feel free to expand this struct as long as you can add tests to demonstrate /// the complexity. -#[derive(Serialize, Debug)] +#[derive(Serialize, Debug, Clone)] struct FsEvent { kind: &'static str, paths: Vec, @@ -92,6 +91,24 @@ impl From for FsEvent { } } +type WatchSender = (Vec, mpsc::Sender>); + +struct WatcherState { + senders: Arc>>, + watcher: RecommendedWatcher, +} + +fn starts_with_canonicalized(path: &Path, prefix: &str) -> bool { + #[allow(clippy::disallowed_methods)] + let path = path.canonicalize().ok(); + #[allow(clippy::disallowed_methods)] + let prefix = std::fs::canonicalize(prefix).ok(); + match (path, prefix) { + (Some(path), Some(prefix)) => path.starts_with(prefix), + _ => false, + } +} + #[derive(Debug, thiserror::Error)] pub enum FsEventsError { #[error(transparent)] @@ -104,44 +121,73 @@ pub enum FsEventsError { Canceled(#[from] deno_core::Canceled), } -#[derive(Deserialize)] -pub struct OpenArgs { - recursive: bool, +fn start_watcher( + state: &mut OpState, paths: Vec, + sender: mpsc::Sender>, +) -> Result<(), FsEventsError> { + if let Some(watcher) = state.try_borrow_mut::() { + watcher.senders.lock().push((paths, sender)); + return Ok(()); + } + + let senders = Arc::new(Mutex::new(vec![(paths, sender)])); + + let sender_clone = senders.clone(); + let watcher: RecommendedWatcher = Watcher::new( + move |res: Result| { + let res2 = res.map(FsEvent::from).map_err(FsEventsError::Notify); + for (paths, sender) in sender_clone.lock().iter() { + // Ignore result, if send failed it means that watcher was already closed, + // but not all messages have been flushed. + + // Only send the event if the path matches one of the paths that the user is watching + if let Ok(event) = &res2 { + if paths.iter().any(|path| { + event.paths.iter().any(|event_path| { + same_file::is_same_file(event_path, path).unwrap_or(false) + || starts_with_canonicalized(event_path, path) + }) + }) { + let _ = sender.try_send(Ok(event.clone())); + } + } + } + }, + Default::default(), + )?; + + state.put::(WatcherState { watcher, senders }); + + Ok(()) } #[op2] #[smi] fn op_fs_events_open( state: &mut OpState, - #[serde] args: OpenArgs, + recursive: bool, + #[serde] paths: Vec, ) -> Result { let (sender, receiver) = mpsc::channel::>(16); - let sender = Mutex::new(sender); - let mut watcher: RecommendedWatcher = Watcher::new( - move |res: Result| { - let res2 = res.map(FsEvent::from); - let sender = sender.lock(); - // Ignore result, if send failed it means that watcher was already closed, - // but not all messages have been flushed. - let _ = sender.try_send(res2); - }, - Default::default(), - )?; - let recursive_mode = if args.recursive { + + start_watcher(state, paths.clone(), sender)?; + + let recursive_mode = if recursive { RecursiveMode::Recursive } else { RecursiveMode::NonRecursive }; - for path in &args.paths { + for path in &paths { let path = state .borrow_mut::() .check_read(path, "Deno.watchFs()") .map_err(FsEventsError::Permission)?; - watcher.watch(&path, recursive_mode)?; + + let watcher = state.borrow_mut::(); + watcher.watcher.watch(&path, recursive_mode)?; } let resource = FsEventsResource { - watcher, receiver: AsyncRefCell::new(receiver), cancel: Default::default(), };