feat: add notify instead of 500 ms polling

src/source.rs

@@ -1,20 +1,33 @@
+use crate::config::Config;
+use anyhow::Result;
+use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
 use std::fs::File;
 use std::io::{BufRead, BufReader, Seek, SeekFrom};
-use std::thread::sleep;
-use std::time::Duration;
-
-pub fn run_source(
-    cfg: crate::config::Config,
-    tx: tokio::sync::mpsc::Sender<String>,
-) -> anyhow::Result<()> {
-    let mut offset: u64 = 0;
-
-    loop {
-        let file = File::open(&cfg.source_path)?;
-        let mut reader = BufReader::new(file);
+use std::sync::mpsc as std_mpsc;
+use tokio::sync::mpsc::Sender;
+
+pub fn run_source(cfg: Config, tx: Sender<String>) -> Result<()> {
+    let file = File::open(&cfg.source_path)?;
+    let mut reader = BufReader::new(file);
+    let mut offset = reader.seek(SeekFrom::End(0))?;
+
+    let (notify_tx, notify_rx) = std_mpsc::channel::<notify::Result<Event>>();
+    let mut watcher: RecommendedWatcher = notify::recommended_watcher(move |res| {
+        let _ = notify_tx.send(res);
+    })?;
+    watcher.watch(&cfg.source_path, RecursiveMode::NonRecursive)?;
+
+    let mut line = String::new();
+
+    for res in notify_rx {
+        let event = res?;
+
+        if !matches!(event.kind, EventKind::Modify(_)) {
+            continue;
+        }
+
         reader.seek(SeekFrom::Start(offset))?;
 
-        let mut line = String::new();
         loop {
             line.clear();
             let bytes_read = reader.read_line(&mut line)?;
@@ -30,7 +43,7 @@ pub fn run_source(
                 return Ok(());
             }
         }
-
-        sleep(Duration::from_millis(500));
     }
+
+    Ok(())
 }

tests/notify_test.rs

@@ -0,0 +1,60 @@
+use std::fs;
+use std::path::PathBuf;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use notify::{EventKind, RecursiveMode, Watcher};
+
+#[test]
+fn notify_detects_file_modification() {
+    let file_name = format!(
+        "notify_test_{}.log",
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    );
+
+    let path = PathBuf::from(file_name);
+
+    fs::write(&path, "").unwrap();
+
+    let (tx, rx) = std::sync::mpsc::channel();
+
+    let mut watcher = notify::recommended_watcher(move |result| {
+        tx.send(result).unwrap();
+    })
+        .unwrap();
+
+    watcher.watch(&path, RecursiveMode::NonRecursive).unwrap();
+
+    fs::write(&path, "changed line\n").unwrap();
+
+    let timeout = Duration::from_secs(3);
+    let deadline = Instant::now() + timeout;
+    let mut received_kinds = Vec::new();
+    let mut saw_modify = false;
+
+    while Instant::now() < deadline {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+
+        let event = rx
+            .recv_timeout(remaining)
+            .expect("timeout: notify did not detect file modification")
+            .expect("notify returned error");
+
+        received_kinds.push(event.kind.clone());
+
+        if matches!(event.kind, EventKind::Modify(_)) {
+            saw_modify = true;
+            break;
+        }
+    }
+
+    fs::remove_file(&path).ok();
+
+    assert!(
+        saw_modify,
+        "expected Modify, but received events were: {:?}",
+        received_kinds
+    );
+}