fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-09-16 22:17:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

Only acks task if it's succesful

Browse Source
This commit is contained in:
Ilan Steemers
2016-01-18 10:51:28 +01:00
parent 2e52a04584
commit 0e8ea25396
1 changed files with 14 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
django_q/cluster.py
View File

@@ -1,8 +1,9 @@
# Future
from __future__ import unicode_literals
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
from builtins import range
from future import standard_library
@@ -320,19 +321,21 @@ def monitor(result_queue, broker=None):
name = current_process().name
logger.info(_("{} monitoring at {}").format(name, current_process().pid))
for task in iter(result_queue.get, 'STOP'):
# acknowledge
ack_id = task.pop('ack_id', False)
if ack_id:
broker.acknowledge(ack_id)
# save the result
if task.get('cached', False):
save_cached(task, broker)
else:
save_task(task, broker)
# log the result
# acknowledge and log the result
if task['success']:
# acknowledge
ack_id = task.pop('ack_id', False)
if ack_id:
broker.acknowledge(ack_id)
# log success
logger.info(_("Processed [{}]").format(task['name']))
else:
# log failure
logger.error(_("Failed [{}] - {}").format(task['name'], task['result']))
logger.info(_("{} stopped monitoring results").format(name))
@@ -521,10 +524,11 @@ def scheduler(broker=None):
# log it
if not s.task:
logger.error(
_('{} failed to create a task from schedule [{}]').format(current_process().name, s.name or s.id))
_('{} failed to create a task from schedule [{}]').format(current_process().name,
s.name or s.id))
else:
logger.info(
_('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id))
_('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id))
# default behavior is to delete a ONCE schedule
if s.schedule_type == s.ONCE:
if s.repeats < 0: