More robust file watching

Cargo.toml

@@ -77,6 +77,8 @@ inherits = "release"
 lto = true
 
 [dev-dependencies]
+filetime = "0.2.21"
 insta = "1.29.0"
 pretty_assertions = "1.3.0"
+tempfile = "3.6.0"
 wiremock = "0.5.18"

src/main.rs

@@ -10,12 +10,14 @@ pub mod renderer;
 pub mod table;
 pub mod text;
 pub mod utils;
+mod watcher;
 
 use crate::image::Image;
 use crate::interpreter::HtmlInterpreter;
 use crate::opts::Opts;
 use crate::table::Table;
 use crate::text::Text;
+use watcher::Watcher;
 
 use crate::image::ImageData;
 use keybindings::{
@@ -41,7 +43,6 @@ use anyhow::Context;
 use copypasta::{nop_clipboard::NopClipboardContext as ClipboardContext, ClipboardProvider};
 #[cfg(feature = "x11")]
 use copypasta::{ClipboardContext, ClipboardProvider};
-use notify::{RecommendedWatcher, RecursiveMode, Watcher};
 
 use winit::event::ModifiersState;
 use winit::event::{ElementState, MouseButton};
@@ -65,7 +66,6 @@ use std::sync::mpsc;
 use std::sync::mpsc::channel;
 use std::sync::Arc;
 use std::sync::Mutex;
-use std::time::Duration;
 
 pub enum InlyneEvent {
     LoadedImage(String, Arc<Mutex<Option<ImageData>>>),
@@ -146,6 +146,7 @@ pub struct Inlyne {
     interpreter_should_queue: Arc<AtomicBool>,
     keycombos: KeyCombos,
     need_repositioning: bool,
+    watcher: Watcher,
 }
 
 /// Gets a relative path extending from the repo root falling back to the full path
@@ -182,55 +183,6 @@ fn root_filepath_to_vcs_dir(path: &Path) -> Option<PathBuf> {
 }
 
 impl Inlyne {
-    pub fn spawn_watcher(&self) {
-        // Create a channel to receive the events.
-        let (watch_tx, watch_rx) = channel();
-
-        // Create a watcher object, delivering raw events.
-        // The notification back-end is selected based on the platform.
-        let mut watcher = RecommendedWatcher::new(watch_tx, notify::Config::default()).unwrap();
-
-        // Add the file path to be watched.
-        let event_proxy = self.event_loop.create_proxy();
-        let file_path = self.opts.file_path.clone();
-        std::thread::spawn(move || {
-            watcher
-                .watch(&file_path, RecursiveMode::NonRecursive)
-                .unwrap();
-
-            loop {
-                let event = match watch_rx.recv() {
-                    Ok(Ok(event)) => event,
-                    Ok(Err(err)) => {
-                        log::warn!("File watcher error: {}", err);
-                        continue;
-                    }
-                    Err(err) => {
-                        log::warn!("File watcher channel dropped unexpectedly: {}", err);
-                        break;
-                    }
-                };
-
-                log::debug!("File event: {:#?}", event);
-                if event.kind.is_modify() {
-                    let _ = event_proxy.send_event(InlyneEvent::FileReload);
-                } else if event.kind.is_remove() {
-                    // Some editors may remove/rename the file as a part of saving.
-                    // Reregister file watching in this case
-                    std::thread::sleep(Duration::from_millis(10));
-                    log::info!(
-                        "File may have been renamed/removed. Attempting to re-register file watcher"
-                    );
-
-                    let _ = watcher.unwatch(&file_path);
-                    if let Err(err) = watcher.watch(&file_path, RecursiveMode::NonRecursive) {
-                        log::warn!("Unable to watch file. No longer reloading: {}", err);
-                    }
-                }
-            }
-        });
-    }
-
     pub fn new(opts: Opts) -> anyhow::Result<Self> {
         let keycombos = KeyCombos::new(opts.keybindings.clone())?;
 
@@ -273,6 +225,8 @@ impl Inlyne {
 
         let lines_to_scroll = opts.lines_to_scroll;
 
+        let watcher = Watcher::spawn(event_loop.create_proxy(), opts.file_path.clone());
+
         Ok(Self {
             opts,
             window,
@@ -287,6 +241,7 @@ impl Inlyne {
             image_cache,
             keycombos,
             need_repositioning: false,
+            watcher,
         })
     }
 
@@ -558,9 +513,7 @@ impl Inlyne {
                                             .expect("Could not spawn new inlyne instance");
                                         } else {
                                             self.opts.file_path = path;
-                                            event_loop_proxy
-                                                .send_event(InlyneEvent::FileReload)
-                                                .expect("new file to reload successfully");
+                                            self.watcher.update_path(&self.opts.file_path);
                                         }
                                     } else if let Some(anchor_pos) =
                                         self.renderer.positioner.anchors.get(link)
@@ -804,9 +757,8 @@ fn main() -> anyhow::Result<()> {
         }),
     };
     let opts = Opts::parse_and_load_from(args, config)?;
-    let inlyne = Inlyne::new(opts)?;
 
-    inlyne.spawn_watcher();
+    let inlyne = Inlyne::new(opts)?;
     inlyne.run();
 
     Ok(())

src/watcher/mod.rs

@@ -0,0 +1,123 @@
+use std::{
+    path::{Path, PathBuf},
+    sync::mpsc,
+    time::Duration,
+};
+
+use crate::InlyneEvent;
+
+use notify::{
+    event::{EventKind, ModifyKind},
+    Event, EventHandler, RecommendedWatcher, RecursiveMode, Watcher as _,
+};
+use winit::event_loop::EventLoopProxy;
+
+#[cfg(test)]
+mod tests;
+
+trait Callback: Send + 'static {
+    fn update(&self);
+}
+
+impl Callback for EventLoopProxy<InlyneEvent> {
+    fn update(&self) {
+        let _ = self.send_event(InlyneEvent::FileReload);
+    }
+}
+
+enum WatcherMsg {
+    // Sent by the registered file watcher
+    Notify(notify::Result<Event>),
+    // Sent by the event loop
+    FileChange(PathBuf),
+}
+
+struct MsgHandler(mpsc::Sender<WatcherMsg>);
+
+impl EventHandler for MsgHandler {
+    fn handle_event(&mut self, event: notify::Result<Event>) {
+        let msg = WatcherMsg::Notify(event);
+        let _ = self.0.send(msg);
+    }
+}
+
+pub struct Watcher(mpsc::Sender<WatcherMsg>);
+
+impl Watcher {
+    pub fn spawn(event_proxy: EventLoopProxy<InlyneEvent>, file_path: PathBuf) -> Self {
+        Self::spawn_inner(event_proxy, file_path)
+    }
+
+    fn spawn_inner<C: Callback>(reload_callback: C, file_path: PathBuf) -> Self {
+        let (msg_tx, msg_rx) = mpsc::channel();
+        let watcher = Self(msg_tx.clone());
+
+        let notify_watcher =
+            RecommendedWatcher::new(MsgHandler(msg_tx), notify::Config::default()).unwrap();
+
+        std::thread::spawn(move || {
+            endlessly_handle_messages(notify_watcher, msg_rx, reload_callback, file_path);
+        });
+
+        watcher
+    }
+
+    pub fn update_path(&self, new_path: &Path) {
+        let msg = WatcherMsg::FileChange(new_path.to_owned());
+        let _ = self.0.send(msg);
+    }
+}
+
+fn endlessly_handle_messages<C: Callback>(
+    mut watcher: RecommendedWatcher,
+    msg_rx: mpsc::Receiver<WatcherMsg>,
+    reload_callback: C,
+    mut file_path: PathBuf,
+) {
+    watcher
+        .watch(&file_path, RecursiveMode::NonRecursive)
+        .unwrap();
+
+    let poll_registering_watcher = |watcher: &mut RecommendedWatcher, file_path: &Path| loop {
+        std::thread::sleep(Duration::from_millis(20));
+
+        let _ = watcher.unwatch(file_path);
+        if watcher
+            .watch(file_path, RecursiveMode::NonRecursive)
+            .is_ok()
+        {
+            break;
+        }
+    };
+
+    while let Ok(msg) = msg_rx.recv() {
+        match msg {
+            WatcherMsg::Notify(Ok(event)) => {
+                log::trace!("File event: {:#?}", event);
+
+                if matches!(
+                    event.kind,
+                    EventKind::Remove(_) | EventKind::Modify(ModifyKind::Name(_))
+                ) {
+                    log::debug!("File may have been renamed/removed. Falling back to polling");
+                    poll_registering_watcher(&mut watcher, &file_path);
+                    log::debug!("Successfully re-registered file watcher");
+                    reload_callback.update();
+                } else if matches!(event.kind, EventKind::Modify(_)) {
+                    log::debug!("Reloading file");
+                    reload_callback.update();
+                }
+            }
+            WatcherMsg::Notify(Err(err)) => log::warn!("File watcher error: {}", err),
+            WatcherMsg::FileChange(new_path) => {
+                log::info!("Updating file watcher path: {}", new_path.display());
+                let _ = watcher.unwatch(&file_path);
+                poll_registering_watcher(&mut watcher, &new_path);
+                file_path = new_path;
+                reload_callback.update();
+            }
+        }
+    }
+
+    log::warn!("File watcher channel dropped unexpectedly");
+}

src/watcher/tests.rs

@@ -0,0 +1,83 @@
+use std::{fs, path::Path, sync::mpsc, time::Duration};
+
+use super::{Callback, Watcher};
+
+impl Callback for mpsc::Sender<()> {
+    fn update(&self) {
+        self.send(()).unwrap();
+    }
+}
+
+const DELAY: Duration = Duration::from_millis(100);
+const LONG_TIMEOUT: Duration = Duration::from_millis(2_000);
+const SHORT_TIMEOUT: Duration = Duration::from_millis(50);
+
+fn delay() {
+    std::thread::sleep(DELAY);
+}
+
+fn touch(file: &Path) {
+    let now = filetime::FileTime::now();
+    filetime::set_file_mtime(file, now).unwrap();
+}
+
+#[track_caller]
+fn assert_no_message(callback: &mpsc::Receiver<()>) {
+    assert!(callback.recv_timeout(SHORT_TIMEOUT).is_err());
+}
+
+#[track_caller]
+fn assert_at_least_one_message(callback: &mpsc::Receiver<()>) {
+    assert!(callback.recv_timeout(LONG_TIMEOUT).is_ok());
+    while callback.recv_timeout(SHORT_TIMEOUT).is_ok() {}
+}
+
+// Unfortunately this needs to be littered with sleeps/timeouts to work right :/
+#[test]
+fn the_gauntlet() {
+    // Create our dummy test env
+    let temp_dir = tempfile::Builder::new()
+        .prefix("inlyne-tests-")
+        .tempdir()
+        .unwrap();
+    let base = temp_dir.path();
+    let main_file = base.join("main.md");
+    let rel_file = base.join("rel.md");
+    let swapped_in_file = base.join("swap_me_in.md");
+    let swapped_out_file = base.join("swap_out_to_me.md");
+    fs::write(&main_file, "# Main\n\n[rel](./rel.md)").unwrap();
+    fs::write(&rel_file, "# Rel").unwrap();
+    fs::write(&swapped_in_file, "# Swapped").unwrap();
+
+    // Setup our watcher
+    let (callback_tx, callback_rx) = mpsc::channel::<()>();
+    let watcher = Watcher::spawn_inner(callback_tx, main_file.clone());
+
+    // Give the watcher time to get comfy :)
+    delay();
+
+    // Sanity check watching
+    touch(&main_file);
+    assert_at_least_one_message(&callback_rx);
+
+    // Updating a file follows the new file and not the old one
+    watcher.update_path(&rel_file);
+    assert_at_least_one_message(&callback_rx);
+    touch(&main_file);
+    assert_no_message(&callback_rx);
+    touch(&rel_file);
+    assert_at_least_one_message(&callback_rx);
+
+    // We can slowly swap out the file and it will only follow the file it's supposed to
+    fs::rename(&rel_file, &swapped_out_file).unwrap();
+    touch(&swapped_out_file);
+    assert_no_message(&callback_rx);
+    // The "slowly" part of this (give the watcher time to fail and start polling)
+    delay();
+    fs::rename(&swapped_in_file, &rel_file).unwrap();
+    assert_at_least_one_message(&callback_rx);
+    fs::remove_file(&swapped_out_file).unwrap();
+    assert_no_message(&callback_rx);
+    touch(&rel_file);
+    assert_at_least_one_message(&callback_rx);
+}