fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-05 14:18:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #126 from kdmukai/close_connection

Browse Source 
Fix for stale ORM Broker connections
This commit is contained in:
Ilan Steemers
2016-01-08 10:40:42 +01:00
parent 2e76f9eb45 e4a87e167b
commit b6dd209856
1 changed files with 17 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/brokers/orm.py
+17 -7
View File
@@ -2,10 +2,12 @@ from datetime import timedelta
from time import sleep from time import sleep
from django.utils import timezone from django.utils import timezone
from django import db
from django.db import transaction
from django_q.brokers import Broker from django_q.brokers import Broker
from django_q.models import OrmQ from django_q.models import OrmQ
from django_q.conf import Conf from django_q.conf import Conf, logger
def _timeout(): def _timeout():
@@ -15,16 +17,23 @@ def _timeout():
class ORM(Broker): class ORM(Broker):
@staticmethod @staticmethod
def get_connection(list_key=Conf.PREFIX): def get_connection(list_key=Conf.PREFIX):
if transaction.get_autocommit(): # Only True when not in an atomic block
# Make sure stale connections in the broker thread are explicitly
# closed before attempting DB access.
# logger.debug("Broker thread calling close_old_connections")
db.close_old_connections()
else:
logger.debug("Broker in an atomic transaction")
return OrmQ.objects.using(Conf.ORM) return OrmQ.objects.using(Conf.ORM)
def queue_size(self): def queue_size(self):
return self.connection.filter(key=self.list_key, lock__lte=_timeout()).count() return self.get_connection().filter(key=self.list_key, lock__lte=_timeout()).count()
def lock_size(self): def lock_size(self):
return self.connection.filter(key=self.list_key, lock__gt=_timeout()).count() return self.get_connection().filter(key=self.list_key, lock__gt=_timeout()).count()
def purge_queue(self): def purge_queue(self):
return self.connection.filter(key=self.list_key).delete() return self.get_connection().filter(key=self.list_key).delete()
def ping(self): def ping(self):
return True return True
@@ -38,11 +47,11 @@ class ORM(Broker):
self.delete(task_id) self.delete(task_id)
def enqueue(self, task): def enqueue(self, task):
package = self.connection.create(key=self.list_key, payload=task, lock=_timeout()) package = self.get_connection().create(key=self.list_key, payload=task, lock=_timeout())
return package.pk return package.pk
def dequeue(self): def dequeue(self):
tasks = self.connection.filter(key=self.list_key, lock__lt=_timeout())[0:Conf.BULK] tasks = self.get_connection().filter(key=self.list_key, lock__lt=_timeout())[0:Conf.BULK]
if tasks: if tasks:
task_list = [] task_list = []
lock = timezone.now() lock = timezone.now()
@@ -58,7 +67,8 @@ class ORM(Broker):
return self.purge_queue() return self.purge_queue()
def delete(self, task_id): def delete(self, task_id):
self.connection.filter(pk=task_id).delete() self.get_connection().filter(pk=task_id).delete()
def acknowledge(self, task_id): def acknowledge(self, task_id):
return self.delete(task_id) return self.delete(task_id)