1""":module: watchdog.observers.inotify_buffer 2:synopsis: A wrapper for ``Inotify``. 3:author: [email protected] (Thomas Amland) 4:author: [email protected] (Mickaël Schoentgen) 5:platforms: linux 6""" 7 8from __future__ import annotations 9 10import logging 11 12from watchdog.observers.inotify_c import Inotify, InotifyEvent 13from watchdog.utils import BaseThread 14from watchdog.utils.delayed_queue import DelayedQueue 15 16logger = logging.getLogger(__name__) 17 18 19class InotifyBuffer(BaseThread): 20 """A wrapper for `Inotify` that holds events for `delay` seconds. During 21 this time, IN_MOVED_FROM and IN_MOVED_TO events are paired. 22 """ 23 24 delay = 0.5 25 26 def __init__(self, path: bytes, *, recursive: bool = False, event_mask: int | None = None) -> None: 27 super().__init__() 28 # XXX: Remove quotes after Python 3.9 drop 29 self._queue = DelayedQueue["InotifyEvent | tuple[InotifyEvent, InotifyEvent]"](self.delay) 30 self._inotify = Inotify(path, recursive=recursive, event_mask=event_mask) 31 self.start() 32 33 def read_event(self) -> InotifyEvent | tuple[InotifyEvent, InotifyEvent] | None: 34 """Returns a single event or a tuple of from/to events in case of a 35 paired move event. If this buffer has been closed, immediately return 36 None. 37 """ 38 return self._queue.get() 39 40 def on_thread_stop(self) -> None: 41 self._inotify.close() 42 self._queue.close() 43 44 def close(self) -> None: 45 self.stop() 46 self.join() 47 48 def _group_events(self, event_list: list[InotifyEvent]) -> list[InotifyEvent | tuple[InotifyEvent, InotifyEvent]]: 49 """Group any matching move events""" 50 grouped: list[InotifyEvent | tuple[InotifyEvent, InotifyEvent]] = [] 51 for inotify_event in event_list: 52 logger.debug("in-event %s", inotify_event) 53 54 def matching_from_event(event: InotifyEvent | tuple[InotifyEvent, InotifyEvent]) -> bool: 55 return not isinstance(event, tuple) and event.is_moved_from and event.cookie == inotify_event.cookie 56 57 if inotify_event.is_moved_to: 58 # Check if move_from is already in the buffer 59 for index, event in enumerate(grouped): 60 if matching_from_event(event): 61 grouped[index] = (event, inotify_event) # type: ignore[assignment] 62 break 63 else: 64 # Check if move_from is in delayqueue already 65 from_event = self._queue.remove(matching_from_event) 66 if from_event is not None: 67 grouped.append((from_event, inotify_event)) # type: ignore[arg-type] 68 else: 69 logger.debug("could not find matching move_from event") 70 grouped.append(inotify_event) 71 else: 72 grouped.append(inotify_event) 73 return grouped 74 75 def run(self) -> None: 76 """Read event from `inotify` and add them to `queue`. When reading a 77 IN_MOVE_TO event, remove the previous added matching IN_MOVE_FROM event 78 and add them back to the queue as a tuple. 79 """ 80 deleted_self = False 81 while self.should_keep_running() and not deleted_self: 82 inotify_events = self._inotify.read_events() 83 grouped_events = self._group_events(inotify_events) 84 for inotify_event in grouped_events: 85 if not isinstance(inotify_event, tuple) and inotify_event.is_ignored: 86 if inotify_event.src_path == self._inotify.path: 87 # Watch was removed explicitly (inotify_rm_watch(2)) or automatically (file 88 # was deleted, or filesystem was unmounted), stop watching for events 89 deleted_self = True 90 continue 91 92 # Only add delay for unmatched move_from events 93 delay = not isinstance(inotify_event, tuple) and inotify_event.is_moved_from 94 self._queue.put(inotify_event, delay=delay) 95 96 if ( 97 not isinstance(inotify_event, tuple) 98 and inotify_event.is_delete_self 99 and inotify_event.src_path == self._inotify.path 100 ): 101 # Deleted the watched directory, stop watching for events 102 deleted_self = True 103