feat: introduce Watcher::paths_mut for adding/removing paths in batch (#692)

Author: branchseer

notify/src/fsevent.rs

@@ -15,7 +15,9 @@
 #![allow(non_upper_case_globals, dead_code)]
 
 use crate::event::*;
-use crate::{unbounded, Config, Error, EventHandler, RecursiveMode, Result, Sender, Watcher};
+use crate::{
+    unbounded, Config, Error, EventHandler, PathsMut, RecursiveMode, Result, Sender, Watcher,
+};
 use fsevent_sys as fs;
 use fsevent_sys::core_foundation as cf;
 use std::collections::HashMap;
@@ -265,6 +267,29 @@ extern "C" {
     fn CFRunLoopIsWaiting(runloop: cf::CFRunLoopRef) -> cf::Boolean;
 }
 
+struct FsEventPathsMut<'a>(&'a mut FsEventWatcher);
+impl<'a> FsEventPathsMut<'a> {
+    fn new(watcher: &'a mut FsEventWatcher) -> Self {
+        watcher.stop();
+        Self(watcher)
+    }
+}
+impl PathsMut for FsEventPathsMut<'_> {
+    fn add(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()> {
+        self.0.append_path(path, recursive_mode)
+    }
+
+    fn remove(&mut self, path: &Path) -> Result<()> {
+        self.0.remove_path(path)
+    }
+
+    fn commit(&mut self) -> Result<()> {
+        // ignore return error: may be empty path list
+        let _ = self.0.run();
+        Ok(())
+    }
+}
+
 impl FsEventWatcher {
     fn from_event_handler(event_handler: Arc<Mutex<dyn EventHandler>>) -> Result<Self> {
         Ok(FsEventWatcher {
@@ -561,6 +586,10 @@ impl Watcher for FsEventWatcher {
         self.watch_inner(path, recursive_mode)
     }
 
+    fn paths_mut<'me>(&'me mut self) -> Box<dyn PathsMut + 'me> {
+        Box::new(FsEventPathsMut::new(self))
+    }
+
     fn unwatch(&mut self, path: &Path) -> Result<()> {
         self.unwatch_inner(path)
     }

notify/src/lib.rs

@@ -293,6 +293,20 @@ pub enum WatcherKind {
     NullWatcher,
 }
 
+/// Providing methods for adding and removing paths to watch.
+///
+/// `Box<dyn PathsMut>` is created by [`Watcher::paths_mut`]. See its documentation for more.
+pub trait PathsMut {
+    /// Add a new path to watch. See [`Watcher::watch`] for more.
+    fn add(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()>;
+
+    /// Remove a path from watching. See [`Watcher::unwatch`] for more.
+    fn remove(&mut self, path: &Path) -> Result<()>;
+
+    /// Ensure previously added/removed paths are applied.
+    fn commit(&mut self) -> Result<()>;
+}
+
 /// Type that can deliver file activity notifications
 ///
 /// `Watcher` is implemented per platform using the best implementation available on that platform.
@@ -328,6 +342,42 @@ pub trait Watcher {
     /// fails.
     fn unwatch(&mut self, path: &Path) -> Result<()>;
 
+    /// Add/remove paths to watch.
+    ///
+    /// For some watcher implementations this method provides better performance than multiple calls to [`Watcher::watch`] and [`Watcher::unwatch`] if you want to add/remove many paths at once.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use notify::{Watcher, RecursiveMode, Result};
+    /// # use std::path::Path;
+    /// # fn main() -> Result<()> {
+    /// # let many_paths_to_add = vec![];
+    /// let mut watcher = notify::recommended_watcher(|_event| { /* event handler */ })?;
+    /// let mut watcher_paths = watcher.paths_mut();
+    /// for path in many_paths_to_add {
+    ///     watcher_paths.add(path, RecursiveMode::Recursive)?;
+    /// }
+    /// watcher_paths.commit()?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    fn paths_mut<'me>(&'me mut self) -> Box<dyn PathsMut + 'me> {
+        struct DefaultPathsMut<'a, T: ?Sized>(&'a mut T);
+        impl<'a, T: Watcher + ?Sized> PathsMut for DefaultPathsMut<'a, T> {
+            fn add(&mut self, path: &Path, recursive_mode: RecursiveMode) -> Result<()> {
+                self.0.watch(path, recursive_mode)
+            }
+            fn remove(&mut self, path: &Path) -> Result<()> {
+                self.0.unwatch(path)
+            }
+            fn commit(&mut self) -> Result<()> {
+                Ok(())
+            }
+        }
+        Box::new(DefaultPathsMut(self))
+    }
+
     /// Configure the watcher at runtime.
     ///
     /// See the [`Config`](config/struct.Config.html) struct for all configuration options.
@@ -392,7 +442,7 @@ where
 #[cfg(test)]
 mod tests {
     use std::{
-        fs,
+        fs, iter,
         time::{Duration, Instant},
     };
 
@@ -425,6 +475,21 @@ mod tests {
         assert_debug_impl!(WatcherKind);
     }
 
+    fn iter_with_timeout(rx: &Receiver<Result<Event>>) -> impl Iterator<Item = Event> + '_ {
+        // wait for up to 10 seconds for the events
+        let deadline = Instant::now() + Duration::from_secs(10);
+        iter::from_fn(move || {
+            if Instant::now() >= deadline {
+                return None;
+            }
+            Some(
+                rx.recv_timeout(deadline - Instant::now())
+                    .expect("did not receive expected event")
+                    .expect("received an error"),
+            )
+        })
+    }
+
     #[test]
     fn integration() -> std::result::Result<(), Box<dyn std::error::Error>> {
         let dir = tempdir()?;
@@ -440,14 +505,8 @@ mod tests {
 
         println!("waiting for event at {}", file_path.display());
 
-        // wait for up to 10 seconds for the create event, ignore all other events
-        let deadline = Instant::now() + Duration::from_secs(10);
-        while deadline > Instant::now() {
-            let event = rx
-                .recv_timeout(deadline - Instant::now())
-                .expect("did not receive expected event")
-                .expect("received an error");
-
+        // wait for the create event, ignore all other events
+        for event in iter_with_timeout(&rx) {
             if event.paths == vec![file_path.clone()]
                 || event.paths == vec![file_path.canonicalize()?]
             {
@@ -478,4 +537,76 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_paths_mut() -> std::result::Result<(), Box<dyn std::error::Error>> {
+        let dir = tempdir()?;
+
+        let dir_a = dir.path().join("a");
+        let dir_b = dir.path().join("b");
+
+        fs::create_dir(&dir_a)?;
+        fs::create_dir(&dir_b)?;
+
+        let (tx, rx) = std::sync::mpsc::channel();
+        let mut watcher = RecommendedWatcher::new(tx, Config::default())?;
+
+        // start watching a and b
+        {
+            let mut watcher_paths = watcher.paths_mut();
+            watcher_paths.add(&dir_a, RecursiveMode::Recursive)?;
+            watcher_paths.add(&dir_b, RecursiveMode::Recursive)?;
+            watcher_paths.commit()?;
+        }
+
+        // create file1 in both a and b
+        let a_file1 = dir_a.join("file1");
+        let b_file1 = dir_b.join("file1");
+        fs::write(&a_file1, b"Lorem ipsum")?;
+        fs::write(&b_file1, b"Lorem ipsum")?;
+
+        // wait for create events of a/file1 and b/file1
+        let mut a_file1_encountered: bool = false;
+        let mut b_file1_encountered: bool = false;
+        for event in iter_with_timeout(&rx) {
+            for path in event.paths {
+                a_file1_encountered =
+                    a_file1_encountered || (path == a_file1 || path == a_file1.canonicalize()?);
+                b_file1_encountered =
+                    b_file1_encountered || (path == b_file1 || path == b_file1.canonicalize()?);
+            }
+            if a_file1_encountered && b_file1_encountered {
+                break;
+            }
+        }
+        assert!(a_file1_encountered, "Did not receive event of {a_file1:?}");
+        assert!(b_file1_encountered, "Did not receive event of {b_file1:?}");
+
+        // stop watching a
+        {
+            let mut watcher_paths = watcher.paths_mut();
+            watcher_paths.remove(&dir_a)?;
+            watcher_paths.commit()?;
+        }
+
+        // create file2 in both a and b
+        let a_file2 = dir_a.join("file2");
+        let b_file2 = dir_b.join("file2");
+        fs::write(&a_file2, b"Lorem ipsum")?;
+        fs::write(&b_file2, b"Lorem ipsum")?;
+
+        // wait for the create event of b/file2 only
+        for event in iter_with_timeout(&rx) {
+            for path in event.paths {
+                assert!(
+                    path != a_file2 || path != a_file2.canonicalize()?,
+                    "Event of {a_file2:?} should not be received"
+                );
+                if path == b_file2 || path == b_file2.canonicalize()? {
+                    return Ok(());
+                }
+            }
+        }
+        panic!("Did not receive the event of {b_file2:?}");
+    }
 }