Merge pull request #482 from ihuk/master

Fix for #424

django_q/brokers/__init__.py

@@ -13,6 +13,14 @@ class Broker:
         self.cache = self.get_cache()
         self._info = None
 
+    def __getstate__(self):
+        return self.list_key, self._info
+
+    def __setstate__(self, state):
+        self.list_key, self._info = state
+        self.connection = self.get_connection(self.list_key)
+        self.cache = self.get_cache()
+
     def enqueue(self, task):
         """
         Puts a task onto the queue

django_q/brokers/aws_sqs.py

@@ -10,6 +10,11 @@ class Sqs(Broker):
         super(Sqs, self).__init__(list_key)
         self.queue = self.get_queue()
 
+    def __setstate__(self, state):
+        super(Sqs, self).__setstate__(state)
+        self.sqs = None
+        self.queue = self.get_queue()
+
     def enqueue(self, task):
         response = self.queue.send_message(MessageBody=task)
         return response.get("MessageId")

django_q/brokers/mongo.py

@@ -19,6 +19,10 @@ class Mongo(Broker):
         super(Mongo, self).__init__(list_key)
         self.collection = self.get_collection()
 
+    def __setstate__(self, state):
+        super(Mongo, self).__setstate__(state)
+        self.collection = self.get_collection()
+
     @staticmethod
     def get_connection(list_key: str = Conf.PREFIX) -> MongoClient:
         return MongoClient(**Conf.MONGO)

django_q/cluster.py

@@ -12,7 +12,15 @@ from time import sleep
 import arrow
 
 # Django
-from django import db
+from django import db, core
+from django.apps.registry import apps
+
+try:
+    apps.check_apps_ready()
+except core.exceptions.AppRegistryNotReady:
+    import django
+    django.setup()
+
 from django.conf import settings
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _

django_q/queues.py

@@ -54,9 +54,15 @@ class Queue(multiprocessing.queues.Queue):
             super(Queue, self).__init__(
                 *args, ctx=multiprocessing.get_context(), **kwargs
             )
-
         self.size = SharedCounter(0)
 
+    def __getstate__(self):
+        return super(Queue, self).__getstate__() + (self.size, )
+
+    def __setstate__(self, state):
+        super(Queue, self).__setstate__(state[:-1])
+        self.size = state[-1]
+
     def put(self, *args, **kwargs):
         super(Queue, self).put(*args, **kwargs)
         self.size.increment(1)