notify/

poll.rs

//! Generic Watcher implementation based on polling
//!
//! Checks the `watch`ed paths periodically to detect changes. This implementation only uses
//! Rust stdlib APIs and should work on all of the platforms it supports.

use crate::{
    Config, Error, EventHandler, PathsMut, Receiver, Result, Sender, WatchMode, Watcher,
    poll::data::WatchData, unbounded,
};
use std::{
    path::{Path, PathBuf},
    sync::mpsc,
    thread,
    time::Duration,
};

/// Event sent for registered handlers on initial directory scans
pub type ScanEvent = crate::Result<PathBuf>;

/// Handler trait for receivers of [`ScanEvent`].
/// Very much the same as [`EventHandler`], but including the Result.
///
/// See the full example for more information.
pub trait ScanEventHandler: Send + 'static {
    /// Handles an event.
    fn handle_event(&mut self, event: ScanEvent);
}

impl<F> ScanEventHandler for F
where
    F: FnMut(ScanEvent) + Send + 'static,
{
    fn handle_event(&mut self, event: ScanEvent) {
        (self)(event);
    }
}

#[cfg(feature = "crossbeam-channel")]
impl ScanEventHandler for crossbeam_channel::Sender<ScanEvent> {
    fn handle_event(&mut self, event: ScanEvent) {
        let result = self.send(event);
        if let Err(e) = result {
            tracing::error!(?e, "failed to send scan event result");
        }
    }
}

#[cfg(feature = "flume")]
impl ScanEventHandler for flume::Sender<ScanEvent> {
    fn handle_event(&mut self, event: ScanEvent) {
        let result = self.send(event);
        if let Err(e) = result {
            tracing::error!(?e, "failed to send scan event result");
        }
    }
}

impl ScanEventHandler for std::sync::mpsc::Sender<ScanEvent> {
    fn handle_event(&mut self, event: ScanEvent) {
        let result = self.send(event);
        if let Err(e) = result {
            tracing::error!(?e, "failed to send scan event result");
        }
    }
}

impl ScanEventHandler for () {
    fn handle_event(&mut self, _event: ScanEvent) {}
}

use data::DataBuilder;
mod data {
    use crate::{
        Error, EventHandler, Result, WatchMode,
        consolidating_path_trie::ConsolidatingPathTrie,
        event::{CreateKind, DataChange, Event, EventKind, MetadataKind, ModifyKind, RemoveKind},
    };
    use rustc_hash::FxBuildHasher;
    use std::{
        cell::RefCell,
        collections::{HashMap, hash_map::RandomState},
        fmt::{self, Debug},
        fs::{File, FileType, Metadata},
        hash::{BuildHasher, Hasher},
        io::{self, Read},
        path::{Path, PathBuf},
        time::{Instant, SystemTime},
    };
    use walkdir::WalkDir;

    use super::ScanEventHandler;

    /// Builder for [`WatchData`] & [`PathData`].
    pub(super) struct DataBuilder {
        emitter: EventEmitter,
        scan_emitter: Option<Box<RefCell<dyn ScanEventHandler>>>,

        // TODO: May allow user setup their custom BuildHasher / BuildHasherDefault
        // in future.
        build_hasher: Option<RandomState>,

        // current timestamp for building Data.
        now: Instant,
    }

    impl DataBuilder {
        pub(super) fn new<F, G>(
            event_handler: F,
            compare_content: bool,
            scan_emitter: Option<G>,
        ) -> Self
        where
            F: EventHandler,
            G: ScanEventHandler,
        {
            let scan_emitter = match scan_emitter {
                None => None,
                Some(v) => {
                    // workaround for a weird type resolution bug when directly going to dyn Trait
                    let intermediate: Box<RefCell<dyn ScanEventHandler>> =
                        Box::new(RefCell::new(v));
                    Some(intermediate)
                }
            };
            Self {
                emitter: EventEmitter::new(event_handler),
                scan_emitter,
                build_hasher: compare_content.then(RandomState::default),
                now: Instant::now(),
            }
        }

        /// Update internal timestamp.
        pub(super) fn update_timestamp(&mut self) {
            self.now = Instant::now();
        }

        /// Create [`PathData`].
        fn build_path_data(&self, meta_path: &MetaPath) -> PathData {
            PathData::new(self, meta_path)
        }
    }

