1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798 |
- # -*- coding: utf-8 -*-
- #
- # Copyright 2014 Thomas Amland <thomas.amland@gmail.com>
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- import logging
- from watchdog.utils import BaseThread
- from watchdog.utils.delayed_queue import DelayedQueue
- from watchdog.observers.inotify_c import Inotify
- logger = logging.getLogger(__name__)
- class InotifyBuffer(BaseThread):
- """A wrapper for `Inotify` that holds events for `delay` seconds. During
- this time, IN_MOVED_FROM and IN_MOVED_TO events are paired.
- """
- delay = 0.5
- def __init__(self, path, recursive=False):
- BaseThread.__init__(self)
- self._queue = DelayedQueue(self.delay)
- self._inotify = Inotify(path, recursive)
- self.start()
- def read_event(self):
- """Returns a single event or a tuple of from/to events in case of a
- paired move event. If this buffer has been closed, immediately return
- None.
- """
- return self._queue.get()
- def on_thread_stop(self):
- self._inotify.close()
- self._queue.close()
- def close(self):
- self.stop()
- self.join()
- def _group_events(self, event_list):
- """Group any matching move events"""
- grouped = []
- for inotify_event in event_list:
- logger.debug("in-event %s", inotify_event)
- def matching_from_event(event):
- return (not isinstance(event, tuple) and event.is_moved_from
- and event.cookie == inotify_event.cookie)
- if inotify_event.is_moved_to:
- # Check if move_from is already in the buffer
- for index, event in enumerate(grouped):
- if matching_from_event(event):
- grouped[index] = (event, inotify_event)
- break
- else:
- # Check if move_from is in delayqueue already
- from_event = self._queue.remove(matching_from_event)
- if from_event is not None:
- grouped.append((from_event, inotify_event))
- else:
- logger.debug("could not find matching move_from event")
- grouped.append(inotify_event)
- else:
- grouped.append(inotify_event)
- return grouped
- def run(self):
- """Read event from `inotify` and add them to `queue`. When reading a
- IN_MOVE_TO event, remove the previous added matching IN_MOVE_FROM event
- and add them back to the queue as a tuple.
- """
- deleted_self = False
- while self.should_keep_running() and not deleted_self:
- inotify_events = self._inotify.read_events()
- grouped_events = self._group_events(inotify_events)
- for inotify_event in grouped_events:
- # Only add delay for unmatched move_from events
- delay = not isinstance(inotify_event, tuple) and inotify_event.is_moved_from
- self._queue.put(inotify_event, delay)
- if not isinstance(inotify_event, tuple) and inotify_event.is_delete_self and \
- inotify_event.src_path == self._inotify.path:
- # Deleted the watched directory, stop watching for events
- deleted_self = True
|