use std::path::PathBuf;
use std::time::Duration;
use notify::event::{EventKind, ModifyKind};
use notify::{Event, RecommendedWatcher, RecursiveMode, Watcher};
use tokio::sync::mpsc::Sender;
use tokio::task::JoinHandle;
use tokio::time;
use crate::workers::WorkerMsg;
pub struct FsWatchHandle {
task: JoinHandle<()>,
}
impl FsWatchHandle {
#[allow(dead_code)]
pub fn cancel(self) {
self.task.abort();
}
}
impl Drop for FsWatchHandle {
fn drop(&mut self) {
self.task.abort();
}
}
pub fn spawn_fswatch(
path: PathBuf,
tx: Sender<WorkerMsg>,
debounce_ms: u64,
) -> Option<FsWatchHandle> {
let (bridge_tx, mut bridge_rx) =
tokio::sync::mpsc::unbounded_channel::<notify::Result<Event>>();
let mut watcher = match RecommendedWatcher::new(
move |res: notify::Result<Event>| {
let _ = bridge_tx.send(res);
},
notify::Config::default(),
) {
Ok(w) => w,
Err(e) => {
tracing::debug!(?path, "failed to create fs watcher: {e}");
return None;
}
};
if let Err(e) = watcher.watch(&path, RecursiveMode::NonRecursive) {
tracing::debug!(?path, "failed to watch directory: {e}");
return None;
}
let git_head_path: Option<PathBuf> = {
let mut search = path.clone();
loop {
let candidate = search.join(".git").join("HEAD");
if candidate.exists() {
break Some(candidate);
}
match search.parent() {
Some(p) => search = p.to_owned(),
None => break None,
}
}
};
if let Some(ref head) = git_head_path {
if let Err(e) = watcher.watch(head, RecursiveMode::NonRecursive) {
tracing::debug!(?head, "failed to watch .git/HEAD: {e}");
} else {
tracing::debug!(?head, "watching .git/HEAD for branch changes");
}
}
let debounce = Duration::from_millis(debounce_ms);
let watch_path = path.clone();
let task = tokio::spawn(async move {
let _watcher = watcher;
loop {
let first = bridge_rx.recv().await;
match first {
None => break, Some(Ok(event)) if !is_relevant_event(&event) => {
continue; }
Some(Err(e)) => {
tracing::debug!(?watch_path, "fs watch error: {e}");
continue;
}
Some(Ok(_)) => {
let deadline = time::Instant::now() + debounce;
loop {
match time::timeout_at(deadline, bridge_rx.recv()).await {
Ok(Some(_)) => {
}
Ok(None) => {
return;
}
Err(_) => {
break;
}
}
}
let msg = WorkerMsg::FsChanged {
path: watch_path.clone(),
};
if tx.send(msg).await.is_err() {
break;
}
tracing::debug!(?watch_path, "FsChanged sent after debounce");
}
}
}
});
Some(FsWatchHandle { task })
}
fn is_relevant_event(event: &Event) -> bool {
match &event.kind {
EventKind::Create(_)
| EventKind::Remove(_)
| EventKind::Modify(ModifyKind::Name(_))
| EventKind::Modify(ModifyKind::Data(_))
| EventKind::Modify(ModifyKind::Metadata(_)) => true,
_ => false,
}
}