    impl Debug for DataBuilder {
        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
            f.debug_struct("DataBuilder")
                .field("build_hasher", &self.build_hasher)
                .field("now", &self.now)
                .finish_non_exhaustive()
        }
    }

    type SingleWatchHandlerMap = HashMap<PathBuf, /* recursive */ bool, FxBuildHasher>;

    #[derive(Debug)]
    struct WatchHandlers {
        current: SingleWatchHandlerMap,
        next: SingleWatchHandlerMap,
        is_stale: bool,
    }

    impl WatchHandlers {
        fn new() -> Self {
            Self {
                current: HashMap::default(),
                next: HashMap::default(),
                is_stale: false,
            }
        }

        /// Recalculate from `watches`.
        fn recalculate(&mut self, watches: &HashMap<PathBuf, WatchMode, FxBuildHasher>) {
            self.next.clear();
            self.is_stale = true;

            let mut trie = ConsolidatingPathTrie::new(false, 0);
            for (path, mode) in watches {
                if mode.recursive_mode == crate::RecursiveMode::Recursive {
                    trie.insert(path);
                }
            }
            // insert non-recursive watches that are not covered by recursive watches
            for (path, mode) in watches {
                if mode.recursive_mode != crate::RecursiveMode::Recursive {
                    self.next.insert(path.clone(), false);
                }
            }
            // insert recursive watches
            for path in trie.values() {
                self.next.insert(path, true);
            }
        }

        fn use_handlers(&mut self) -> (&SingleWatchHandlerMap, Option<SingleWatchHandlerMap>) {
            if self.is_stale {
                let old_next = std::mem::take(&mut self.next);
                let old_current = std::mem::replace(&mut self.current, old_next);
                self.is_stale = false;
                return (&self.current, Some(old_current));
            }
            (&self.current, None)
        }
    }

    #[derive(Debug)]
    pub(super) struct WatchData {
        // config part, won't change.
        follow_symlinks: bool,

        // current status part.
        watches: HashMap<PathBuf, WatchMode, FxBuildHasher>,
        watch_handlers: WatchHandlers,
        all_path_data: HashMap<PathBuf, PathData, FxBuildHasher>,
    }

    impl WatchData {
        /// Create a new `WatchData`.
        pub fn new(follow_symlinks: bool) -> Self {
            Self {
                follow_symlinks,
                watches: HashMap::default(),
                watch_handlers: WatchHandlers::new(),
                all_path_data: HashMap::default(),
            }
        }

        pub fn add_watch(&mut self, path: PathBuf, mode: WatchMode) -> Result<()> {
            if mode.target_mode == crate::TargetMode::NoTrack && !path.exists() {
                return Err(crate::Error::path_not_found().add_path(path));
            }

            self.watches.insert(path, mode);
            self.watch_handlers.recalculate(&self.watches);
            Ok(())
        }

        pub fn add_watch_multiple(&mut self, paths: Vec<(PathBuf, WatchMode)>) -> Result<()> {
            for (path, mode) in paths {
                if mode.target_mode == crate::TargetMode::NoTrack && !path.exists() {
                    return Err(crate::Error::path_not_found().add_path(path));
                }

                self.watches.insert(path, mode);
            }
            self.watch_handlers.recalculate(&self.watches);
            Ok(())
        }

        pub fn remove_watch(&mut self, path: &Path) -> Result<()> {
            self.watches.remove(path).ok_or(Error::watch_not_found())?;
            self.watch_handlers.recalculate(&self.watches);
            Ok(())
        }

        /// Rescan filesystem and update this `WatchData`.
        ///
        /// # Side effect
        ///
        /// This function may emit event by `data_builder.emitter`.
        pub(super) fn rescan(&mut self, data_builder: &DataBuilder) {
            let (watch_handlers, old_watch_handlers) = self.watch_handlers.use_handlers();

            // scan current filesystem.
            for (path, new_path_data) in
                Self::scan_all_path_data(data_builder, watch_handlers, self.follow_symlinks)
            {
                let event_kind = if let Some(old_path_data) = self.all_path_data.get_mut(&path) {
                    let event_kind =
                        PathData::compare_to_kind(Some(&*old_path_data), Some(&new_path_data));
                    *old_path_data = new_path_data;
                    event_kind
                } else {
                    let event_kind = PathData::compare_to_kind(None, Some(&new_path_data));
                    self.all_path_data.insert(path.clone(), new_path_data);
                    event_kind
                };

                let is_initial = old_watch_handlers
                    .as_ref()
                    .is_some_and(|old_watch_handlers| {
                        !old_watch_handlers.contains_key(&path)
                            && !path.ancestors().skip(1).any(|ancestor| {
                                old_watch_handlers
                                    .get(ancestor)
                                    .is_some_and(|is_recursive| *is_recursive)
                            })
                    });
                if is_initial {
                    // emit initial scans
                    if let Some(ref emitter) = data_builder.scan_emitter {
                        emitter.borrow_mut().handle_event(Ok(path.clone()));
                    }
                } else if let Some(event_kind) = event_kind {
                    let event = Event::new(event_kind).add_path(path);
                    data_builder.emitter.emit_ok(event);
                }
            }

            // scan for disappeared paths.
            let mut disappeared_paths = Vec::new();
            for (path, path_data) in &self.all_path_data {
                if path_data.last_check < data_builder.now {
                    disappeared_paths.push(path.clone());
                }
            }

            // remove disappeared paths
            for path in disappeared_paths {
                let old_path_data = self.all_path_data.remove(&path);

                if let Some(event_kind) = PathData::compare_to_kind(old_path_data.as_ref(), None) {
                    let event = Event::new(event_kind).add_path(path);
                    data_builder.emitter.emit_ok(event);
                }
            }
        }

        /// Get all `PathData` by given configuration.
        ///
        /// # Side Effect
        ///
        /// This function may emit some IO Error events by `data_builder.emitter`.
        fn scan_all_path_data(
            data_builder: &DataBuilder,
            watch_handlers: &HashMap<PathBuf, /* recursive */ bool, FxBuildHasher>,
            follow_symlinks: bool,
        ) -> impl Iterator<Item = (PathBuf, PathData)> {
            tracing::trace!("rescanning");

            watch_handlers.iter().flat_map(move |(path, is_recursive)| {
                tracing::trace!(?path, is_recursive, "scanning watch handler");

                // WalkDir return only one entry if root is a file (not a folder),
                // so we can use single logic to do the both file & dir's jobs.
                //
                // See: https://docs.rs/walkdir/2.0.1/walkdir/struct.WalkDir.html#method.new
                WalkDir::new(path)
                    .follow_links(follow_symlinks)
                    .max_depth(if *is_recursive { usize::MAX } else { 1 })
                    .into_iter()
                    .filter_map(|entry_res| match entry_res {
                        Ok(entry) => Some(entry),
                        Err(err) => {
                            tracing::warn!("walkdir error scanning {err:?}");

                            if let Some(io_error) = err.io_error() {
                                if io_error.kind() == io::ErrorKind::NotFound {
                                    return None;
                                }
                                // clone an io::Error, so we have to create a new one.
                                let new_io_error = io::Error::new(io_error.kind(), err.to_string());
                                data_builder.emitter.emit_io_err(new_io_error, err.path());
                            } else {
                                let crate_err =
                                    Error::new(crate::ErrorKind::Generic(err.to_string()));
                                data_builder.emitter.emit(Err(crate_err));
                            }
                            None
                        }
                    })
                    .filter_map(move |entry| match entry.metadata() {
                        Ok(metadata) => {
                            let path = entry.into_path();
                            let meta_path = MetaPath::from_parts_unchecked(path, metadata);
                            let data_path = data_builder.build_path_data(&meta_path);

                            Some((meta_path.into_path(), data_path))
                        }
                        Err(err) => {
                            if let Some(io_error) = err.io_error()
                                && io_error.kind() == io::ErrorKind::NotFound
                            {
                                return None;
                            }

                            // emit event.
                            let path = entry.into_path();
                            data_builder.emitter.emit_io_err(err, Some(path));

                            None
                        }
                    })
            })
        }
    }

    /// Stored data for a one path locations.
    ///
    /// See [`WatchData`] for more detail.
    #[derive(Debug, Clone)]
    struct PathData {
        /// File updated time.
        mtime: SystemTime,

        file_type: FileType,

        /// Content's hash value, only available if user request compare file
        /// contents and read successful.
        hash: Option<u64>,

        /// Checked time.
        last_check: Instant,
    }

    impl PathData {
        /// Create a new `PathData`.
        fn new(data_builder: &DataBuilder, meta_path: &MetaPath) -> PathData {
            let metadata = meta_path.metadata();

            PathData {
                mtime: metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH),
                file_type: metadata.file_type(),
                hash: data_builder
                    .build_hasher
                    .as_ref()
                    .filter(|_| metadata.is_file())
                    .and_then(|build_hasher| {
                        Self::get_content_hash(build_hasher, meta_path.path()).ok()
                    }),

                last_check: data_builder.now,
            }
        }

        /// Get hash value for the data content in given file `path`.
        fn get_content_hash(build_hasher: &RandomState, path: &Path) -> io::Result<u64> {
            let mut hasher = build_hasher.build_hasher();
            let mut file = File::open(path)?;
            let mut buf = [0; 512];

            loop {
                let n = match file.read(&mut buf) {
                    Ok(0) => break,
                    Ok(len) => len,
                    Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
                    Err(e) => return Err(e),
                };

                hasher.write(&buf[..n]);
            }

            Ok(hasher.finish())
        }

        /// Get `CreateKind` by file type.
        fn get_create_kind(&self) -> CreateKind {
            #[expect(clippy::filetype_is_file)]
            if self.file_type.is_dir() {
                CreateKind::Folder
            } else if self.file_type.is_file() {
                CreateKind::File
            } else {
                CreateKind::Any
            }
        }

        /// Get `RemoveKind` by file type.
        fn get_remove_kind(&self) -> RemoveKind {
            #[expect(clippy::filetype_is_file)]
            if self.file_type.is_dir() {
                RemoveKind::Folder
            } else if self.file_type.is_file() {
                RemoveKind::File
            } else {
                RemoveKind::Any
            }
        }

        /// Get [`EventKind`] by compare two optional [`PathData`].
        fn compare_to_kind(old: Option<&PathData>, new: Option<&PathData>) -> Option<EventKind> {
            match (old, new) {
                (Some(old), Some(new)) => {
                    if new.mtime > old.mtime {
                        Some(EventKind::Modify(ModifyKind::Metadata(
                            MetadataKind::WriteTime,
                        )))
                    } else if new.hash != old.hash {
                        Some(EventKind::Modify(ModifyKind::Data(DataChange::Any)))
                    } else {
                        None
                    }
                }
                (None, Some(new)) => Some(EventKind::Create(new.get_create_kind())),
                (Some(old), None) => Some(EventKind::Remove(old.get_remove_kind())),
                (None, None) => None,
            }
        }
    }

    /// Compose path and its metadata.
    ///
    /// This data structure designed for make sure path and its metadata can be
    /// transferred in consistent way, and may avoid some duplicated
    /// `fs::metadata()` function call in some situations.
    #[derive(Debug)]
    pub(super) struct MetaPath {
        path: PathBuf,
        metadata: Metadata,
    }

    impl MetaPath {
        /// Create `MetaPath` by given parts.
        ///
        /// # Invariant
        ///
        /// User must make sure the input `metadata` are associated with `path`.
        fn from_parts_unchecked(path: PathBuf, metadata: Metadata) -> Self {
            Self { path, metadata }
        }

        fn path(&self) -> &Path {
            &self.path
        }

        fn metadata(&self) -> &Metadata {
            &self.metadata
        }

        fn into_path(self) -> PathBuf {
            self.path
        }
    }

    /// Thin wrapper for outer event handler, for easy to use.
    struct EventEmitter(
        // Use `RefCell` to make sure `emit()` only need shared borrow of self (&self).
        // Use `Box` to make sure EventEmitter is Sized.
        Box<RefCell<dyn EventHandler>>,
    );

    impl EventEmitter {
        fn new<F: EventHandler>(event_handler: F) -> Self {
            Self(Box::new(RefCell::new(event_handler)))
        }

        /// Emit single event.
        fn emit(&self, event: crate::Result<Event>) {
            self.0.borrow_mut().handle_event(event);
        }

        /// Emit event.
        fn emit_ok(&self, event: Event) {
            self.emit(Ok(event));
        }

        /// Emit io error event.
        fn emit_io_err<E, P>(&self, err: E, path: Option<P>)
        where
            E: Into<io::Error>,
            P: Into<PathBuf>,
        {
            let e = crate::Error::io(err.into());
            if let Some(path) = path {
                self.emit(Err(e.add_path(path.into())));
            } else {
                self.emit(Err(e));
            }
        }
    }
}

enum EventLoopMsg {
    AddWatch(PathBuf, WatchMode, Sender<Result<()>>),
    AddWatchMultiple(Vec<(PathBuf, WatchMode)>, Sender<Result<()>>),
    RemoveWatch(PathBuf, Sender<Result<()>>),
    #[cfg(test)]
    WaitNextScan(Sender<Result<()>>),
    /// currently used only for manual polling
    Poll,
    Shutdown,
}

struct PollPathsMut<'a> {
    inner: &'a mut PollWatcher,
    add_paths: Vec<(PathBuf, WatchMode)>,
}
impl<'a> PollPathsMut<'a> {
    fn new(watcher: &'a mut PollWatcher) -> Self {
        Self {
            inner: watcher,
            add_paths: Vec::new(),
        }
    }
}
impl PathsMut for PollPathsMut<'_> {
    #[tracing::instrument(level = "debug", skip(self))]
    fn add(&mut self, path: &Path, watch_mode: WatchMode) -> Result<()> {
        self.add_paths.push((path.to_owned(), watch_mode));
        Ok(())
    }

    #[tracing::instrument(level = "debug", skip(self))]
    fn remove(&mut self, path: &Path) -> Result<()> {
        self.inner.unwatch_inner(path)
    }

    #[tracing::instrument(level = "debug", skip(self))]
    fn commit(self: Box<Self>) -> Result<()> {
        let paths = self.add_paths;
        self.inner.watch_multiple_inner(paths)
    }
}

