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, idx, trigger, migrator, cleanup):
  11             self.idx = idx
  12             self.trigger = trigger
  13             self.migrator = migrator
  14             self.cleanup = cleanup
  15
  16     def __init__(self, idx, cfg):
  17         self._idx = idx
  18         source_idx = 1
  19         self._sources = []
  20         for src_dir_cfg in cfg['src_dirs']:
  21             self._sources.append(
  22                 Bunch.TriggeredSource(
  23                     source_idx,
  24                     DirTrigger(src_dir_cfg['path'], src_dir_cfg['cool_time'],
  25                                src_dir_cfg['max_time']),
  26                     Migrator(src_dir_cfg['path'], cfg['dst_dir']['path']),
  27                     src_dir_cfg['cleanup']
  28                 )
  29             )
  30             source_idx += 1
  31         logging.info("Created bunch #%d with %d sources.", self._idx, len(self._sources))
  32
  33     def start(self):
  34         logging.info("Starting bunch #%d...", self._idx)
  35         self._stop = False
  36         self._worker_thread = threading.Thread(target=self._worker)
  37         self._worker_thread.start()
  38
  39     def stop(self):
  40         logging.info("Stopping bunch #%d...", self._idx)
  41         self._stop = True
  42         self._worker_thread.join()
  43         logging.info("Stopped bunch #%d.", self._idx)
  44
  45     def is_running(self):
  46         return self._worker_thread.is_alive()
  47
  48     def _worker(self):
  49         for src in self._sources:
  50             src.trigger.start()
  51
  52         while not self._stop:
  53             for src in self._sources:
  54                 try:
  55                     if src.trigger.is_triggering():
  56                         logging.info("Got trigger for source #%d.%d.", self._idx, src.idx)
  57                         before = time.time()
  58                         src.trigger.reset()
  59                         src.migrator.migrate(src.cleanup)
  60                         if src.cleanup:
  61                             src.migrator.cleanup()
  62                         logging.info("Migration took %.1fs for source #%d.%d.",
  63                                      time.time() - before, self._idx, src.idx)
  64                 except Exception as e:
  65                     logging.error("Error migrating source #%d.%d: %s",
  66                                   self._idx, src.idx, str(e))
  67             time.sleep(10.0)
  68
  69         for src in self._sources:
  70             src.trigger.stop()