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