/// Polling based `Watcher` implementation.
///
/// By default scans through all files and checks for changed entries based on their change date.
/// Can also be changed to perform file content change checks.
///
/// See [Config] for more details.
#[derive(Debug)]
pub struct PollWatcher {
    delay: Option<Duration>,
    follow_symlinks: bool,

    event_loop_tx: Sender<EventLoopMsg>,
}

impl PollWatcher {
    /// Create a new [`PollWatcher`], configured as needed.
    pub fn new<F: EventHandler>(event_handler: F, config: Config) -> crate::Result<PollWatcher> {
        Ok(Self::with_opt::<_, ()>(event_handler, config, None))
    }

    /// Actively poll for changes. Can be combined with a timeout of 0 to perform only manual polling.
    pub fn poll(&self) -> crate::Result<()> {
        self.event_loop_tx
            .send(EventLoopMsg::Poll)
            .map_err(|_| Error::generic("failed to send poll message"))?;
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn wait_next_scan(&self) -> crate::Result<()> {
        let (tx, rx) = unbounded();
        self.event_loop_tx
            .send(EventLoopMsg::WaitNextScan(tx))
            .map_err(|_| Error::generic("failed to send WaitNextScan message"))?;
        rx.recv().unwrap()
    }

    /// Returns a sender to initiate changes detection.
    #[cfg(test)]
    pub(crate) fn poll_sender(&self) -> Sender<()> {
        let inner_tx = self.event_loop_tx.clone();
        let (tx, rx) = unbounded();
        thread::Builder::new()
            .name("notify-rs poll loop".to_string())
            .spawn(move || {
                for () in &rx {
                    if let Err(err) = inner_tx.send(EventLoopMsg::Poll) {
                        tracing::error!(?err, "failed to send poll message");
                    }
                }
            })
            .unwrap();
        tx
    }

    /// Create a new [`PollWatcher`] with an scan event handler.
    ///
    /// `scan_fallback` is called on the initial scan with all files seen by the pollwatcher.
    pub fn with_initial_scan<F: EventHandler, G: ScanEventHandler>(
        event_handler: F,
        config: Config,
        scan_callback: G,
    ) -> crate::Result<PollWatcher> {
        Ok(Self::with_opt(event_handler, config, Some(scan_callback)))
    }

    /// create a new [`PollWatcher`] with all options.
    fn with_opt<F: EventHandler, G: ScanEventHandler>(
        event_handler: F,
        config: Config,
        scan_callback: Option<G>,
    ) -> PollWatcher {
        let (tx, rx) = unbounded();

        let poll_watcher = PollWatcher {
            delay: config.poll_interval(),
            follow_symlinks: config.follow_symlinks(),

            event_loop_tx: tx,
        };

        let data_builder =
            DataBuilder::new(event_handler, config.compare_contents(), scan_callback);
        poll_watcher.run(rx, data_builder);

        poll_watcher
    }

    fn run(&self, rx: Receiver<EventLoopMsg>, mut data_builder: DataBuilder) {
        let delay = self.delay;
        let follow_symlinks = self.follow_symlinks;

        let result = thread::Builder::new()
            .name("notify-rs poll loop".to_string())
            .spawn(move || {
                let mut watch_data = WatchData::new(follow_symlinks);

                loop {
                    data_builder.update_timestamp();
                    watch_data.rescan(&data_builder);

                    // TODO: v7.0 use delay - (Instant::now().saturating_duration_since(start))
                    let result = if let Some(delay) = delay {
                        rx.recv_timeout(delay).or_else(|e| match e {
                            mpsc::RecvTimeoutError::Timeout => Ok(EventLoopMsg::Poll),
                            mpsc::RecvTimeoutError::Disconnected => Err(mpsc::RecvError),
                        })
                    } else {
                        rx.recv()
                    };
                    match result {
                        Ok(EventLoopMsg::AddWatch(path, mode, resp_tx)) => {
                            let result = resp_tx.send(watch_data.add_watch(path, mode));
                            if let Err(e) = result {
                                tracing::error!(?e, "failed to send AddWatch response");
                            }
                        }
                        Ok(EventLoopMsg::AddWatchMultiple(paths, resp_tx)) => {
                            let result = resp_tx.send(watch_data.add_watch_multiple(paths));
                            if let Err(e) = result {
                                tracing::error!(?e, "failed to send AddWatchMultiple response");
                            }
                        }
                        Ok(EventLoopMsg::RemoveWatch(path, resp_tx)) => {
                            let result = resp_tx.send(watch_data.remove_watch(&path));
                            if let Err(e) = result {
                                tracing::error!(?e, "failed to send RemoveWatch response");
                            }
                        }
                        Ok(EventLoopMsg::Poll) => {
                            // continue the loop
                        }
                        #[cfg(test)]
                        Ok(EventLoopMsg::WaitNextScan(resp_tx)) => {
                            let result = resp_tx.send(Ok(()));
                            if let Err(e) = result {
                                tracing::error!(?e, "failed to send WaitNextScan response");
                            }
                        }
                        Ok(EventLoopMsg::Shutdown) => {
                            break;
                        }
                        Err(e) => {
                            tracing::error!(?e, "failed to receive poll message");
                        }
                    }
                }
            });
        if let Err(e) = result {
            tracing::error!(?e, "failed to start poll watcher thread");
        }
    }

    /// Watch a path location.
    fn watch_inner(&self, path: &Path, watch_mode: WatchMode) -> crate::Result<()> {
        let (tx, rx) = unbounded();
        self.event_loop_tx
            .send(EventLoopMsg::AddWatch(path.to_path_buf(), watch_mode, tx))?;
        rx.recv().unwrap()
    }

    fn watch_multiple_inner(&self, paths: Vec<(PathBuf, WatchMode)>) -> crate::Result<()> {
        let (tx, rx) = unbounded();
        self.event_loop_tx
            .send(EventLoopMsg::AddWatchMultiple(paths, tx))?;
        rx.recv().unwrap()
    }

    /// Unwatch a path.
    ///
    /// Return `Err(_)` if given path has't be monitored.
    fn unwatch_inner(&self, path: &Path) -> crate::Result<()> {
        let (tx, rx) = unbounded();
        self.event_loop_tx
            .send(EventLoopMsg::RemoveWatch(path.to_path_buf(), tx))?;
        rx.recv().unwrap()
    }
}

impl Watcher for PollWatcher {
    /// Create a new [`PollWatcher`].
    #[tracing::instrument(level = "debug", skip(event_handler))]
    fn new<F: EventHandler>(event_handler: F, config: Config) -> crate::Result<Self> {
        Self::new(event_handler, config)
    }

