Merge pull request #135 from Koed00/dev

Fixes acknowledgement bug for failed tasks

django_q/cluster.py

@@ -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:

requirements.txt

@@ -7,20 +7,21 @@
 arrow==0.7.0
 blessed==1.14.1
 boto3==1.2.3
-botocore==1.3.17          # via boto3
+botocore==1.3.20          # via boto3
 django-picklefield==0.3.2
 django-redis==4.3.0
 docutils==0.12            # via botocore
 future==0.15.2
+futures==3.0.4            # via boto3
 hiredis==0.2.0
 iron-core==1.2.0          # via iron-mq
 iron-mq==0.8
 jmespath==0.9.0           # via boto3, botocore
-psutil==3.3.0
+psutil==3.4.1
 pymongo==3.2
 python-dateutil==2.4.2    # via arrow, botocore, iron-core
 redis==2.10.5
 requests==2.9.1           # via iron-core, rollbar
 rollbar==0.11.1
 six==1.10.0               # via blessed, python-dateutil, rollbar
-wcwidth==0.1.5            # via blessed
+wcwidth==0.1.6            # via blessed