Change EventFn to take FnMut

CHANGELOG.md

@@ -19,6 +19,10 @@
 
 ## unreleased
 
+- CHANGE: Change EventFn to take FnMut [#333]
+
+[#333]: https://github.com/notify-rs/notify/pull/333
+
 ## 5.0.0-pre.10 (2021-06-04)
 
 - FIX: Make StreamContextInfo `Send` to fix soundness issue [#325]

Cargo.toml

@@ -40,6 +40,7 @@ fsevent-sys = "4"
 winapi = { version = "0.3.8", features = ["fileapi", "handleapi", "ioapiset", "minwinbase", "synchapi", "winbase", "winnt"] }
 
 [dev-dependencies]
+futures = "0.3"
 serde_json = "1.0.39"
 tempfile = "3.2.0"
 

examples/async_monitor.rs

@@ -0,0 +1,47 @@
+use notify::{RecommendedWatcher, RecursiveMode, Event, Watcher};
+use std::path::Path;
+use futures::{SinkExt, StreamExt, channel::mpsc::{channel, Receiver}};
+
+fn async_watcher() -> notify::Result<(RecommendedWatcher, Receiver<notify::Result<Event>>)> {
+    let (mut tx, rx) = channel(1);
+
+    // Automatically select the best implementation for your platform.
+    // You can also access each implementation directly e.g. INotifyWatcher.
+    let watcher = Watcher::new_immediate(move |res| {
+        futures::executor::block_on(async {
+            tx.send(res).await.unwrap();
+        })
+    })?;
+
+    Ok((watcher, rx))
+}
+
+async fn async_watch<P: AsRef<Path>>(path: P) -> notify::Result<()> {
+    let (mut watcher, mut rx) = async_watcher()?;
+
+    // Add a path to be watched. All files and directories at that path and
+    // below will be monitored for changes.
+    watcher.watch(path, RecursiveMode::Recursive)?;
+
+    while let Some(res) = rx.next().await {
+        match res {
+            Ok(event) => println!("changed: {:?}", event),
+            Err(e) => println!("watch error: {:?}", e),
+        }
+    }
+
+    Ok(())
+}
+
+fn main() {
+    let path = std::env::args()
+        .nth(1)
+        .expect("Argument 1 needs to be a path");
+    println!("watching {}", path);
+
+    futures::executor::block_on(async {
+        if let Err(e) = async_watch(path).await {
+            println!("error: {:?}", e)
+        }
+    });
+}

src/fsevent.rs

@@ -534,7 +534,7 @@ unsafe fn callback_impl(
         for ev in translate_flags(flag, true).into_iter() {
             // TODO: precise
             let ev = ev.add_path(path.clone());
-            let event_fn = event_fn.lock().expect("lock not to be poisoned");
+            let mut event_fn = event_fn.lock().expect("lock not to be poisoned");
             (event_fn)(Ok(ev));
         }
     }

src/inotify.rs

@@ -56,7 +56,7 @@ enum EventLoopMsg {
 }
 
 #[inline]
-fn send_pending_rename_event(rename_event: &mut Option<Event>, event_fn: &dyn EventFn) {
+fn send_pending_rename_event(rename_event: &mut Option<Event>, event_fn: &mut dyn EventFn) {
     if let Some(e) = rename_event.take() {
         event_fn(Ok(e));
     }
@@ -188,7 +188,7 @@ impl EventLoop {
                     let current_cookie = self.rename_event.as_ref().and_then(|e| e.tracker());
                     // send pending rename event only if the rename event for which the timer has been created hasn't been handled already; otherwise ignore this timeout
                     if current_cookie == Some(cookie) {
-                        send_pending_rename_event(&mut self.rename_event, &*self.event_fn);
+                        send_pending_rename_event(&mut self.rename_event, &mut *self.event_fn);
                     }
                 }
                 EventLoopMsg::Configure(config, tx) => {
@@ -229,7 +229,7 @@ impl EventLoop {
                             };
 
                             if event.mask.contains(EventMask::MOVED_FROM) {
-                                send_pending_rename_event(&mut self.rename_event, &*self.event_fn);
+                                send_pending_rename_event(&mut self.rename_event, &mut *self.event_fn);
                                 remove_watch_by_event(&path, &self.watches, &mut remove_watches);
                                 self.rename_event = Some(
                                     Event::new(EventKind::Modify(ModifyKind::Name(
@@ -384,7 +384,7 @@ impl EventLoop {
                                 if !evs.is_empty() {
                                     send_pending_rename_event(
                                         &mut self.rename_event,
-                                        &*self.event_fn,
+                                        &mut *self.event_fn,
                                     );
                                 }
 

src/lib.rs

@@ -127,9 +127,9 @@ mod config;
 mod error;
 
 /// The set of requirements for watcher event handling functions.
-pub trait EventFn: 'static + Fn(Result<Event>) + Send {}
+pub trait EventFn: 'static + FnMut(Result<Event>) + Send {}
 
-impl<F> EventFn for F where F: 'static + Fn(Result<Event>) + Send {}
+impl<F> EventFn for F where F: 'static + FnMut(Result<Event>) + Send {}
 
 /// Type that can deliver file activity notifications
 ///

src/poll.rs

@@ -36,8 +36,8 @@ pub struct PollWatcher {
 }
 
 fn emit_event(event_fn: &Mutex<dyn EventFn>, res: Result<Event>) {
-    if let Ok(guard) = event_fn.lock() {
-        let f: &dyn EventFn = &*guard;
+    if let Ok(mut guard) = event_fn.lock() {
+        let f: &mut dyn EventFn = &mut *guard;
         f(res);
     }
 }

src/windows.rs

@@ -346,8 +346,8 @@ unsafe extern "system" fn handle_event(
             let newe = Event::new(EventKind::Any).add_path(path);
 
             fn emit_event(event_fn: &Mutex<dyn EventFn>, res: Result<Event>) {
-                if let Ok(guard) = event_fn.lock() {
-                    let f: &dyn EventFn = &*guard;
+                if let Ok(mut guard) = event_fn.lock() {
+                    let f: &mut dyn EventFn = &mut *guard;
                     f(res);
                 }
             }