    #[tracing::instrument(level = "debug", skip(self))]
    fn watch(&mut self, path: &Path, watch_mode: WatchMode) -> crate::Result<()> {
        self.watch_inner(path, watch_mode)
    }

    #[tracing::instrument(level = "debug", skip(self))]
    fn paths_mut<'me>(&'me mut self) -> Box<dyn PathsMut + 'me> {
        Box::new(PollPathsMut::new(self))
    }

    #[tracing::instrument(level = "debug", skip(self))]
    fn unwatch(&mut self, path: &Path) -> crate::Result<()> {
        self.unwatch_inner(path)
    }

    fn kind() -> crate::WatcherKind {
        crate::WatcherKind::PollWatcher
    }
}

impl Drop for PollWatcher {
    fn drop(&mut self) {
        let result = self.event_loop_tx.send(EventLoopMsg::Shutdown);
        if let Err(e) = result {
            tracing::error!(?e, "failed to send shutdown message to poll watcher thread");
        }
    }
}

#[cfg(test)]
mod tests {
    #[cfg(target_family = "wasm")]
    use std::thread::sleep;
    #[cfg(target_family = "wasm")]
    use std::time::Duration;

    use super::PollWatcher;
    use crate::{
        Error, ErrorKind, RecursiveMode, TargetMode, WatchMode, Watcher, event::EventKind, test::*,
    };

