diff --git a/django_q/cluster.py b/django_q/cluster.py index c28c4ad..5faf7d5 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -37,8 +37,8 @@ from django_q.brokers import get_broker class Cluster(object): - def __init__(self, broker=get_broker()): - self.broker = broker + def __init__(self, broker=None): + self.broker = broker or get_broker() self.sentinel = None self.stop_event = None self.start_event = None @@ -106,14 +106,14 @@ class Cluster(object): class Sentinel(object): - def __init__(self, stop_event, start_event, broker=get_broker(), timeout=Conf.TIMEOUT, start=True): + def __init__(self, stop_event, start_event, broker=None, timeout=Conf.TIMEOUT, start=True): # Make sure we catch signals for the pool signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGTERM, signal.SIG_DFL) self.pid = current_process().pid self.parent_pid = get_ppid() self.name = current_process().name - self.broker = broker + self.broker = broker or get_broker() self.reincarnations = 0 self.tob = timezone.now() self.stop_event = stop_event @@ -287,12 +287,14 @@ class Sentinel(object): Stat(self).save() -def pusher(task_queue, event, broker=get_broker()): +def pusher(task_queue, event, broker=None): """ Pulls tasks of the Redis List and puts them in the task queue :type task_queue: multiprocessing.Queue :type event: multiprocessing.Event """ + if not broker: + broker = get_broker() logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid)) while True: try: @@ -318,11 +320,13 @@ def pusher(task_queue, event, broker=get_broker()): logger.info(_("{} stopped pushing tasks").format(current_process().name)) -def monitor(result_queue, broker=get_broker()): +def monitor(result_queue, broker=None): """ Gets finished tasks from the result queue and saves them to Django :type result_queue: multiprocessing.Queue """ + if not broker: + broker = get_broker() name = current_process().name logger.info(_("{} monitoring at {}").format(name, current_process().pid)) db.close_old_connections() @@ -413,10 +417,12 @@ def save_task(task): logger.error(e) -def scheduler(broker=get_broker()): +def scheduler(broker=None): """ Creates a task from a schedule at the scheduled time and schedules next run """ + if not broker: + broker = get_broker() try: for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()): args = () diff --git a/django_q/monitor.py b/django_q/monitor.py index c978607..db0d167 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -10,13 +10,15 @@ from django.utils import timezone from django.utils.translation import ugettext as _ # local -from django_q.conf import Conf, redis_client +from django_q.conf import Conf from django_q.status import Stat from django_q.brokers import get_broker from django_q import models -def monitor(run_once=False, broker=get_broker()): +def monitor(run_once=False, broker=None): + if not broker: + broker = get_broker() term = Terminal() broker.ping() with term.fullscreen(), term.hidden_cursor(), term.cbreak(): @@ -85,7 +87,9 @@ def monitor(run_once=False, broker=get_broker()): val = term.inkey(timeout=1) -def info(broker=get_broker()): +def info(broker=None): + if not broker: + broker = get_broker() term = Terminal() broker.ping() stat = Stat.get_all(broker=broker) diff --git a/django_q/status.py b/django_q/status.py index 7c4aa17..db03825 100644 --- a/django_q/status.py +++ b/django_q/status.py @@ -72,12 +72,14 @@ class Stat(Status): return self.done_q_size + self.task_q_size == 0 @staticmethod - def get(cluster_id, broker=get_broker()): + def get(cluster_id, broker=None): """ gets the current status for the cluster :param cluster_id: id of the cluster :return: Stat or Status """ + if not broker: + broker = get_broker() pack = broker.get_stat(Stat.get_key(cluster_id)) if pack: try: @@ -87,12 +89,14 @@ class Stat(Status): return Status(cluster_id) @staticmethod - def get_all(broker=get_broker()): + def get_all(broker=None): """ Get the status for all currently running clusters with the same prefix and secret key. :return: list of type Stat """ + if not broker: + broker = get_broker() stats = [] packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or [] for pack in packs: diff --git a/django_q/tasks.py b/django_q/tasks.py index 599078f..17a8952 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -152,16 +152,17 @@ def delete_group(group_id, tasks=False): return Task.delete_group(group_id, tasks) -def queue_size(broker=get_broker()): +def queue_size(broker=None): """ Returns the current queue size. Note that this doesn't count any tasks currently being processed by workers. - :param list_key: optional list key :param broker: optional broker :return: current queue size :rtype: int """ + if not broker: + broker = get_broker() return broker.queue_size()