diff --git a/django_q/cluster.py b/django_q/cluster.py index e63ed30..e98460b 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -315,7 +315,13 @@ def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client): sleep(10) break if task: - task_queue.put(task[1]) + # unpack the task + try: + task = signing.SignedPackage.loads(task[1]) + except (TypeError, signing.BadSignature) as e: + logger.error(e) + continue + task_queue.put(task) logger.debug(_('queueing from {}').format(list_key)) if event.is_set(): break @@ -351,16 +357,10 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT): db.close_old_connections() task_count = 0 # Start reading the task queue - for pack in iter(task_queue.get, 'STOP'): + for task in iter(task_queue.get, 'STOP'): result = None timer.value = -1 # Idle task_count += 1 - # unpickle the task - try: - task = signing.SignedPackage.loads(pack) - except (TypeError, signing.BadSignature) as e: - logger.error(e) - continue # Get the function from the task logger.info(_('{} processing [{}]').format(name, task['name'])) f = task['func'] diff --git a/django_q/tasks.py b/django_q/tasks.py index 07870d4..abaea1a 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -40,7 +40,7 @@ def async(func, *args, **kwargs): # sign it pack = signing.SignedPackage.dumps(task) if sync: - return _sync(task['id'], pack) + return _sync(pack) # push it redis.rpush(list_key, pack) logger.debug('Pushed {}'.format(tag)) @@ -150,13 +150,14 @@ def delete_group(group_id, tasks=False): return Task.delete_group(group_id, tasks) -def _sync(task_id, pack): +def _sync(pack): """Simulate a package travelling through the cluster.""" task_queue = Queue() result_queue = Queue() - task_queue.put(pack) + task = signing.SignedPackage.loads(pack) + task_queue.put(task) task_queue.put('STOP') - cluster.worker(task_queue, result_queue, Value('b', -1)) + cluster.worker(task_queue, result_queue, Value('f', -1)) result_queue.put('STOP') cluster.monitor(result_queue) - return task_id + return task['id']