From f75593aebbeb307dfdce2ff85991f2b30e421427 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Fri, 31 Jul 2015 12:29:10 +0200
Subject: [PATCH] Moves unpacking task from Worker to Pusher.

Makes more sense and gives us more options to work with control queues from the pusher in the future. Does not seem to affect performance.
---
 django_q/cluster.py | 16 ++++++++--------
 django_q/tasks.py   | 11 ++++++-----
 2 files changed, 14 insertions(+), 13 deletions(-)

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']