src/bunch.py

   1 import logging
   2 import time
   3 import threading
   4
   5 from dirtrigger import DirTrigger
   6 from migrator import Migrator
   7
   8 class Bunch:
   9     class TriggeredSource:
  10         def __init__(self, trigger, migrator, cleanup):
  11             self.trigger = trigger
  12             self.migrator = migrator
  13             self.cleanup = cleanup
  14
  15     def __init__(self, cfg):
  16         self._id = cfg['id']
  17         self._sources = []
  18         for src_dir_cfg in cfg['src_dirs']:
  19             self._sources.append(
  20                 Bunch.TriggeredSource(
  21                     DirTrigger(src_dir_cfg['path'], src_dir_cfg['cool_time'], src_dir_cfg['max_time']),
  22                     Migrator(src_dir_cfg['path'], cfg['dst_dir']['path']),
  23                     src_dir_cfg['cleanup']
  24                 )
  25             )
  26         logging.info("Created bunch %s with %d sources.", self._id, len(self._sources))
  27
  28     def start(self):
  29         logging.info("Starting bunch %s...", self._id)
  30         self._stop = False
  31         self._worker_thread = threading.Thread(target=self._worker)
  32         self._worker_thread.start()
  33
  34     def stop(self):
  35         logging.info("Stopping bunch %s...", self._id)
  36         self._stop = True
  37         self._worker_thread.join()
  38         logging.info("Stopped bunch %s.", self._id)
  39
  40     def is_running(self):
  41         return self._worker_thread.is_alive()
  42
  43     def _worker(self):
  44         for source in self._sources:
  45             source.trigger.start()
  46
  47         while not self._stop:
  48             for source in self._sources:
  49                 try:
  50                     if source.trigger.is_triggering():
  51                         logging.info("Got source trigger for bunch %s.", self._id)
  52                         source.trigger.reset()
  53                         source.migrator.migrate(source.cleanup)
  54                         if source.cleanup:
  55                             source.migrator.cleanup()
  56                 except Exception as e:
  57                     logging.error("Error migrating source for bunch %s: %s",
  58                                   self._id, str(e))
  59             time.sleep(10.0)
  60
  61         for source in self._sources:
  62             source.trigger.stop()