mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-04 17:58:11 +08:00
Move timeout function from ORM broker into class to allow easy customization (#274)
* move timeout function to class to allow customization * format * fix format
This commit is contained in:
@@ -10,10 +10,6 @@ from django_q.conf import Conf, logger
|
|||||||
from django_q.models import OrmQ
|
from django_q.models import OrmQ
|
||||||
|
|
||||||
|
|
||||||
def _timeout():
|
|
||||||
return timezone.now() + timedelta(seconds=Conf.RETRY)
|
|
||||||
|
|
||||||
|
|
||||||
class ORM(Broker):
|
class ORM(Broker):
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_connection(list_key: str = None):
|
def get_connection(list_key: str = None):
|
||||||
@@ -28,6 +24,9 @@ class ORM(Broker):
|
|||||||
logger.debug("Broker in an atomic transaction")
|
logger.debug("Broker in an atomic transaction")
|
||||||
return OrmQ.objects.using(Conf.ORM)
|
return OrmQ.objects.using(Conf.ORM)
|
||||||
|
|
||||||
|
def timeout(self, task):
|
||||||
|
return timezone.now() + timedelta(seconds=Conf.RETRY)
|
||||||
|
|
||||||
def queue_size(self) -> int:
|
def queue_size(self) -> int:
|
||||||
return (
|
return (
|
||||||
self.get_connection()
|
self.get_connection()
|
||||||
@@ -75,7 +74,7 @@ class ORM(Broker):
|
|||||||
if (
|
if (
|
||||||
self.get_connection()
|
self.get_connection()
|
||||||
.filter(id=task.id, lock=task.lock)
|
.filter(id=task.id, lock=task.lock)
|
||||||
.update(lock=_timeout())
|
.update(lock=self.timeout(task))
|
||||||
):
|
):
|
||||||
task_list.append((task.pk, task.payload))
|
task_list.append((task.pk, task.payload))
|
||||||
# else don't process, as another cluster has been faster than us on
|
# else don't process, as another cluster has been faster than us on
|
||||||
|
|||||||
Reference in New Issue
Block a user