fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-05 02:18:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

#50 removes redis connection argument from pusher process

Browse Source
This commit is contained in:
Ilan Steemers
2015-08-24 10:08:43 +02:00
parent b414b637eb
commit 3e92fddcdd
3 changed files with 9 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/cluster.py
+3 -2
View File
@@ -165,7 +165,7 @@ class Sentinel(object):
return p return p
def spawn_pusher(self): def spawn_pusher(self):
return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key, self.r) return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key)
def spawn_worker(self): def spawn_worker(self):
self.spawn_process(worker, self.task_queue, self.result_queue, Value('f', -1), self.timeout) self.spawn_process(worker, self.task_queue, self.result_queue, Value('f', -1), self.timeout)
@@ -287,7 +287,7 @@ class Sentinel(object):
Stat(self).save() Stat(self).save()
def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client): def pusher(task_queue, event, list_key=Conf.Q_LIST):
""" """
Pulls tasks of the Redis List and puts them in the task queue Pulls tasks of the Redis List and puts them in the task queue
:type task_queue: multiprocessing.Queue :type task_queue: multiprocessing.Queue
@@ -295,6 +295,7 @@ def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client):
:type list_key: str :type list_key: str
""" """
logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid)) logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid))
r = redis_client
while True: while True:
try: try:
task = r.blpop(list_key, 1) task = r.blpop(list_key, 1)
django_q/tests/test_cluster.py
+5 -5
View File
@@ -85,7 +85,7 @@ def test_cluster(r):
event = Event() event = Event()
event.set() event.set()
# Test push # Test push
pusher(task_queue, event, list_key=list_key, r=r) pusher(task_queue, event, list_key=list_key)
assert task_queue.qsize() == 1 assert task_queue.qsize() == 1
assert queue_size(list_key=list_key, r=r) == 0 assert queue_size(list_key=list_key, r=r) == 0
# Test work # Test work
@@ -148,7 +148,7 @@ def test_async(r, admin_user):
stop_event.set() stop_event.set()
# push the tasks # push the tasks
for i in range(task_count): for i in range(task_count):
pusher(task_queue, stop_event, list_key=list_key, r=r) pusher(task_queue, stop_event, list_key=list_key)
assert queue_size(list_key=list_key, r=r) == 0 assert queue_size(list_key=list_key, r=r) == 0
assert task_queue.qsize() == task_count assert task_queue.qsize() == task_count
task_queue.put('STOP') task_queue.put('STOP')
@@ -292,8 +292,8 @@ def test_recycle(r):
task_queue = Queue() task_queue = Queue()
result_queue = Queue() result_queue = Queue()
# push two tasks # push two tasks
pusher(task_queue, stop_event, list_key=list_key, r=r) pusher(task_queue, stop_event, list_key=list_key)
pusher(task_queue, stop_event, list_key=list_key, r=r) pusher(task_queue, stop_event, list_key=list_key)
# worker should exit on recycle # worker should exit on recycle
worker(task_queue, result_queue, Value('f', -1)) worker(task_queue, result_queue, Value('f', -1))
# check if the work has been done # check if the work has been done
@@ -322,7 +322,7 @@ def test_bad_secret(r, monkeypatch):
assert len(stat) == 0 assert len(stat) == 0
assert Stat.get(s.parent_pid, r) is None assert Stat.get(s.parent_pid, r) is None
task_queue = Queue() task_queue = Queue()
pusher(task_queue, stop_event, list_key=list_key, r=r) pusher(task_queue, stop_event, list_key=list_key)
result_queue = Queue() result_queue = Queue()
task_queue.put('STOP') task_queue.put('STOP')
worker(task_queue, result_queue, Value('f', -1), ) worker(task_queue, result_queue, Value('f', -1), )
django_q/tests/test_scheduler.py
+1 -1
View File
@@ -34,7 +34,7 @@ def test_scheduler(r):
stop_event = Event() stop_event = Event()
stop_event.set() stop_event.set()
# push it # push it
pusher(task_queue, stop_event, list_key=list_key, r=r) pusher(task_queue, stop_event, list_key=list_key)
assert task_queue.qsize() == 1 assert task_queue.qsize() == 1
assert queue_size(list_key=list_key, r=r) == 0 assert queue_size(list_key=list_key, r=r) == 0
task_queue.put('STOP') task_queue.put('STOP')