mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-08 06:58:11 +08:00
Adds Fastack option to Disque broker
This commit is contained in:
@@ -28,7 +28,8 @@ class Disque(Broker):
|
|||||||
return self.connection.execute_command('QLEN {}'.format(self.list_key))
|
return self.connection.execute_command('QLEN {}'.format(self.list_key))
|
||||||
|
|
||||||
def acknowledge(self, task_id):
|
def acknowledge(self, task_id):
|
||||||
return self.connection.execute_command('ACKJOB {}'.format(task_id))
|
command = 'FASTACK' if Conf.DISQUE_FASTACK else 'ACKJOB'
|
||||||
|
return self.connection.execute_command('{} {}'.format(command,task_id))
|
||||||
|
|
||||||
def ping(self):
|
def ping(self):
|
||||||
return self.connection.execute_command('HELLO')[0] > 0
|
return self.connection.execute_command('HELLO')[0] > 0
|
||||||
|
|||||||
@@ -34,9 +34,13 @@ class Conf(object):
|
|||||||
|
|
||||||
# Disque broker
|
# Disque broker
|
||||||
DISQUE_NODES = conf.get('disque_nodes', None)
|
DISQUE_NODES = conf.get('disque_nodes', None)
|
||||||
|
|
||||||
# Optional Authentication
|
# Optional Authentication
|
||||||
DISQUE_AUTH = conf.get('disque_auth', None)
|
DISQUE_AUTH = conf.get('disque_auth', None)
|
||||||
|
|
||||||
|
# Optional Fast acknowledge
|
||||||
|
DISQUE_FASTACK = conf.get('disque_fastack', False)
|
||||||
|
|
||||||
# IronMQ broker
|
# IronMQ broker
|
||||||
IRON_MQ = conf.get('iron_mq', None)
|
IRON_MQ = conf.get('iron_mq', None)
|
||||||
|
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ def test_redis():
|
|||||||
broker = get_broker()
|
broker = get_broker()
|
||||||
assert broker.ping() is True
|
assert broker.ping() is True
|
||||||
assert broker.info() is not None
|
assert broker.info() is not None
|
||||||
Conf.REDIS = {'host': '127.0.0.1', 'port': 7712}
|
Conf.REDIS = {'host': '127.0.0.1', 'port': 7799}
|
||||||
broker = get_broker()
|
broker = get_broker()
|
||||||
with pytest.raises(Exception):
|
with pytest.raises(Exception):
|
||||||
broker.ping()
|
broker.ping()
|
||||||
@@ -66,10 +66,6 @@ def test_disque():
|
|||||||
broker.acknowledge(task[0])
|
broker.acknowledge(task[0])
|
||||||
sleep(1.5)
|
sleep(1.5)
|
||||||
assert broker.queue_size() == 0
|
assert broker.queue_size() == 0
|
||||||
# connection test
|
|
||||||
Conf.DISQUE_NODES = ['127.0.0.1:7712', '127.0.0.1:7713']
|
|
||||||
with pytest.raises(redis.exceptions.ConnectionError):
|
|
||||||
broker.get_connection()
|
|
||||||
# delete job
|
# delete job
|
||||||
task_id = broker.enqueue('test')
|
task_id = broker.enqueue('test')
|
||||||
broker.delete(task_id)
|
broker.delete(task_id)
|
||||||
@@ -81,6 +77,7 @@ def test_disque():
|
|||||||
for i in range(5):
|
for i in range(5):
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
Conf.BULK = 5
|
Conf.BULK = 5
|
||||||
|
Conf.DISQUE_FASTACK = True
|
||||||
for i in range(5):
|
for i in range(5):
|
||||||
task = broker.dequeue()
|
task = broker.dequeue()
|
||||||
assert task is not None
|
assert task is not None
|
||||||
@@ -92,13 +89,20 @@ def test_disque():
|
|||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
assert broker.queue_size() == 0
|
assert broker.queue_size() == 0
|
||||||
|
# connection test
|
||||||
|
Conf.DISQUE_NODES = ['127.0.0.1:7798', '127.0.0.1:7799']
|
||||||
|
with pytest.raises(redis.exceptions.ConnectionError):
|
||||||
|
broker.get_connection()
|
||||||
# back to django-redis
|
# back to django-redis
|
||||||
Conf.DISQUE_NODES = None
|
Conf.DISQUE_NODES = None
|
||||||
|
Conf.DISQUE_FASTACK = False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(not os.getenv('IRON_MQ_TOKEN'),
|
@pytest.mark.skipif(not os.getenv('IRON_MQ_TOKEN'),
|
||||||
reason="requires IronMQ credentials")
|
reason="requires IronMQ credentials")
|
||||||
def test_ironmq():
|
def test_ironmq():
|
||||||
|
Conf.DISQUE_NODES = None
|
||||||
|
Conf.SQS = None
|
||||||
Conf.IRON_MQ = {'token': os.getenv('IRON_MQ_TOKEN'),
|
Conf.IRON_MQ = {'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
|
||||||
@@ -157,6 +161,8 @@ def test_ironmq():
|
|||||||
@pytest.mark.skipif(not os.getenv('AWS_ACCESS_KEY_ID'),
|
@pytest.mark.skipif(not os.getenv('AWS_ACCESS_KEY_ID'),
|
||||||
reason="requires AWS credentials")
|
reason="requires AWS credentials")
|
||||||
def test_sqs():
|
def test_sqs():
|
||||||
|
Conf.IRON_MQ = None
|
||||||
|
Conf.DISQUE_NODES = None
|
||||||
Conf.SQS = {'aws_region': os.getenv('AWS_REGION'),
|
Conf.SQS = {'aws_region': os.getenv('AWS_REGION'),
|
||||||
'aws_access_key_id': os.getenv('AWS_ACCESS_KEY_ID'),
|
'aws_access_key_id': os.getenv('AWS_ACCESS_KEY_ID'),
|
||||||
'aws_secret_access_key': os.getenv('AWS_SECRET_ACCESS_KEY')}
|
'aws_secret_access_key': os.getenv('AWS_SECRET_ACCESS_KEY')}
|
||||||
|
|||||||
Reference in New Issue
Block a user