    fn watcher() -> (TestWatcher<PollWatcher>, Receiver) {
        poll_watcher_channel()
    }

    #[test]
    fn poll_watcher_is_send_and_sync() {
        fn check<T: Send + Sync>() {}
        check::<PollWatcher>();
    }

    #[test]
    fn create_file() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        watcher.watch_recursively(&tmpdir);
        watcher.watcher.wait_next_scan().expect("wait next scan");

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("Unable to create");

        rx.sleep_until_parent_contains(&path);
        rx.sleep_until_exists(&path);

        rx.wait_unordered_exact([
            expected(&path).create_file(),
            expected(tmpdir.path()).modify_meta_mtime().optional(),
        ]);
    }

    #[test]
    fn create_self_file() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();

        let path = tmpdir.path().join("entry");

        watcher.watch_nonrecursively(&path);
        watcher.watcher.wait_next_scan().expect("wait next scan");

        std::fs::File::create_new(&path).expect("create");

        rx.sleep_until_exists(&path);
        rx.wait_ordered_exact([expected(&path).create_file()]);
    }

    #[test]
    fn create_self_file_no_track() {
        let tmpdir = testdir();
        let (mut watcher, _) = watcher();

        let path = tmpdir.path().join("entry");

        let result = watcher.watcher.watch(
            &path,
            WatchMode {
                recursive_mode: RecursiveMode::NonRecursive,
                target_mode: TargetMode::NoTrack,
            },
        );
        assert!(matches!(
            result,
            Err(Error {
                paths: _,
                kind: ErrorKind::PathNotFound
            })
        ));
    }

    #[test]
    fn create_self_file_nested() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();

        let path = tmpdir.path().join("entry/nested");

        watcher.watch_nonrecursively(&path);
        watcher.watcher.wait_next_scan().expect("wait next scan");

        std::fs::create_dir_all(path.parent().unwrap()).expect("create");
        std::fs::File::create_new(&path).expect("create");

        rx.wait_ordered_exact([expected(&path).create_file()]);
    }

    #[test]
    fn create_dir() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        watcher.watch_recursively(&tmpdir);
        watcher.watcher.wait_next_scan().expect("wait next scan");

        let path = tmpdir.path().join("entry");
        std::fs::create_dir(&path).expect("Unable to create");

        rx.sleep_until_parent_contains(&path);
        rx.sleep_until_exists(&path);

        rx.wait_unordered_exact([
            expected(&path).create_folder(),
            expected(tmpdir.path()).modify_meta_mtime().optional(),
        ]);
    }

    #[test]
    fn modify_file() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("Unable to create");

        rx.sleep_until_parent_contains(&path);

        watcher.watch_recursively(&tmpdir);
        watcher.watcher.wait_next_scan().expect("wait next scan");
        std::fs::write(&path, b"123").expect("Unable to write");

        assert!(
            rx.sleep_until(|| std::fs::read_to_string(&path).is_ok_and(|content| content == "123")),
            "the file wasn't modified"
        );
        rx.wait_unordered_exact([expected(&path).modify().multiple()]);
    }

    #[test]
    fn rename_file() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        let path = tmpdir.path().join("entry");
        let new_path = tmpdir.path().join("new_entry");
        std::fs::File::create_new(&path).expect("Unable to create");

        rx.sleep_until_parent_contains(&path);

        watcher.watch_recursively(&tmpdir);

        watcher.watcher.wait_next_scan().expect("wait next scan");
        std::fs::rename(&path, &new_path).expect("Unable to remove");

        rx.sleep_while_exists(&path);
        rx.sleep_until_exists(&new_path);

        rx.sleep_while_parent_contains(&path);
        rx.sleep_until_parent_contains(&new_path);

        rx.wait_unordered_exact([
            expected(&path).remove_file(),
            expected(&new_path).create_file(),
            expected(tmpdir.path()).modify_meta_mtime().optional(),
        ]);
    }

    #[test]
    fn rename_self_file() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        watcher.watch_nonrecursively(&path);
        watcher.watcher.wait_next_scan().expect("wait next scan");
        let new_path = tmpdir.path().join("renamed");

        std::fs::rename(&path, &new_path).expect("rename");

        rx.sleep_while_exists(&path);
        rx.sleep_until_exists(&new_path);

        rx.wait_unordered_exact([expected(&path).remove_file()])
            .ensure_no_tail();

        std::fs::rename(&new_path, &path).expect("rename2");
        watcher.watcher.wait_next_scan().expect("wait next scan");

        rx.sleep_while_exists(&new_path);
        rx.sleep_until_exists(&path);

        rx.wait_unordered_exact([expected(&path).create_file()])
            .ensure_no_tail();
    }

    #[test]
    fn rename_self_file_no_track() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();

        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("create");

        watcher.watch(
            &path,
            WatchMode {
                recursive_mode: RecursiveMode::NonRecursive,
                target_mode: TargetMode::NoTrack,
            },
        );
        watcher.watcher.wait_next_scan().expect("wait next scan");

        let new_path = tmpdir.path().join("renamed");

        std::fs::rename(&path, &new_path).expect("rename");

        rx.sleep_while_exists(&path);
        rx.sleep_until_exists(&new_path);

        #[cfg(target_family = "wasm")]
        sleep(Duration::from_millis(100));

        rx.wait_unordered_exact([
            expected(&path).modify_data_any().optional(),
            expected(&path).remove_file(),
        ])
        .ensure_no_tail();

        let result = watcher.watcher.watch(
            &path,
            WatchMode {
                recursive_mode: RecursiveMode::NonRecursive,
                target_mode: TargetMode::NoTrack,
            },
        );
        assert!(matches!(
            result,
            Err(Error {
                paths: _,
                kind: ErrorKind::PathNotFound
            })
        ));
    }

    #[test]
    fn delete_file() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("Unable to create");

        rx.sleep_until_parent_contains(&path);

        watcher.watch_recursively(&tmpdir);
        watcher.watcher.wait_next_scan().expect("wait next scan");

        std::fs::remove_file(&path).expect("Unable to remove");

        rx.sleep_while_exists(&path);
        rx.sleep_while_parent_contains(&path);

        rx.wait_unordered_exact([
            expected(&path).modify_data_any().optional(),
            expected(&path).remove_file(),
            expected(tmpdir.path()).modify_meta_mtime().optional(),
        ]);
    }

    #[test]
    fn delete_self_file() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("Unable to create");

        watcher.watch_nonrecursively(&path);
        watcher.watcher.wait_next_scan().expect("wait next scan");

        std::fs::remove_file(&path).expect("Unable to remove");

        rx.sleep_while_exists(&path);
        rx.wait_ordered_exact([
            expected(&path).modify_data_any().optional(),
            expected(&path).remove_file(),
        ]);

        std::fs::write(&path, "").expect("write");

        rx.sleep_until_exists(&path);
        rx.wait_ordered_exact([expected(&path).create_file()]);
    }

    #[test]
    fn delete_self_file_no_track() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        let path = tmpdir.path().join("entry");
        std::fs::File::create_new(&path).expect("Unable to create");

        watcher.watch(
            &path,
            WatchMode {
                recursive_mode: RecursiveMode::NonRecursive,
                target_mode: TargetMode::NoTrack,
            },
        );
        watcher.watcher.wait_next_scan().expect("wait next scan");

        std::fs::remove_file(&path).expect("Unable to remove");

        rx.sleep_while_exists(&path);
        rx.wait_ordered_exact([
            expected(&path).modify_data_any().optional(),
            expected(&path).remove_file(),
        ]);

        #[cfg(target_family = "wasm")]
        sleep(Duration::from_millis(100));

        std::fs::write(&path, "").expect("write");

        rx.ensure_empty_with_wait();
    }

    #[test]
    fn create_write_overwrite() {
        let tmpdir = testdir();
        let (mut watcher, rx) = watcher();
        let overwritten_file = tmpdir.path().join("overwritten_file");
        let overwriting_file = tmpdir.path().join("overwriting_file");
        std::fs::write(&overwritten_file, "123").expect("write1");

        rx.sleep_until_parent_contains(&overwritten_file);
        rx.sleep_until_exists(&overwritten_file);

        watcher.watch_nonrecursively(&tmpdir);
        watcher.watcher.wait_next_scan().expect("wait next scan");

        std::fs::File::create(&overwriting_file).expect("create");
        std::fs::write(&overwriting_file, "321").expect("write2");
        std::fs::rename(&overwriting_file, &overwritten_file).expect("rename");

        rx.sleep_while_exists(&overwriting_file);
        rx.sleep_while_parent_contains(&overwriting_file);

        assert!(
            rx.sleep_until(
                || std::fs::read_to_string(&overwritten_file).is_ok_and(|cnt| cnt == "321")
            ),
            "file {overwritten_file:?} was not replaced"
        );

        rx.wait_unordered([expected(&overwritten_file).modify()]);
    }

    fn assert_track_path_continues_after_recreating_file_in_nested_directory(
        upgrade_from_no_track: bool,
    ) {
        let tmpdir = testdir();
        let (mut watcher, mut rx) = watcher();
        let nested_dir = tmpdir.path().join("nested");
        let watched_file = nested_dir.join("watched");
        let moved_file = tmpdir.path().join("moved");
        std::fs::create_dir(&nested_dir).expect("create nested dir");
        std::fs::write(&watched_file, "initial").expect("write watched file");

        watcher.watch_nonrecursively(&tmpdir);
        if upgrade_from_no_track {
            watcher.watch(
                &watched_file,
                WatchMode {
                    recursive_mode: RecursiveMode::NonRecursive,
                    target_mode: TargetMode::NoTrack,
                },
            );
        }
        watcher.watch_nonrecursively(&watched_file);

        std::fs::rename(&watched_file, &moved_file).expect("move watched file");
        std::fs::copy(&moved_file, &watched_file).expect("recreate watched file");
        std::fs::remove_file(&moved_file).expect("remove moved file");

        // Complete a scan, then drain the replacement events before checking the next write.
        watcher.watcher.poll().expect("scan replacement");
        watcher.watcher.wait_next_scan().expect("wait for scan");
        for _ in rx.iter() {}

        std::fs::write(&watched_file, "updated").expect("update watched file");
        watcher.watcher.poll().expect("scan update");
        watcher.watcher.wait_next_scan().expect("wait for scan");
        let received_change = rx.iter().any(|event| {
            event.paths.iter().any(|path| path == &watched_file)
                && matches!(event.kind, EventKind::Create(_) | EventKind::Modify(_))
        });

        assert!(
            received_change,
            "expected a change event after recreating the watched file"
        );
    }

    #[test]
    fn track_path_continues_after_recreating_file_in_nested_directory() {
        assert_track_path_continues_after_recreating_file_in_nested_directory(false);
    }

    #[test]
    fn track_path_upgrade_continues_after_recreating_file_in_nested_directory() {
        assert_track_path_continues_after_recreating_file_in_nested_directory(true);
    }
}