From 8543c5b328a36800d52fc824aeb92f556764360a Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Fri, 11 Sep 2015 17:58:16 +0200 Subject: [PATCH] Adds Fastack option to Disque broker --- django_q/brokers/disque.py | 3 ++- django_q/conf.py | 4 ++++ django_q/tests/test_brokers.py | 16 +++++++++++----- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/django_q/brokers/disque.py b/django_q/brokers/disque.py index bd9c20b..fb0edd2 100644 --- a/django_q/brokers/disque.py +++ b/django_q/brokers/disque.py @@ -28,7 +28,8 @@ class Disque(Broker): return self.connection.execute_command('QLEN {}'.format(self.list_key)) 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): return self.connection.execute_command('HELLO')[0] > 0 diff --git a/django_q/conf.py b/django_q/conf.py index c723fac..3984a28 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -34,9 +34,13 @@ class Conf(object): # Disque broker DISQUE_NODES = conf.get('disque_nodes', None) + # Optional Authentication DISQUE_AUTH = conf.get('disque_auth', None) + # Optional Fast acknowledge + DISQUE_FASTACK = conf.get('disque_fastack', False) + # IronMQ broker IRON_MQ = conf.get('iron_mq', None) diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py index 34f2e62..96c07f8 100644 --- a/django_q/tests/test_brokers.py +++ b/django_q/tests/test_brokers.py @@ -29,7 +29,7 @@ def test_redis(): broker = get_broker() assert broker.ping() is True 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() with pytest.raises(Exception): broker.ping() @@ -66,10 +66,6 @@ def test_disque(): broker.acknowledge(task[0]) sleep(1.5) 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 task_id = broker.enqueue('test') broker.delete(task_id) @@ -81,6 +77,7 @@ def test_disque(): for i in range(5): broker.enqueue('test') Conf.BULK = 5 + Conf.DISQUE_FASTACK = True for i in range(5): task = broker.dequeue() assert task is not None @@ -92,13 +89,20 @@ def test_disque(): broker.enqueue('test') broker.delete_queue() 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 Conf.DISQUE_NODES = None + Conf.DISQUE_FASTACK = False @pytest.mark.skipif(not os.getenv('IRON_MQ_TOKEN'), reason="requires IronMQ credentials") def test_ironmq(): + Conf.DISQUE_NODES = None + Conf.SQS = None Conf.IRON_MQ = {'token': os.getenv('IRON_MQ_TOKEN'), 'project_id': os.getenv('IRON_MQ_PROJECT_ID')} # check broker @@ -157,6 +161,8 @@ def test_ironmq(): @pytest.mark.skipif(not os.getenv('AWS_ACCESS_KEY_ID'), reason="requires AWS credentials") def test_sqs(): + Conf.IRON_MQ = None + Conf.DISQUE_NODES = None Conf.SQS = {'aws_region': os.getenv('AWS_REGION'), 'aws_access_key_id': os.getenv('AWS_ACCESS_KEY_ID'), 'aws_secret_access_key': os.getenv('AWS_SECRET_ACCESS_KEY')}