use std::{
collections::{HashMap, HashSet},
fs::metadata,
mem::take,
path::{Path, PathBuf},
time::Duration,
};
use async_priority_channel as priority;
use normalize_path::NormalizePath;
use notify::{Config, Watcher as _};
use tokio::sync::{mpsc, watch};
use tracing::{debug, error, trace};
use crate::{
error::{CriticalError, FsWatcherError, RuntimeError},
event::{Event, Priority, Source, Tag},
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum Watcher {
Native,
Poll(Duration),
}
impl Default for Watcher {
fn default() -> Self {
Self::Native
}
}
impl Watcher {
fn create(
self,
f: impl notify::EventHandler,
) -> Result<Box<dyn notify::Watcher + Send>, RuntimeError> {
match self {
Self::Native => {
notify::RecommendedWatcher::new(f, Config::default()).map(|w| Box::new(w) as _)
}
Self::Poll(delay) => {
notify::PollWatcher::new(f, Config::default().with_poll_interval(delay))
.map(|w| Box::new(w) as _)
}
}
.map_err(|err| RuntimeError::FsWatcher {
kind: self,
err: if cfg!(target_os = "linux")
&& (matches!(err.kind, notify::ErrorKind::MaxFilesWatch)
|| matches!(err.kind, notify::ErrorKind::Io(ref ioerr) if ioerr.raw_os_error() == Some(28)))
{
FsWatcherError::TooManyWatches(err)
} else if cfg!(target_os = "linux")
&& matches!(err.kind, notify::ErrorKind::Io(ref ioerr) if ioerr.raw_os_error() == Some(24))
{
FsWatcherError::TooManyHandles(err)
} else {
FsWatcherError::Create(err)
},
})
}
}
#[derive(Clone, Debug, Default)]
#[non_exhaustive]
pub struct WorkingData {
pub pathset: Vec<WatchedPath>,
pub watcher: Watcher,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct WatchedPath(PathBuf);
impl From<PathBuf> for WatchedPath {
fn from(path: PathBuf) -> Self {
Self(path)
}
}
impl From<&str> for WatchedPath {
fn from(path: &str) -> Self {
Self(path.into())
}
}
impl From<&Path> for WatchedPath {
fn from(path: &Path) -> Self {
Self(path.into())
}
}
impl From<WatchedPath> for PathBuf {
fn from(path: WatchedPath) -> Self {
path.0
}
}
impl AsRef<Path> for WatchedPath {
fn as_ref(&self) -> &Path {
self.0.as_ref()
}
}
pub async fn worker(
mut working: watch::Receiver<WorkingData>,
errors: mpsc::Sender<RuntimeError>,
events: priority::Sender<Event, Priority>,
) -> Result<(), CriticalError> {
debug!("launching filesystem worker");
let mut watcher_type = Watcher::default();
let mut watcher = None;
let mut pathset = HashSet::new();
while working.changed().await.is_ok() {
let (new_watcher, to_watch, to_drop) = {
let data = working.borrow();
trace!(?data, "filesystem worker got a working data change");
if data.pathset.is_empty() {
trace!("no more watched paths, dropping watcher");
watcher.take();
pathset.drain();
continue;
}
if watcher.is_none() || watcher_type != data.watcher {
pathset.drain();
(Some(data.watcher), data.pathset.clone(), Vec::new())
} else {
let mut to_watch = Vec::with_capacity(data.pathset.len());
let mut to_drop = Vec::with_capacity(pathset.len());
for path in &data.pathset {
if !pathset.contains(path) {
to_watch.push(path.clone());
}
}
for path in &pathset {
if !data.pathset.contains(path) {
to_drop.push(path.clone());
}
}
(None, to_watch, to_drop)
}
};
if let Some(kind) = new_watcher {
debug!(?kind, "creating new watcher");
let n_errors = errors.clone();
let n_events = events.clone();
match kind.create(move |nev: Result<notify::Event, notify::Error>| {
trace!(event = ?nev, "receiving possible event from watcher");
if let Err(e) = process_event(nev, kind, &n_events) {
n_errors.try_send(e).ok();
}
}) {
Ok(w) => {
watcher = Some(w);
watcher_type = kind;
}
Err(e) => {
errors.send(e).await?;
}
}
}
if let Some(w) = watcher.as_mut() {
debug!(?to_watch, ?to_drop, "applying changes to the watcher");
for path in to_drop {
trace!(?path, "removing path from the watcher");
if let Err(err) = w.unwatch(path.as_ref()) {
error!(?err, "notify unwatch() error");
for e in notify_multi_path_errors(watcher_type, path, err, true) {
errors.send(e).await?;
}
} else {
pathset.remove(&path);
}
}
for path in to_watch {
trace!(?path, "adding path to the watcher");
if let Err(err) = w.watch(path.as_ref(), notify::RecursiveMode::Recursive) {
error!(?err, "notify watch() error");
for e in notify_multi_path_errors(watcher_type, path, err, false) {
errors.send(e).await?;
}
} else {
pathset.insert(path);
}
}
}
}
debug!("ending file watcher");
Ok(())
}
fn notify_multi_path_errors(
kind: Watcher,
path: WatchedPath,
mut err: notify::Error,
rm: bool,
) -> Vec<RuntimeError> {
let mut paths = take(&mut err.paths);
if paths.is_empty() {
paths.push(path.into());
}
let generic = err.to_string();
let mut err = Some(err);
let mut errs = Vec::with_capacity(paths.len());
for path in paths {
let e = err
.take()
.unwrap_or_else(|| notify::Error::generic(&generic))
.add_path(path.clone());
errs.push(RuntimeError::FsWatcher {
kind,
err: if rm {
FsWatcherError::PathRemove { path, err: e }
} else {
FsWatcherError::PathAdd { path, err: e }
},
});
}
errs
}
fn process_event(
nev: Result<notify::Event, notify::Error>,
kind: Watcher,
n_events: &priority::Sender<Event, Priority>,
) -> Result<(), RuntimeError> {
let nev = nev.map_err(|err| RuntimeError::FsWatcher {
kind,
err: FsWatcherError::Event(err),
})?;
let mut tags = Vec::with_capacity(4);
tags.push(Tag::Source(Source::Filesystem));
tags.push(Tag::FileEventKind(nev.kind));
for path in nev.paths {
tags.push(Tag::Path {
file_type: metadata(&path).ok().map(|m| m.file_type().into()),
path: path.normalize(),
});
}
if let Some(pid) = nev.attrs.process_id() {
tags.push(Tag::Process(pid));
}
let mut metadata = HashMap::new();
if let Some(uid) = nev.attrs.info() {
metadata.insert("file-event-info".to_string(), vec![uid.to_string()]);
}
if let Some(src) = nev.attrs.source() {
metadata.insert("notify-backend".to_string(), vec![src.to_string()]);
}
let ev = Event { tags, metadata };
trace!(event = ?ev, "processed notify event into watchexec event");
n_events
.try_send(ev, Priority::Normal)
.map_err(|err| RuntimeError::EventChannelTrySend {
ctx: "fs watcher",
err,
})?;
Ok(())
}