34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
|
import logging
|
||
|
|
|
||
|
|
from workers.worker import Worker
|
||
|
|
from redlock import RedLock, RedLockError
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
class GlobalWorker(Worker):
|
||
|
|
def __init__(self, app, sleep_period_seconds=3600, lock_ttl=600):
|
||
|
|
super(GlobalWorker, self).__init__()
|
||
|
|
|
||
|
|
worker_name = self.__class__.__name__
|
||
|
|
|
||
|
|
self._redis_info = app.config['USER_EVENTS_REDIS']
|
||
|
|
self._lock_name = '%s.global_lock' % (worker_name)
|
||
|
|
self._lock_ttl = lock_ttl
|
||
|
|
|
||
|
|
self.add_operation(self._prepare_global_work, sleep_period_seconds)
|
||
|
|
|
||
|
|
def _prepare_global_work(self):
|
||
|
|
try:
|
||
|
|
logger.debug('Acquiring lock %s', self._lock_name)
|
||
|
|
with RedLock(self._lock_name, connection_details=[self._redis_info],
|
||
|
|
ttl=self._lock_ttl):
|
||
|
|
logger.debug('Acquired lock %s, performing work', self._lock_name)
|
||
|
|
self.perform_global_work()
|
||
|
|
|
||
|
|
logger.debug('Work complete, releasing lock %s', self._lock_name)
|
||
|
|
except RedLockError:
|
||
|
|
logger.debug('Could not acquire lock %s, going to sleep', self._lock_name)
|
||
|
|
|
||
|
|
def perform_global_work(self):
|
||
|
|
raise NotImplementedError('Workers must implement perform_global_work.')
|