notify/
poll.rs

1//! Generic Watcher implementation based on polling
2//!
3//! Checks the `watch`ed paths periodically to detect changes. This implementation only uses
4//! Rust stdlib APIs and should work on all of the platforms it supports.
5
6use crate::{
7    Config, Error, EventHandler, PathsMut, Receiver, Result, Sender, WatchMode, Watcher,
8    poll::data::WatchData, unbounded,
9};
10use std::{
11    path::{Path, PathBuf},
12    sync::mpsc,
13    thread,
14    time::Duration,
15};
16
17/// Event sent for registered handlers on initial directory scans
18pub type ScanEvent = crate::Result<PathBuf>;
19
20/// Handler trait for receivers of [`ScanEvent`].
21/// Very much the same as [`EventHandler`], but including the Result.
22///
23/// See the full example for more information.
24pub trait ScanEventHandler: Send + 'static {
25    /// Handles an event.
26    fn handle_event(&mut self, event: ScanEvent);
27}
28
29impl<F> ScanEventHandler for F
30where
31    F: FnMut(ScanEvent) + Send + 'static,
32{
33    fn handle_event(&mut self, event: ScanEvent) {
34        (self)(event);
35    }
36}
37
38#[cfg(feature = "crossbeam-channel")]
39impl ScanEventHandler for crossbeam_channel::Sender<ScanEvent> {
40    fn handle_event(&mut self, event: ScanEvent) {
41        let result = self.send(event);
42        if let Err(e) = result {
43            tracing::error!(?e, "failed to send scan event result");
44        }
45    }
46}
47
48#[cfg(feature = "flume")]
49impl ScanEventHandler for flume::Sender<ScanEvent> {
50    fn handle_event(&mut self, event: ScanEvent) {
51        let result = self.send(event);
52        if let Err(e) = result {
53            tracing::error!(?e, "failed to send scan event result");
54        }
55    }
56}
57
58impl ScanEventHandler for std::sync::mpsc::Sender<ScanEvent> {
59    fn handle_event(&mut self, event: ScanEvent) {
60        let result = self.send(event);
61        if let Err(e) = result {
62            tracing::error!(?e, "failed to send scan event result");
63        }
64    }
65}
66
67impl ScanEventHandler for () {
68    fn handle_event(&mut self, _event: ScanEvent) {}
69}
70
71use data::DataBuilder;
72mod data {
73    use crate::{
74        Error, EventHandler, Result, WatchMode,
75        consolidating_path_trie::ConsolidatingPathTrie,
76        event::{CreateKind, DataChange, Event, EventKind, MetadataKind, ModifyKind, RemoveKind},
77    };
78    use rustc_hash::FxBuildHasher;
79    use std::{
80        cell::RefCell,
81        collections::{HashMap, hash_map::RandomState},
82        fmt::{self, Debug},
83        fs::{File, FileType, Metadata},
84        hash::{BuildHasher, Hasher},
85        io::{self, Read},
86        path::{Path, PathBuf},
87        time::Instant,
88    };
89    use walkdir::WalkDir;
90
91    use super::ScanEventHandler;
92
93    fn system_time_to_seconds(time: std::time::SystemTime) -> i64 {
94        #[expect(clippy::cast_possible_wrap)]
95        match time.duration_since(std::time::SystemTime::UNIX_EPOCH) {
96            Ok(d) => d.as_secs() as i64,
97            Err(e) => -(e.duration().as_secs() as i64),
98        }
99    }
100
101    /// Builder for [`WatchData`] & [`PathData`].
102    pub(super) struct DataBuilder {
103        emitter: EventEmitter,
104        scan_emitter: Option<Box<RefCell<dyn ScanEventHandler>>>,
105
106        // TODO: May allow user setup their custom BuildHasher / BuildHasherDefault
107        // in future.
108        build_hasher: Option<RandomState>,
109
110        // current timestamp for building Data.
111        now: Instant,
112    }
113
114    impl DataBuilder {
115        pub(super) fn new<F, G>(
116            event_handler: F,
117            compare_content: bool,
118            scan_emitter: Option<G>,
119        ) -> Self
120        where
121            F: EventHandler,
122            G: ScanEventHandler,
123        {
124            let scan_emitter = match scan_emitter {
125                None => None,
126                Some(v) => {
127                    // workaround for a weird type resolution bug when directly going to dyn Trait
128                    let intermediate: Box<RefCell<dyn ScanEventHandler>> =
129                        Box::new(RefCell::new(v));
130                    Some(intermediate)
131                }
132            };
133            Self {
134                emitter: EventEmitter::new(event_handler),
135                scan_emitter,
136                build_hasher: compare_content.then(RandomState::default),
137                now: Instant::now(),
138            }
139        }
140
141        /// Update internal timestamp.
142        pub(super) fn update_timestamp(&mut self) {
143            self.now = Instant::now();
144        }
145
146        /// Create [`PathData`].
147        fn build_path_data(&self, meta_path: &MetaPath) -> PathData {
148            PathData::new(self, meta_path)
149        }
150    }
151
152    impl Debug for DataBuilder {
153        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
154            f.debug_struct("DataBuilder")
155                .field("build_hasher", &self.build_hasher)
156                .field("now", &self.now)
157                .finish_non_exhaustive()
158        }
159    }
160
161    type SingleWatchHandlerMap = HashMap<PathBuf, /* recursive */ bool, FxBuildHasher>;
162
163    #[derive(Debug)]
164    struct WatchHandlers {
165        current: SingleWatchHandlerMap,
166        next: SingleWatchHandlerMap,
167        is_stale: bool,
168    }
169
170    impl WatchHandlers {
171        fn new() -> Self {
172            Self {
173                current: HashMap::default(),
174                next: HashMap::default(),
175                is_stale: false,
176            }
177        }
178
179        /// Recalculate from `watches`.
180        fn recalculate(&mut self, watches: &HashMap<PathBuf, WatchMode, FxBuildHasher>) {
181            self.next.clear();
182            self.is_stale = true;
183
184            let mut trie = ConsolidatingPathTrie::new(false, 0);
185            for (path, mode) in watches {
186                if mode.recursive_mode == crate::RecursiveMode::Recursive {
187                    trie.insert(path);
188                }
189            }
190            // insert non-recursive watches that are not covered by recursive watches
191            for (path, mode) in watches {
192                if mode.recursive_mode != crate::RecursiveMode::Recursive {
193                    self.next.insert(path.clone(), false);
194                }
195            }
196            // insert recursive watches
197            for path in trie.values() {
198                self.next.insert(path, true);
199            }
200        }
201
202        fn use_handlers(&mut self) -> (&SingleWatchHandlerMap, Option<SingleWatchHandlerMap>) {
203            if self.is_stale {
204                let old_next = std::mem::take(&mut self.next);
205                let old_current = std::mem::replace(&mut self.current, old_next);
206                self.is_stale = false;
207                return (&self.current, Some(old_current));
208            }
209            (&self.current, None)
210        }
211    }
212
213    #[derive(Debug)]
214    pub(super) struct WatchData {
215        // config part, won't change.
216        follow_symlinks: bool,
217
218        // current status part.
219        watches: HashMap<PathBuf, WatchMode, FxBuildHasher>,
220        watch_handlers: WatchHandlers,
221        all_path_data: HashMap<PathBuf, PathData, FxBuildHasher>,
222    }
223
224    impl WatchData {
225        /// Create a new `WatchData`.
226        pub fn new(follow_symlinks: bool) -> Self {
227            Self {
228                follow_symlinks,
229                watches: HashMap::default(),
230                watch_handlers: WatchHandlers::new(),
231                all_path_data: HashMap::default(),
232            }
233        }
234
235        pub fn add_watch(&mut self, path: PathBuf, mode: WatchMode) -> Result<()> {
236            if mode.target_mode == crate::TargetMode::NoTrack && !path.exists() {
237                return Err(crate::Error::path_not_found().add_path(path));
238            }
239
240            self.watches.insert(path, mode);
241            self.watch_handlers.recalculate(&self.watches);
242            Ok(())
243        }
244
245        pub fn add_watch_multiple(&mut self, paths: Vec<(PathBuf, WatchMode)>) -> Result<()> {
246            for (path, mode) in paths {
247                if mode.target_mode == crate::TargetMode::NoTrack && !path.exists() {
248                    return Err(crate::Error::path_not_found().add_path(path));
249                }
250
251                self.watches.insert(path, mode);
252            }
253            self.watch_handlers.recalculate(&self.watches);
254            Ok(())
255        }
256
257        pub fn remove_watch(&mut self, path: &Path) -> Result<()> {
258            self.watches.remove(path).ok_or(Error::watch_not_found())?;
259            self.watch_handlers.recalculate(&self.watches);
260            Ok(())
261        }
262
263        /// Rescan filesystem and update this `WatchData`.
264        ///
265        /// # Side effect
266        ///
267        /// This function may emit event by `data_builder.emitter`.
268        pub(super) fn rescan(&mut self, data_builder: &DataBuilder) {
269            let (watch_handlers, old_watch_handlers) = self.watch_handlers.use_handlers();
270
271            // scan current filesystem.
272            for (path, new_path_data) in
273                Self::scan_all_path_data(data_builder, watch_handlers, self.follow_symlinks)
274            {
275                let event_kind = if let Some(old_path_data) = self.all_path_data.get_mut(&path) {
276                    let event_kind =
277                        PathData::compare_to_kind(Some(&*old_path_data), Some(&new_path_data));
278                    *old_path_data = new_path_data;
279                    event_kind
280                } else {
281                    let event_kind = PathData::compare_to_kind(None, Some(&new_path_data));
282                    self.all_path_data.insert(path.clone(), new_path_data);
283                    event_kind
284                };
285
286                let is_initial = old_watch_handlers
287                    .as_ref()
288                    .is_some_and(|old_watch_handlers| {
289                        !old_watch_handlers.contains_key(&path)
290                            && !path.ancestors().skip(1).any(|ancestor| {
291                                old_watch_handlers
292                                    .get(ancestor)
293                                    .is_some_and(|is_recursive| *is_recursive)
294                            })
295                    });
296                if is_initial {
297                    // emit initial scans
298                    if let Some(ref emitter) = data_builder.scan_emitter {
299                        emitter.borrow_mut().handle_event(Ok(path.clone()));
300                    }
301                } else if let Some(event_kind) = event_kind {
302                    let event = Event::new(event_kind).add_path(path);
303                    data_builder.emitter.emit_ok(event);
304                }
305            }
306
307            // scan for disappeared paths.
308            let mut disappeared_paths = Vec::new();
309            for (path, path_data) in &self.all_path_data {
310                if path_data.last_check < data_builder.now {
311                    disappeared_paths.push(path.clone());
312                }
313            }
314
315            // remove disappeared paths
316            for path in disappeared_paths {
317                let old_path_data = self.all_path_data.remove(&path);
318
319                if let Some(event_kind) = PathData::compare_to_kind(old_path_data.as_ref(), None) {
320                    let event = Event::new(event_kind).add_path(path);
321                    data_builder.emitter.emit_ok(event);
322                }
323            }
324        }
325
326        /// Get all `PathData` by given configuration.
327        ///
328        /// # Side Effect
329        ///
330        /// This function may emit some IO Error events by `data_builder.emitter`.
331        fn scan_all_path_data(
332            data_builder: &DataBuilder,
333            watch_handlers: &HashMap<PathBuf, /* recursive */ bool, FxBuildHasher>,
334            follow_symlinks: bool,
335        ) -> impl Iterator<Item = (PathBuf, PathData)> {
336            tracing::trace!("rescanning");
337
338            watch_handlers.iter().flat_map(move |(path, is_recursive)| {
339                tracing::trace!(?path, is_recursive, "scanning watch handler");
340
341                // WalkDir return only one entry if root is a file (not a folder),
342                // so we can use single logic to do the both file & dir's jobs.
343                //
344                // See: https://docs.rs/walkdir/2.0.1/walkdir/struct.WalkDir.html#method.new
345                WalkDir::new(path)
346                    .follow_links(follow_symlinks)
347                    .max_depth(if *is_recursive { usize::MAX } else { 1 })
348                    .into_iter()
349                    .filter_map(|entry_res| match entry_res {
350                        Ok(entry) => Some(entry),
351                        Err(err) => {
352                            tracing::warn!("walkdir error scanning {err:?}");
353
354                            if let Some(io_error) = err.io_error() {
355                                if io_error.kind() == io::ErrorKind::NotFound {
356                                    return None;
357                                }
358                                // clone an io::Error, so we have to create a new one.
359                                let new_io_error = io::Error::new(io_error.kind(), err.to_string());
360                                data_builder.emitter.emit_io_err(new_io_error, err.path());
361                            } else {
362                                let crate_err =
363                                    Error::new(crate::ErrorKind::Generic(err.to_string()));
364                                data_builder.emitter.emit(Err(crate_err));
365                            }
366                            None
367                        }
368                    })
369                    .filter_map(move |entry| match entry.metadata() {
370                        Ok(metadata) => {
371                            let path = entry.into_path();
372                            let meta_path = MetaPath::from_parts_unchecked(path, metadata);
373                            let data_path = data_builder.build_path_data(&meta_path);
374
375                            Some((meta_path.into_path(), data_path))
376                        }
377                        Err(err) => {
378                            if let Some(io_error) = err.io_error()
379                                && io_error.kind() == io::ErrorKind::NotFound
380                            {
381                                return None;
382                            }
383
384                            // emit event.
385                            let path = entry.into_path();
386                            data_builder.emitter.emit_io_err(err, Some(path));
387
388                            None
389                        }
390                    })
391            })
392        }
393    }
394
395    /// Stored data for a one path locations.
396    ///
397    /// See [`WatchData`] for more detail.
398    #[derive(Debug, Clone)]
399    struct PathData {
400        /// File updated time.
401        mtime: i64,
402
403        file_type: FileType,
404
405        /// Content's hash value, only available if user request compare file
406        /// contents and read successful.
407        hash: Option<u64>,
408
409        /// Checked time.
410        last_check: Instant,
411    }
412
413    impl PathData {
414        /// Create a new `PathData`.
415        fn new(data_builder: &DataBuilder, meta_path: &MetaPath) -> PathData {
416            let metadata = meta_path.metadata();
417
418            PathData {
419                mtime: metadata.modified().map_or(0, system_time_to_seconds),
420                file_type: metadata.file_type(),
421                hash: data_builder
422                    .build_hasher
423                    .as_ref()
424                    .filter(|_| metadata.is_file())
425                    .and_then(|build_hasher| {
426                        Self::get_content_hash(build_hasher, meta_path.path()).ok()
427                    }),
428
429                last_check: data_builder.now,
430            }
431        }
432
433        /// Get hash value for the data content in given file `path`.
434        fn get_content_hash(build_hasher: &RandomState, path: &Path) -> io::Result<u64> {
435            let mut hasher = build_hasher.build_hasher();
436            let mut file = File::open(path)?;
437            let mut buf = [0; 512];
438
439            loop {
440                let n = match file.read(&mut buf) {
441                    Ok(0) => break,
442                    Ok(len) => len,
443                    Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
444                    Err(e) => return Err(e),
445                };
446
447                hasher.write(&buf[..n]);
448            }
449
450            Ok(hasher.finish())
451        }
452
453        /// Get `CreateKind` by file type.
454        fn get_create_kind(&self) -> CreateKind {
455            #[expect(clippy::filetype_is_file)]
456            if self.file_type.is_dir() {
457                CreateKind::Folder
458            } else if self.file_type.is_file() {
459                CreateKind::File
460            } else {
461                CreateKind::Any
462            }
463        }
464
465        /// Get `RemoveKind` by file type.
466        fn get_remove_kind(&self) -> RemoveKind {
467            #[expect(clippy::filetype_is_file)]
468            if self.file_type.is_dir() {
469                RemoveKind::Folder
470            } else if self.file_type.is_file() {
471                RemoveKind::File
472            } else {
473                RemoveKind::Any
474            }
475        }
476
477        /// Get [`EventKind`] by compare two optional [`PathData`].
478        fn compare_to_kind(old: Option<&PathData>, new: Option<&PathData>) -> Option<EventKind> {
479            match (old, new) {
480                (Some(old), Some(new)) => {
481                    if new.mtime > old.mtime {
482                        Some(EventKind::Modify(ModifyKind::Metadata(
483                            MetadataKind::WriteTime,
484                        )))
485                    } else if new.hash != old.hash {
486                        Some(EventKind::Modify(ModifyKind::Data(DataChange::Any)))
487                    } else {
488                        None
489                    }
490                }
491                (None, Some(new)) => Some(EventKind::Create(new.get_create_kind())),
492                (Some(old), None) => Some(EventKind::Remove(old.get_remove_kind())),
493                (None, None) => None,
494            }
495        }
496    }
497
498    /// Compose path and its metadata.
499    ///
500    /// This data structure designed for make sure path and its metadata can be
501    /// transferred in consistent way, and may avoid some duplicated
502    /// `fs::metadata()` function call in some situations.
503    #[derive(Debug)]
504    pub(super) struct MetaPath {
505        path: PathBuf,
506        metadata: Metadata,
507    }
508
509    impl MetaPath {
510        /// Create `MetaPath` by given parts.
511        ///
512        /// # Invariant
513        ///
514        /// User must make sure the input `metadata` are associated with `path`.
515        fn from_parts_unchecked(path: PathBuf, metadata: Metadata) -> Self {
516            Self { path, metadata }
517        }
518
519        fn path(&self) -> &Path {
520            &self.path
521        }
522
523        fn metadata(&self) -> &Metadata {
524            &self.metadata
525        }
526
527        fn into_path(self) -> PathBuf {
528            self.path
529        }
530    }
531
532    /// Thin wrapper for outer event handler, for easy to use.
533    struct EventEmitter(
534        // Use `RefCell` to make sure `emit()` only need shared borrow of self (&self).
535        // Use `Box` to make sure EventEmitter is Sized.
536        Box<RefCell<dyn EventHandler>>,
537    );
538
539    impl EventEmitter {
540        fn new<F: EventHandler>(event_handler: F) -> Self {
541            Self(Box::new(RefCell::new(event_handler)))
542        }
543
544        /// Emit single event.
545        fn emit(&self, event: crate::Result<Event>) {
546            self.0.borrow_mut().handle_event(event);
547        }
548
549        /// Emit event.
550        fn emit_ok(&self, event: Event) {
551            self.emit(Ok(event));
552        }
553
554        /// Emit io error event.
555        fn emit_io_err<E, P>(&self, err: E, path: Option<P>)
556        where
557            E: Into<io::Error>,
558            P: Into<PathBuf>,
559        {
560            let e = crate::Error::io(err.into());
561            if let Some(path) = path {
562                self.emit(Err(e.add_path(path.into())));
563            } else {
564                self.emit(Err(e));
565            }
566        }
567    }
568}
569
570enum EventLoopMsg {
571    AddWatch(PathBuf, WatchMode, Sender<Result<()>>),
572    AddWatchMultiple(Vec<(PathBuf, WatchMode)>, Sender<Result<()>>),
573    RemoveWatch(PathBuf, Sender<Result<()>>),
574    #[cfg(test)]
575    WaitNextScan(Sender<Result<()>>),
576    /// currently used only for manual polling
577    Poll,
578    Shutdown,
579}
580
581struct PollPathsMut<'a> {
582    inner: &'a mut PollWatcher,
583    add_paths: Vec<(PathBuf, WatchMode)>,
584}
585impl<'a> PollPathsMut<'a> {
586    fn new(watcher: &'a mut PollWatcher) -> Self {
587        Self {
588            inner: watcher,
589            add_paths: Vec::new(),
590        }
591    }
592}
593impl PathsMut for PollPathsMut<'_> {
594    #[tracing::instrument(level = "debug", skip(self))]
595    fn add(&mut self, path: &Path, watch_mode: WatchMode) -> Result<()> {
596        self.add_paths.push((path.to_owned(), watch_mode));
597        Ok(())
598    }
599
600    #[tracing::instrument(level = "debug", skip(self))]
601    fn remove(&mut self, path: &Path) -> Result<()> {
602        self.inner.unwatch_inner(path)
603    }
604
605    #[tracing::instrument(level = "debug", skip(self))]
606    fn commit(self: Box<Self>) -> Result<()> {
607        let paths = self.add_paths;
608        self.inner.watch_multiple_inner(paths)
609    }
610}
611
612/// Polling based `Watcher` implementation.
613///
614/// By default scans through all files and checks for changed entries based on their change date.
615/// Can also be changed to perform file content change checks.
616///
617/// See [Config] for more details.
618#[derive(Debug)]
619pub struct PollWatcher {
620    delay: Option<Duration>,
621    follow_symlinks: bool,
622
623    event_loop_tx: Sender<EventLoopMsg>,
624}
625
626impl PollWatcher {
627    /// Create a new [`PollWatcher`], configured as needed.
628    pub fn new<F: EventHandler>(event_handler: F, config: Config) -> crate::Result<PollWatcher> {
629        Ok(Self::with_opt::<_, ()>(event_handler, config, None))
630    }
631
632    /// Actively poll for changes. Can be combined with a timeout of 0 to perform only manual polling.
633    pub fn poll(&self) -> crate::Result<()> {
634        self.event_loop_tx
635            .send(EventLoopMsg::Poll)
636            .map_err(|_| Error::generic("failed to send poll message"))?;
637        Ok(())
638    }
639
640    #[cfg(test)]
641    pub(crate) fn wait_next_scan(&self) -> crate::Result<()> {
642        let (tx, rx) = unbounded();
643        self.event_loop_tx
644            .send(EventLoopMsg::WaitNextScan(tx))
645            .map_err(|_| Error::generic("failed to send WaitNextScan message"))?;
646        rx.recv().unwrap()
647    }
648
649    /// Returns a sender to initiate changes detection.
650    #[cfg(test)]
651    pub(crate) fn poll_sender(&self) -> Sender<()> {
652        let inner_tx = self.event_loop_tx.clone();
653        let (tx, rx) = unbounded();
654        thread::Builder::new()
655            .name("notify-rs poll loop".to_string())
656            .spawn(move || {
657                for () in &rx {
658                    if let Err(err) = inner_tx.send(EventLoopMsg::Poll) {
659                        tracing::error!(?err, "failed to send poll message");
660                    }
661                }
662            })
663            .unwrap();
664        tx
665    }
666
667    /// Create a new [`PollWatcher`] with an scan event handler.
668    ///
669    /// `scan_fallback` is called on the initial scan with all files seen by the pollwatcher.
670    pub fn with_initial_scan<F: EventHandler, G: ScanEventHandler>(
671        event_handler: F,
672        config: Config,
673        scan_callback: G,
674    ) -> crate::Result<PollWatcher> {
675        Ok(Self::with_opt(event_handler, config, Some(scan_callback)))
676    }
677
678    /// create a new [`PollWatcher`] with all options.
679    fn with_opt<F: EventHandler, G: ScanEventHandler>(
680        event_handler: F,
681        config: Config,
682        scan_callback: Option<G>,
683    ) -> PollWatcher {
684        let (tx, rx) = unbounded();
685
686        let poll_watcher = PollWatcher {
687            delay: config.poll_interval(),
688            follow_symlinks: config.follow_symlinks(),
689
690            event_loop_tx: tx,
691        };
692
693        let data_builder =
694            DataBuilder::new(event_handler, config.compare_contents(), scan_callback);
695        poll_watcher.run(rx, data_builder);
696
697        poll_watcher
698    }
699
700    fn run(&self, rx: Receiver<EventLoopMsg>, mut data_builder: DataBuilder) {
701        let delay = self.delay;
702        let follow_symlinks = self.follow_symlinks;
703
704        let result = thread::Builder::new()
705            .name("notify-rs poll loop".to_string())
706            .spawn(move || {
707                let mut watch_data = WatchData::new(follow_symlinks);
708
709                loop {
710                    data_builder.update_timestamp();
711                    watch_data.rescan(&data_builder);
712
713                    // TODO: v7.0 use delay - (Instant::now().saturating_duration_since(start))
714                    let result = if let Some(delay) = delay {
715                        rx.recv_timeout(delay).or_else(|e| match e {
716                            mpsc::RecvTimeoutError::Timeout => Ok(EventLoopMsg::Poll),
717                            mpsc::RecvTimeoutError::Disconnected => Err(mpsc::RecvError),
718                        })
719                    } else {
720                        rx.recv()
721                    };
722                    match result {
723                        Ok(EventLoopMsg::AddWatch(path, mode, resp_tx)) => {
724                            let result = resp_tx.send(watch_data.add_watch(path, mode));
725                            if let Err(e) = result {
726                                tracing::error!(?e, "failed to send AddWatch response");
727                            }
728                        }
729                        Ok(EventLoopMsg::AddWatchMultiple(paths, resp_tx)) => {
730                            let result = resp_tx.send(watch_data.add_watch_multiple(paths));
731                            if let Err(e) = result {
732                                tracing::error!(?e, "failed to send AddWatchMultiple response");
733                            }
734                        }
735                        Ok(EventLoopMsg::RemoveWatch(path, resp_tx)) => {
736                            let result = resp_tx.send(watch_data.remove_watch(&path));
737                            if let Err(e) = result {
738                                tracing::error!(?e, "failed to send RemoveWatch response");
739                            }
740                        }
741                        Ok(EventLoopMsg::Poll) => {
742                            // continue the loop
743                        }
744                        #[cfg(test)]
745                        Ok(EventLoopMsg::WaitNextScan(resp_tx)) => {
746                            let result = resp_tx.send(Ok(()));
747                            if let Err(e) = result {
748                                tracing::error!(?e, "failed to send WaitNextScan response");
749                            }
750                        }
751                        Ok(EventLoopMsg::Shutdown) => {
752                            break;
753                        }
754                        Err(e) => {
755                            tracing::error!(?e, "failed to receive poll message");
756                        }
757                    }
758                }
759            });
760        if let Err(e) = result {
761            tracing::error!(?e, "failed to start poll watcher thread");
762        }
763    }
764
765    /// Watch a path location.
766    fn watch_inner(&self, path: &Path, watch_mode: WatchMode) -> crate::Result<()> {
767        let (tx, rx) = unbounded();
768        self.event_loop_tx
769            .send(EventLoopMsg::AddWatch(path.to_path_buf(), watch_mode, tx))?;
770        rx.recv().unwrap()
771    }
772
773    fn watch_multiple_inner(&self, paths: Vec<(PathBuf, WatchMode)>) -> crate::Result<()> {
774        let (tx, rx) = unbounded();
775        self.event_loop_tx
776            .send(EventLoopMsg::AddWatchMultiple(paths, tx))?;
777        rx.recv().unwrap()
778    }
779
780    /// Unwatch a path.
781    ///
782    /// Return `Err(_)` if given path has't be monitored.
783    fn unwatch_inner(&self, path: &Path) -> crate::Result<()> {
784        let (tx, rx) = unbounded();
785        self.event_loop_tx
786            .send(EventLoopMsg::RemoveWatch(path.to_path_buf(), tx))?;
787        rx.recv().unwrap()
788    }
789}
790
791impl Watcher for PollWatcher {
792    /// Create a new [`PollWatcher`].
793    #[tracing::instrument(level = "debug", skip(event_handler))]
794    fn new<F: EventHandler>(event_handler: F, config: Config) -> crate::Result<Self> {
795        Self::new(event_handler, config)
796    }
797
798    #[tracing::instrument(level = "debug", skip(self))]
799    fn watch(&mut self, path: &Path, watch_mode: WatchMode) -> crate::Result<()> {
800        self.watch_inner(path, watch_mode)
801    }
802
803    #[tracing::instrument(level = "debug", skip(self))]
804    fn paths_mut<'me>(&'me mut self) -> Box<dyn PathsMut + 'me> {
805        Box::new(PollPathsMut::new(self))
806    }
807
808    #[tracing::instrument(level = "debug", skip(self))]
809    fn unwatch(&mut self, path: &Path) -> crate::Result<()> {
810        self.unwatch_inner(path)
811    }
812
813    fn kind() -> crate::WatcherKind {
814        crate::WatcherKind::PollWatcher
815    }
816}
817
818impl Drop for PollWatcher {
819    fn drop(&mut self) {
820        let result = self.event_loop_tx.send(EventLoopMsg::Shutdown);
821        if let Err(e) = result {
822            tracing::error!(?e, "failed to send shutdown message to poll watcher thread");
823        }
824    }
825}
826
827#[cfg(test)]
828mod tests {
829    #[cfg(target_family = "wasm")]
830    use std::thread::sleep;
831    #[cfg(target_family = "wasm")]
832    use std::time::Duration;
833
834    use super::PollWatcher;
835    use crate::{Error, ErrorKind, RecursiveMode, TargetMode, WatchMode, Watcher, test::*};
836
837    fn watcher() -> (TestWatcher<PollWatcher>, Receiver) {
838        poll_watcher_channel()
839    }
840
841    #[test]
842    fn poll_watcher_is_send_and_sync() {
843        fn check<T: Send + Sync>() {}
844        check::<PollWatcher>();
845    }
846
847    #[test]
848    fn create_file() {
849        let tmpdir = testdir();
850        let (mut watcher, rx) = watcher();
851        watcher.watch_recursively(&tmpdir);
852        watcher.watcher.wait_next_scan().expect("wait next scan");
853
854        let path = tmpdir.path().join("entry");
855        std::fs::File::create_new(&path).expect("Unable to create");
856
857        rx.sleep_until_parent_contains(&path);
858        rx.sleep_until_exists(&path);
859
860        rx.wait_ordered_exact([expected(&path).create_file()]);
861    }
862
863    #[test]
864    fn create_self_file() {
865        let tmpdir = testdir();
866        let (mut watcher, rx) = watcher();
867
868        let path = tmpdir.path().join("entry");
869
870        watcher.watch_nonrecursively(&path);
871        watcher.watcher.wait_next_scan().expect("wait next scan");
872
873        std::fs::File::create_new(&path).expect("create");
874
875        rx.sleep_until_exists(&path);
876        rx.wait_ordered_exact([expected(&path).create_file()]);
877    }
878
879    #[test]
880    fn create_self_file_no_track() {
881        let tmpdir = testdir();
882        let (mut watcher, _) = watcher();
883
884        let path = tmpdir.path().join("entry");
885
886        let result = watcher.watcher.watch(
887            &path,
888            WatchMode {
889                recursive_mode: RecursiveMode::NonRecursive,
890                target_mode: TargetMode::NoTrack,
891            },
892        );
893        assert!(matches!(
894            result,
895            Err(Error {
896                paths: _,
897                kind: ErrorKind::PathNotFound
898            })
899        ));
900    }
901
902    #[test]
903    fn create_self_file_nested() {
904        let tmpdir = testdir();
905        let (mut watcher, rx) = watcher();
906
907        let path = tmpdir.path().join("entry/nested");
908
909        watcher.watch_nonrecursively(&path);
910        watcher.watcher.wait_next_scan().expect("wait next scan");
911
912        std::fs::create_dir_all(path.parent().unwrap()).expect("create");
913        std::fs::File::create_new(&path).expect("create");
914
915        rx.wait_ordered_exact([expected(&path).create_file()]);
916    }
917
918    #[test]
919    fn create_dir() {
920        let tmpdir = testdir();
921        let (mut watcher, rx) = watcher();
922        watcher.watch_recursively(&tmpdir);
923        watcher.watcher.wait_next_scan().expect("wait next scan");
924
925        let path = tmpdir.path().join("entry");
926        std::fs::create_dir(&path).expect("Unable to create");
927
928        rx.sleep_until_parent_contains(&path);
929        rx.sleep_until_exists(&path);
930
931        rx.wait_ordered_exact([expected(&path).create_folder()]);
932    }
933
934    #[test]
935    fn modify_file() {
936        let tmpdir = testdir();
937        let (mut watcher, rx) = watcher();
938        let path = tmpdir.path().join("entry");
939        std::fs::File::create_new(&path).expect("Unable to create");
940
941        rx.sleep_until_parent_contains(&path);
942
943        watcher.watch_recursively(&tmpdir);
944        watcher.watcher.wait_next_scan().expect("wait next scan");
945        std::fs::write(&path, b"123").expect("Unable to write");
946
947        assert!(
948            rx.sleep_until(|| std::fs::read_to_string(&path).is_ok_and(|content| content == "123")),
949            "the file wasn't modified"
950        );
951        rx.wait_ordered_exact([expected(&path).modify_data_any()]);
952    }
953
954    #[test]
955    fn rename_file() {
956        let tmpdir = testdir();
957        let (mut watcher, rx) = watcher();
958        let path = tmpdir.path().join("entry");
959        let new_path = tmpdir.path().join("new_entry");
960        std::fs::File::create_new(&path).expect("Unable to create");
961
962        rx.sleep_until_parent_contains(&path);
963
964        watcher.watch_recursively(&tmpdir);
965
966        watcher.watcher.wait_next_scan().expect("wait next scan");
967        std::fs::rename(&path, &new_path).expect("Unable to remove");
968
969        rx.sleep_while_exists(&path);
970        rx.sleep_until_exists(&new_path);
971
972        rx.sleep_while_parent_contains(&path);
973        rx.sleep_until_parent_contains(&new_path);
974
975        rx.wait_unordered_exact([
976            expected(&path).remove_file(),
977            expected(&new_path).create_file(),
978        ]);
979    }
980
981    #[test]
982    fn rename_self_file() {
983        let tmpdir = testdir();
984        let (mut watcher, rx) = watcher();
985
986        let path = tmpdir.path().join("entry");
987        std::fs::File::create_new(&path).expect("create");
988
989        watcher.watch_nonrecursively(&path);
990        watcher.watcher.wait_next_scan().expect("wait next scan");
991        let new_path = tmpdir.path().join("renamed");
992
993        std::fs::rename(&path, &new_path).expect("rename");
994
995        rx.sleep_while_exists(&path);
996        rx.sleep_until_exists(&new_path);
997
998        rx.wait_unordered_exact([expected(&path).remove_file()])
999            .ensure_no_tail();
1000
1001        std::fs::rename(&new_path, &path).expect("rename2");
1002        watcher.watcher.wait_next_scan().expect("wait next scan");
1003
1004        rx.sleep_while_exists(&new_path);
1005        rx.sleep_until_exists(&path);
1006
1007        rx.wait_unordered_exact([expected(&path).create_file()])
1008            .ensure_no_tail();
1009    }
1010
1011    #[test]
1012    fn rename_self_file_no_track() {
1013        let tmpdir = testdir();
1014        let (mut watcher, rx) = watcher();
1015
1016        let path = tmpdir.path().join("entry");
1017        std::fs::File::create_new(&path).expect("create");
1018
1019        watcher.watch(
1020            &path,
1021            WatchMode {
1022                recursive_mode: RecursiveMode::NonRecursive,
1023                target_mode: TargetMode::NoTrack,
1024            },
1025        );
1026        watcher.watcher.wait_next_scan().expect("wait next scan");
1027
1028        let new_path = tmpdir.path().join("renamed");
1029
1030        std::fs::rename(&path, &new_path).expect("rename");
1031
1032        rx.sleep_while_exists(&path);
1033        rx.sleep_until_exists(&new_path);
1034
1035        #[cfg(target_family = "wasm")]
1036        sleep(Duration::from_millis(100));
1037
1038        rx.wait_unordered_exact([
1039            expected(&path).modify_data_any().optional(),
1040            expected(&path).remove_file(),
1041        ])
1042        .ensure_no_tail();
1043
1044        let result = watcher.watcher.watch(
1045            &path,
1046            WatchMode {
1047                recursive_mode: RecursiveMode::NonRecursive,
1048                target_mode: TargetMode::NoTrack,
1049            },
1050        );
1051        assert!(matches!(
1052            result,
1053            Err(Error {
1054                paths: _,
1055                kind: ErrorKind::PathNotFound
1056            })
1057        ));
1058    }
1059
1060    #[test]
1061    fn delete_file() {
1062        let tmpdir = testdir();
1063        let (mut watcher, rx) = watcher();
1064        let path = tmpdir.path().join("entry");
1065        std::fs::File::create_new(&path).expect("Unable to create");
1066
1067        rx.sleep_until_parent_contains(&path);
1068
1069        watcher.watch_recursively(&tmpdir);
1070        watcher.watcher.wait_next_scan().expect("wait next scan");
1071
1072        std::fs::remove_file(&path).expect("Unable to remove");
1073
1074        rx.sleep_while_exists(&path);
1075        rx.sleep_while_parent_contains(&path);
1076
1077        rx.wait_ordered_exact([
1078            expected(&path).modify_data_any().optional(),
1079            expected(&path).remove_file(),
1080        ]);
1081    }
1082
1083    #[test]
1084    fn delete_self_file() {
1085        let tmpdir = testdir();
1086        let (mut watcher, rx) = watcher();
1087        let path = tmpdir.path().join("entry");
1088        std::fs::File::create_new(&path).expect("Unable to create");
1089
1090        watcher.watch_nonrecursively(&path);
1091        watcher.watcher.wait_next_scan().expect("wait next scan");
1092
1093        std::fs::remove_file(&path).expect("Unable to remove");
1094
1095        rx.sleep_while_exists(&path);
1096        rx.wait_ordered_exact([
1097            expected(&path).modify_data_any().optional(),
1098            expected(&path).remove_file(),
1099        ]);
1100
1101        std::fs::write(&path, "").expect("write");
1102
1103        rx.sleep_until_exists(&path);
1104        rx.wait_ordered_exact([expected(&path).create_file()]);
1105    }
1106
1107    #[test]
1108    fn delete_self_file_no_track() {
1109        let tmpdir = testdir();
1110        let (mut watcher, rx) = watcher();
1111        let path = tmpdir.path().join("entry");
1112        std::fs::File::create_new(&path).expect("Unable to create");
1113
1114        watcher.watch(
1115            &path,
1116            WatchMode {
1117                recursive_mode: RecursiveMode::NonRecursive,
1118                target_mode: TargetMode::NoTrack,
1119            },
1120        );
1121        watcher.watcher.wait_next_scan().expect("wait next scan");
1122
1123        std::fs::remove_file(&path).expect("Unable to remove");
1124
1125        rx.sleep_while_exists(&path);
1126        rx.wait_ordered_exact([
1127            expected(&path).modify_data_any().optional(),
1128            expected(&path).remove_file(),
1129        ]);
1130
1131        #[cfg(target_family = "wasm")]
1132        sleep(Duration::from_millis(100));
1133
1134        std::fs::write(&path, "").expect("write");
1135
1136        rx.ensure_empty_with_wait();
1137    }
1138
1139    #[test]
1140    fn create_write_overwrite() {
1141        let tmpdir = testdir();
1142        let (mut watcher, rx) = watcher();
1143        let overwritten_file = tmpdir.path().join("overwritten_file");
1144        let overwriting_file = tmpdir.path().join("overwriting_file");
1145        std::fs::write(&overwritten_file, "123").expect("write1");
1146
1147        rx.sleep_until_parent_contains(&overwritten_file);
1148        rx.sleep_until_exists(&overwritten_file);
1149
1150        watcher.watch_nonrecursively(&tmpdir);
1151        watcher.watcher.wait_next_scan().expect("wait next scan");
1152
1153        std::fs::File::create(&overwriting_file).expect("create");
1154        std::fs::write(&overwriting_file, "321").expect("write2");
1155        std::fs::rename(&overwriting_file, &overwritten_file).expect("rename");
1156
1157        rx.sleep_while_exists(&overwriting_file);
1158        rx.sleep_while_parent_contains(&overwriting_file);
1159
1160        assert!(
1161            rx.sleep_until(
1162                || std::fs::read_to_string(&overwritten_file).is_ok_and(|cnt| cnt == "321")
1163            ),
1164            "file {overwritten_file:?} was not replaced"
1165        );
1166
1167        rx.wait_unordered([expected(&overwritten_file).modify_data_any()]);
1168    }
1169}
notify/poll.rs

notify/
poll.rs
