Skip to content

Instantly share code, notes, and snippets.

@sigmaris
Created February 28, 2014 13:56
Show Gist options
  • Save sigmaris/9271509 to your computer and use it in GitHub Desktop.
Save sigmaris/9271509 to your computer and use it in GitHub Desktop.
#!/usr/bin/python
import sys
import subprocess
import Queue
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class QueuingEventHandler(FileSystemEventHandler):
def __init__(self, queue):
super(QueuingEventHandler, self).__init__()
self.queue = queue
def on_any_event(self, event):
try:
self.queue.put_nowait(event)
except Queue.Full:
print("Queue is full, discarding {0}".format(event))
if __name__ == "__main__":
queue = Queue.Queue(maxsize=10)
event_handler = QueuingEventHandler(queue)
observer = Observer()
observer.schedule(event_handler, path='.', recursive=True)
observer.start()
try:
while True:
try:
evt_count = 0
event = queue.get(block=True, timeout=1)
evt_count += 1
try:
for attempt in xrange(10):
event = queue.get(block=True, timeout=1)
evt_count += 1
except Queue.Empty:
pass
subprocess.call(sys.argv[1:])
except Queue.Empty:
pass
except KeyboardInterrupt:
observer.stop()
observer.join()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment