mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-09 11:28:11 +08:00
Prevents crash on duplicate acknowledge
This commit is contained in:
@@ -1,3 +1,4 @@
|
|||||||
|
from requests.exceptions import HTTPError
|
||||||
from django_q.conf import Conf
|
from django_q.conf import Conf
|
||||||
from django_q.brokers import Broker
|
from django_q.brokers import Broker
|
||||||
from iron_mq import IronMQ
|
from iron_mq import IronMQ
|
||||||
@@ -31,13 +32,19 @@ class IronMQBroker(Broker):
|
|||||||
return self.connection.size()
|
return self.connection.size()
|
||||||
|
|
||||||
def delete_queue(self):
|
def delete_queue(self):
|
||||||
return self.connection.delete_queue()['msg']
|
try:
|
||||||
|
return self.connection.delete_queue()['msg']
|
||||||
|
except HTTPError:
|
||||||
|
return False
|
||||||
|
|
||||||
def purge_queue(self):
|
def purge_queue(self):
|
||||||
return self.connection.clear()
|
return self.connection.clear()
|
||||||
|
|
||||||
def delete(self, task_id):
|
def delete(self, task_id):
|
||||||
return self.connection.delete(task_id)['msg']
|
try:
|
||||||
|
return self.connection.delete(task_id)['msg']
|
||||||
|
except HTTPError:
|
||||||
|
return False
|
||||||
|
|
||||||
def fail(self, task_id):
|
def fail(self, task_id):
|
||||||
self.delete(task_id)
|
self.delete(task_id)
|
||||||
|
|||||||
@@ -85,6 +85,13 @@ def test_disque():
|
|||||||
task = broker.dequeue()
|
task = broker.dequeue()
|
||||||
assert task is not None
|
assert task is not None
|
||||||
broker.acknowledge(task[0])
|
broker.acknowledge(task[0])
|
||||||
|
# test duplicate acknowledge
|
||||||
|
# broker.acknowledge(task[0])
|
||||||
|
#
|
||||||
|
# this crashes Disque when followed by a JSCAN
|
||||||
|
# https://github.com/antirez/disque/issues/113
|
||||||
|
# confirmed fix and merge is on the way
|
||||||
|
#
|
||||||
# delete queue
|
# delete queue
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
@@ -139,6 +146,8 @@ def test_ironmq():
|
|||||||
task = broker.dequeue()
|
task = broker.dequeue()
|
||||||
assert task is not None
|
assert task is not None
|
||||||
broker.acknowledge(task[0])
|
broker.acknowledge(task[0])
|
||||||
|
# duplicate acknowledge
|
||||||
|
broker.acknowledge(task[0])
|
||||||
# delete queue
|
# delete queue
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
@@ -195,6 +204,8 @@ def test_sqs():
|
|||||||
task = broker.dequeue()
|
task = broker.dequeue()
|
||||||
assert task is not None
|
assert task is not None
|
||||||
broker.acknowledge(task[0])
|
broker.acknowledge(task[0])
|
||||||
|
# duplicate acknowledge
|
||||||
|
broker.acknowledge(task[0])
|
||||||
# delete queue
|
# delete queue
|
||||||
broker.enqueue('test')
|
broker.enqueue('test')
|
||||||
broker.purge_queue()
|
broker.purge_queue()
|
||||||
|
|||||||
Reference in New Issue
Block a user