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']),
  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 _worker(self):
  41         # start triggers
  42
  43         while not self._stop:
  44             for source in self._sources:
  45                 try:
  46                     if source.trigger.is_triggering():
  47                         logging.info("Got source trigger for bunch %s.", self._id)
  48                         source.trigger.reset()
  49                         source.migrator.migrate()
  50                         if source.cleanup:
  51                             source.migrator.cleanup()
  52                 except Exception as e:
  53                     logging.error("Error migrating source for bunch %s: %s",
  54                                   self._id, str(e))
  55             time.sleep(10.0)
  56
  57         # stop triggers