diff --git a/src/source.rs b/src/source.rs index 0d34ba8..eafc6b0 100644 --- a/src/source.rs +++ b/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, -) -> 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) -> 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::>(); + 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(()) } \ No newline at end of file diff --git a/tests/notify_test.rs b/tests/notify_test.rs new file mode 100644 index 0000000..72561c5 --- /dev/null +++ b/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 + ); +} \ No newline at end of file