mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-27 20:58:12 +08:00
Ilan Steemers
3f75e7fa05
Both brokers are not atomic and cause all kinds of problems during testing. Needs more time. Meanwhile I want to release the pluggable broker backend.
62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
import random
|
|
import redis
|
|
from django_q.brokers import Broker
|
|
from django_q.conf import Conf
|
|
|
|
|
|
class Disque(Broker):
|
|
def enqueue(self, task):
|
|
retry = Conf.RETRY if Conf.RETRY > 0 else '{} REPLICATE 1'.format(Conf.RETRY)
|
|
return self.connection.execute_command(
|
|
'ADDJOB {} {} 500 RETRY {}'.format(self.list_key, task, retry)).decode()
|
|
|
|
def dequeue(self):
|
|
task = self.connection.execute_command('GETJOB TIMEOUT 1000 FROM {}'.format(self.list_key))
|
|
if task:
|
|
return task[0][1].decode(), task[0][2].decode()
|
|
|
|
def queue_size(self):
|
|
return self.connection.execute_command('QLEN {}'.format(self.list_key))
|
|
|
|
def acknowledge(self, task_id):
|
|
return self.connection.execute_command('ACKJOB {}'.format(task_id))
|
|
|
|
def ping(self):
|
|
return self.connection.execute_command('HELLO')[0] > 0
|
|
|
|
def delete(self, task_id):
|
|
return self.connection.execute_command('DELJOB {}'.format(task_id))
|
|
|
|
def fail(self, task_id):
|
|
return self.delete(task_id)
|
|
|
|
def delete_queue(self):
|
|
jobs = self.connection.execute_command('JSCAN QUEUE {}'.format(self.list_key))[1]
|
|
if jobs:
|
|
job_ids = ' '.join(jid.decode() for jid in jobs)
|
|
self.connection.execute_command('DELJOB {}'.format(job_ids))
|
|
return len(jobs)
|
|
|
|
def info(self):
|
|
info = self.connection.info('server')
|
|
return 'Disque {}'.format(info['disque_version'])
|
|
|
|
@staticmethod
|
|
def get_connection(list_key=Conf.PREFIX):
|
|
# randomize nodes
|
|
random.shuffle(Conf.DISQUE_NODES)
|
|
# find one that works
|
|
for node in Conf.DISQUE_NODES:
|
|
host, port = node.split(':')
|
|
kwargs = {'host': host, 'port': port}
|
|
if Conf.DISQUE_AUTH:
|
|
kwargs['password'] = Conf.DISQUE_AUTH
|
|
redis_client = redis.Redis(**kwargs)
|
|
redis_client.decode_responses = True
|
|
try:
|
|
redis_client.execute_command('HELLO')
|
|
return redis_client
|
|
except redis.exceptions.ConnectionError:
|
|
continue
|
|
raise redis.exceptions.ConnectionError('Could not connect to any Disque nodes')
|