Merge branch 'master' into mcronce-disable_encryption

src/documents/management/commands/document_consumer.py

@@ -1,6 +1,7 @@
 import datetime
 import logging
 import os
+import sys
 import time
 
 from django.conf import settings
@@ -9,6 +10,11 @@ from django.core.management.base import BaseCommand, CommandError
 from ...consumer import Consumer, ConsumerError
 from ...mail import MailFetcher, MailFetcherError
 
+try:
+    from inotify_simple import INotify, flags
+except ImportError:
+    pass
+
 
 class Command(BaseCommand):
     """
@@ -53,13 +59,20 @@ class Command(BaseCommand):
             action="store_true",
             help="Run only once."
         )
+        parser.add_argument(
+            "--no-inotify",
+            action="store_true",
+            help="Don't use inotify, even if it's available."
+        )
 
     def handle(self, *args, **options):
 
         self.verbosity = options["verbosity"]
         directory = options["directory"]
         loop_time = options["loop_time"]
-        mail_delta = datetime.timedelta(minutes=options["mail_delta"])
+        mail_delta = options["mail_delta"] * 60
+        use_inotify = (not options["no_inotify"]
+                       and "inotify_simple" in sys.modules)
 
         try:
             self.file_consumer = Consumer(consume=directory)
@@ -67,39 +80,68 @@ class Command(BaseCommand):
         except (ConsumerError, MailFetcherError) as e:
             raise CommandError(e)
 
-        for path in (self.ORIGINAL_DOCS, self.THUMB_DOCS):
-            try:
-                os.makedirs(path)
-            except FileExistsError:
-                pass
+        for d in (self.ORIGINAL_DOCS, self.THUMB_DOCS):
+            os.makedirs(d, exist_ok=True)
 
         logging.getLogger(__name__).info(
-            "Starting document consumer at {}".format(directory)
+            "Starting document consumer at {}{}".format(
+                directory,
+                " with inotify" if use_inotify else ""
+            )
         )
 
         if options["oneshot"]:
-            self.loop(mail_delta=mail_delta)
+            self.loop_step(mail_delta)
         else:
             try:
-                while True:
-                    self.loop(mail_delta=mail_delta)
-                    time.sleep(loop_time)
-                    if self.verbosity > 1:
-                        print(".", int(time.time()))
+                if use_inotify:
+                    self.loop_inotify(mail_delta)
+                else:
+                    self.loop(loop_time, mail_delta)
             except KeyboardInterrupt:
                 print("Exiting")
 
-    def loop(self, mail_delta):
+    def loop(self, loop_time, mail_delta):
+        while True:
+            start_time = time.time()
+            if self.verbosity > 1:
+                print(".", int(start_time))
+            self.loop_step(mail_delta, start_time)
+            # Sleep until the start of the next loop step
+            time.sleep(max(0, start_time + loop_time - time.time()))
+
+    def loop_step(self, mail_delta, time_now=None):
 
         # Occasionally fetch mail and store it to be consumed on the next loop
         # We fetch email when we first start up so that it is not necessary to
         # wait for 10 minutes after making changes to the config file.
-        delta = self.mail_fetcher.last_checked + mail_delta
-        if self.first_iteration or delta < datetime.datetime.now():
+        next_mail_time = self.mail_fetcher.last_checked + mail_delta
+        if self.first_iteration or time_now > next_mail_time:
             self.first_iteration = False
             self.mail_fetcher.pull()
 
-        # Consume whatever files we can.
-        # We have to run twice as the first run checks for file readiness
-        for i in range(2):
-            self.file_consumer.run()
+        self.file_consumer.consume_new_files()
+
+    def loop_inotify(self, mail_delta):
+        directory = self.file_consumer.consume
+        inotify = INotify()
+        inotify.add_watch(directory, flags.CLOSE_WRITE | flags.MOVED_TO)
+
+        # Run initial mail fetch and consume all currently existing documents
+        self.loop_step(mail_delta)
+        next_mail_time = self.mail_fetcher.last_checked + mail_delta
+
+        while True:
+            # Consume documents until next_mail_time
+            while True:
+                delta = next_mail_time - time.time()
+                if delta > 0:
+                    for event in inotify.read(timeout=delta):
+                        file = os.path.join(directory, event.name)
+                        if os.path.isfile(file):
+                            self.file_consumer.try_consume_file(file)
+                else:
+                    break
+
+            self.mail_fetcher.pull()
+            next_mail_time = self.mail_fetcher.last_checked + mail_delta