feat: rework PathsMut to be more consistent

The feature to add / remove paths to watch in batch was introduced in #692
There was a problem with the implementation: if you drop the instance without `commit` it may leave the watcher in an inconsistent state (especially fsevents).

This commit changes behavior and api and introduces `update_watches` method of the `Watcher` trait, which do the same stuff but in more consistent way.
The `paths_mut` method was kept, but returning struct is just a builder for the method `update_watches` that will be called by `PathsMut::commit`

related #653
closes #704

Author: riberk

notify/src/error.rs

@@ -1,7 +1,8 @@
 //! Error types
 
-use crate::Config;
+use crate::{Config, WatchOp};
 use std::error::Error as StdError;
+use std::fmt::Debug;
 use std::path::PathBuf;
 use std::result::Result as StdResult;
 use std::{self, fmt, io};
@@ -158,14 +159,55 @@ impl<T> From<std::sync::PoisonError<T>> for Error {
     }
 }
 
-#[test]
-fn display_formatted_errors() {
-    let expected = "Some error";
+/// The error provided by [`crate::Watcher::update_watches`] method
+#[derive(Debug)]
+pub struct UpdateWatchesError {
+    /// The original error
+    pub source: Error,
+
+    /// The remaining operations that haven't been applied
+    pub remaining: Vec<WatchOp>,
+}
+
+impl fmt::Display for UpdateWatchesError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "unable to apply the batch operation: {}", self.source)
+    }
+}
+
+impl StdError for UpdateWatchesError {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        Some(&self.source)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn display_formatted_errors() {
+        let expected = "Some error";
+
+        assert_eq!(expected, format!("{}", Error::generic(expected)));
 
-    assert_eq!(expected, format!("{}", Error::generic(expected)));
+        assert_eq!(
+            expected,
+            format!("{}", Error::io(io::Error::other(expected)))
+        );
+    }
+
+    #[test]
+    fn display_update_watches() {
+        let actual = UpdateWatchesError {
+            source: Error::generic("Some error"),
+            remaining: Default::default(),
+        }
+        .to_string();
 
-    assert_eq!(
-        expected,
-        format!("{}", Error::io(io::Error::other(expected)))
-    );
+        assert_eq!(
+            format!("unable to apply the batch operation: Some error"),
+            actual
+        );
+    }
 }

notify/src/fsevent.rs

@@ -14,10 +14,8 @@
 
 #![allow(non_upper_case_globals, dead_code)]
 
-use crate::event::*;
-use crate::{
-    unbounded, Config, Error, EventHandler, PathsMut, RecursiveMode, Result, Sender, Watcher,
-};
+use crate::{event::*, WatchOp};
+use crate::{unbounded, Config, Error, EventHandler, RecursiveMode, Result, Sender, Watcher};
 use fsevent_sys as fs;
 use fsevent_sys::core_foundation as cf;
 use std::collections::HashMap;
@@ -267,29 +265,6 @@ 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(self: Box<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 {
@@ -321,6 +296,23 @@ impl FsEventWatcher {
         result
     }
 
+    fn update_watches_inner(
+        &mut self,
+        ops: Vec<crate::WatchOp>,
+    ) -> crate::StdResult<(), crate::UpdateWatchesError> {
+        self.stop();
+
+        let result = crate::update_watches(ops, |op| match op {
+            crate::WatchOp::Watch(path, recursive_mode) => self.append_path(&path, recursive_mode),
+            crate::WatchOp::Unwatch(path) => self.remove_path(&path),
+        });
+
+        // ignore return error: may be empty path list
+        let _ = self.run();
+
+        result
+    }
+
     #[inline]
     fn is_running(&self) -> bool {
         self.runloop.is_some()
@@ -586,14 +578,17 @@ 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)
     }
 
+    fn update_watches(
+        &mut self,
+        ops: Vec<WatchOp>,
+    ) -> crate::StdResult<(), crate::UpdateWatchesError> {
+        self.update_watches_inner(ops)
+    }
+
     fn configure(&mut self, config: Config) -> Result<bool> {
         let (tx, rx) = unbounded();
         self.configure_raw_mode(config, tx);
@@ -603,6 +598,10 @@ impl Watcher for FsEventWatcher {
     fn kind() -> crate::WatcherKind {
         crate::WatcherKind::Fsevent
     }
+
+    fn paths_mut(&mut self) -> crate::PathsMut<'_> {
+        crate::PathsMut::new(self)
+    }
 }
 
 impl Drop for FsEventWatcher {

notify/src/inotify.rs

@@ -601,6 +601,10 @@ impl Watcher for INotifyWatcher {
     fn kind() -> crate::WatcherKind {
         crate::WatcherKind::Inotify
     }
+
+    fn paths_mut(&mut self) -> crate::PathsMut<'_> {
+        crate::PathsMut::new(self)
+    }
 }
 
 impl Drop for INotifyWatcher {

notify/src/kqueue.rs

@@ -447,6 +447,10 @@ impl Watcher for KqueueWatcher {
     fn kind() -> crate::WatcherKind {
         crate::WatcherKind::Kqueue
     }
+
+    fn paths_mut(&mut self) -> crate::PathsMut<'_> {
+        crate::PathsMut::new(self)
+    }
 }
 
 impl Drop for KqueueWatcher {

notify/src/lib.rs

@@ -163,10 +163,11 @@
 #![deny(missing_docs)]
 
 pub use config::{Config, RecursiveMode};
-pub use error::{Error, ErrorKind, Result};
+pub use error::{Error, ErrorKind, Result, UpdateWatchesError};
 pub use notify_types::event::{self, Event, EventKind};
-use std::path::Path;
+use std::path::{Path, PathBuf};
 
+pub(crate) type StdResult<T, E> = std::result::Result<T, E>;
 pub(crate) type Receiver<T> = std::sync::mpsc::Receiver<T>;
 pub(crate) type Sender<T> = std::sync::mpsc::Sender<T>;
 #[cfg(any(target_os = "linux", target_os = "android", target_os = "windows"))]
@@ -295,19 +296,56 @@ pub enum WatcherKind {
 
 /// 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<()>;
+/// It is created by [`Watcher::paths_mut`].
+pub struct PathsMut<'a> {
+    ops: Vec<WatchOp>,
+    watcher: &'a mut dyn Watcher,
+}
+
+impl<'a> Extend<WatchOp> for PathsMut<'a> {
+    fn extend<T: IntoIterator<Item = WatchOp>>(&mut self, iter: T) {
+        self.ops.extend(iter);
+    }
+}
+
+impl<'a> PathsMut<'a> {
+    /// Creates new [`PathsMut`] to operate with watched paths in a bulk manner
+    pub fn new<W: Watcher + 'a>(watcher: &'a mut W) -> Self {
+        Self {
+            ops: Default::default(),
+            watcher,
+        }
+    }
+
+    /// Plan a path to be added to watching. See [`Watcher::unwatch`] for more information.
+    pub fn watch(&mut self, path: &Path, recursive_mode: RecursiveMode) {
+        self.ops
+            .push(WatchOp::Watch(path.to_owned(), recursive_mode));
+    }
 
-    /// Remove a path from watching. See [`Watcher::unwatch`] for more.
-    fn remove(&mut self, path: &Path) -> Result<()>;
+    /// Plan a path to be removed from watching. See [`Watcher::unwatch`] for more information.
+    pub fn unwatch(&mut self, path: &Path) {
+        self.ops.push(WatchOp::Unwatch(path.to_owned()));
+    }
 
-    /// Ensure added/removed paths are applied.
+    /// Commit all the operations.
     ///
-    /// The behaviour of dropping a [`PathsMut`] without calling [`commit`] is unspecified.
-    /// The implementation is free to ignore the changes or not, and may leave the watcher in a started or stopped state.
-    fn commit(self: Box<Self>) -> Result<()>;
+    /// Dropping a [`PathsMut`] without calling [`Self::commit`] is noop.
+    pub fn commit(self) -> StdResult<(), UpdateWatchesError> {
+        self.watcher.update_watches(self.ops)
+    }
+}
+
+/// An operation to apply to a watcher
+///
+/// See [`Watcher::paths_mut`] for more information
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum WatchOp {
+    /// Path should be watcher
+    Watch(PathBuf, RecursiveMode),
+
+    /// Path should be unwatched
+    Unwatch(PathBuf),
 }
 
 /// Type that can deliver file activity notifications
@@ -345,40 +383,47 @@ pub trait Watcher {
     /// fails.
     fn unwatch(&mut self, path: &Path) -> Result<()>;
 
-    /// Add/remove paths to watch.
+    /// Add/remove paths to watch in batch.
     ///
-    /// 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.
+    /// It is ergonomic way to call [`Watcher::update_watches`]
     ///
     /// # Examples
     ///
     /// ```
-    /// # use notify::{Watcher, RecursiveMode, Result};
+    /// # use notify::{Watcher, RecursiveMode};
     /// # use std::path::Path;
-    /// # fn main() -> Result<()> {
+    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
     /// # let many_paths_to_add = vec![];
+    /// # let many_paths_to_remove = 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)?;
+    ///     // just some memory stuff
+    ///     watcher_paths.watch(path, RecursiveMode::Recursive);
     /// }
+    ///
+    /// for path in many_paths_to_remove {
+    ///     // just some memory stuff
+    ///     watcher_paths.unwatch(path);
+    /// }
+    ///
+    /// // real work is done there
     /// 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(self: Box<Self>) -> Result<()> {
-                Ok(())
-            }
-        }
-        Box::new(DefaultPathsMut(self))
+    fn paths_mut(&mut self) -> crate::PathsMut<'_>;
+
+    /// Add/remove paths to watch in batch.
+    ///
+    /// 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.
+    fn update_watches(&mut self, ops: Vec<WatchOp>) -> StdResult<(), UpdateWatchesError> {
+        update_watches(ops, |op| match op {
+            WatchOp::Watch(path, recursive_mode) => self.watch(&path, recursive_mode),
+            WatchOp::Unwatch(path) => self.unwatch(&path),
+        })
     }
 
     /// Configure the watcher at runtime.
