mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-06 09:58:11 +08:00
Adds bulk option to disque broker
adds tests
This commit is contained in:
@@ -11,9 +11,18 @@ class Disque(Broker):
|
|||||||
'ADDJOB {} {} 500 RETRY {}'.format(self.list_key, task, retry)).decode()
|
'ADDJOB {} {} 500 RETRY {}'.format(self.list_key, task, retry)).decode()
|
||||||
|
|
||||||
def dequeue(self):
|
def dequeue(self):
|
||||||
task = self.connection.execute_command('GETJOB TIMEOUT 1000 FROM {}'.format(self.list_key))
|
t = None
|
||||||
if task:
|
if len(self.task_cache) > 0:
|
||||||
return task[0][1].decode(), task[0][2].decode()
|
t = self.task_cache.pop()
|
||||||
|
else:
|
||||||
|
tasks = self.connection.execute_command(
|
||||||
|
'GETJOB COUNT {} TIMEOUT 1000 FROM {}'.format(Conf.BULK, self.list_key))
|
||||||
|
if tasks:
|
||||||
|
t = tasks.pop()
|
||||||
|
if tasks:
|
||||||
|
self.task_cache = tasks
|
||||||
|
if t:
|
||||||
|
return t[1].decode(), t[2].decode()
|
||||||
|
|
||||||
def queue_size(self):
|
def queue_size(self):
|
||||||
return self.connection.execute_command('QLEN {}'.format(self.list_key))
|
return self.connection.execute_command('QLEN {}'.format(self.list_key))
|
||||||
|
|||||||
@@ -76,6 +76,14 @@ def test_disque():
|
|||||||
# fail
|
# fail
|
||||||
task_id = broker.enqueue('test')
|
task_id = broker.enqueue('test')
|
||||||
broker.fail(task_id)
|
broker.fail(task_id)
|
||||||
|
# bulk test
|
||||||
|
for i in range(5):
|
||||||
|
broker.enqueue('test')
|
||||||
|
Conf.BULK = 5
|
||||||
|
for i in range(5):
|
||||||
|
task = broker.dequeue()
|
||||||
|
assert task is not None
|
||||||
|
broker.acknowledge(task[0])
|
||||||
# delete queue
|
# delete queue
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
@@ -89,8 +97,8 @@ def test_disque():
|
|||||||
reason="requires IronMQ credentials")
|
reason="requires IronMQ credentials")
|
||||||
def test_ironmq():
|
def test_ironmq():
|
||||||
Conf.IRON_MQ = {'host': os.getenv('IRON_MQ_HOST'),
|
Conf.IRON_MQ = {'host': os.getenv('IRON_MQ_HOST'),
|
||||||
'token': os.getenv('IRON_MQ_TOKEN'),
|
'token': os.getenv('IRON_MQ_TOKEN'),
|
||||||
'project_id': os.getenv('IRON_MQ_PROJECT_ID')}
|
'project_id': os.getenv('IRON_MQ_PROJECT_ID')}
|
||||||
# check broker
|
# check broker
|
||||||
broker = get_broker(list_key='djangoQ')
|
broker = get_broker(list_key='djangoQ')
|
||||||
assert broker.ping() is True
|
assert broker.ping() is True
|
||||||
@@ -122,11 +130,20 @@ def test_ironmq():
|
|||||||
# fail
|
# fail
|
||||||
task_id = broker.enqueue('test')
|
task_id = broker.enqueue('test')
|
||||||
broker.fail(task_id)
|
broker.fail(task_id)
|
||||||
|
# bulk test
|
||||||
|
for i in range(5):
|
||||||
|
broker.enqueue('test')
|
||||||
|
Conf.BULK = 5
|
||||||
|
for i in range(5):
|
||||||
|
task = broker.dequeue()
|
||||||
|
assert task is not None
|
||||||
|
broker.acknowledge(task[0])
|
||||||
# delete queue
|
# delete queue
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
broker.purge_queue()
|
broker.purge_queue()
|
||||||
assert broker.queue_size() == 0
|
assert broker.queue_size() == 0
|
||||||
|
broker.delete_queue()
|
||||||
# back to django-redis
|
# back to django-redis
|
||||||
Conf.IRON_MQ = None
|
Conf.IRON_MQ = None
|
||||||
Conf.DJANGO_REDIS = 'default'
|
Conf.DJANGO_REDIS = 'default'
|
||||||
|
|||||||
Reference in New Issue
Block a user