@@ -442,6 +487,25 @@ where
     RecommendedWatcher::new(event_handler, Config::default())
 }
 
+pub(crate) fn update_watches<F>(
+    ops: Vec<WatchOp>,
+    mut apply: F,
+) -> StdResult<(), UpdateWatchesError>
+where
+    F: FnMut(WatchOp) -> Result<()>,
+{
+    let mut iter = ops.into_iter();
+    while let Some(op) = iter.next() {
+        if let Err(source) = apply(op) {
+            return Err(UpdateWatchesError {
+                source,
+                remaining: iter.collect(),
+            });
+        }
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use std::{
@@ -557,8 +621,8 @@ mod tests {
         // 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.watch(&dir_a, RecursiveMode::Recursive);
+            watcher_paths.watch(&dir_b, RecursiveMode::Recursive);
             watcher_paths.commit()?;
         }
 
@@ -588,7 +652,7 @@ mod tests {
         // stop watching a
         {
             let mut watcher_paths = watcher.paths_mut();
-            watcher_paths.remove(&dir_a)?;
+            watcher_paths.unwatch(&dir_a);
             watcher_paths.commit()?;
         }
 

notify/src/null.rs

@@ -36,4 +36,8 @@ impl Watcher for NullWatcher {
     fn kind() -> crate::WatcherKind {
         crate::WatcherKind::NullWatcher
     }
+
+    fn paths_mut(&mut self) -> crate::PathsMut<'_> {
+        crate::PathsMut::new(self)
+    }
 }