Merge pull request #141 from Koed00/dev

Adds custom broker setting

django_q/brokers/__init__.py

@@ -1,6 +1,9 @@
-from django_q.conf import Conf
+import importlib
+
 from django.core.cache import caches, InvalidCacheBackendError
 
+from django_q.conf import Conf
+
 
 class Broker(object):
     def __init__(self, list_key=Conf.PREFIX):
@@ -155,19 +158,29 @@ def get_broker(list_key=Conf.PREFIX):
     :type list_key: str
     :return:
     """
+    # custom
+    if Conf.BROKER_CLASS:
+        module, func = Conf.BROKER_CLASS.rsplit('.', 1)
+        m = importlib.import_module(module)
+        broker = getattr(m, func)
+        return broker(list_key=list)
     # disque
-    if Conf.DISQUE_NODES:
+    elif Conf.DISQUE_NODES:
         from brokers import disque
         return disque.Disque(list_key=list_key)
+    # Iron MQ
     elif Conf.IRON_MQ:
         from brokers import ironmq
         return ironmq.IronMQBroker(list_key=list_key)
+    # SQS
     elif Conf.SQS:
         from brokers import aws_sqs
         return aws_sqs.Sqs(list_key=list_key)
+    # ORM
     elif Conf.ORM:
         from brokers import orm
         return orm.ORM(list_key=list_key)
+    # Mongo
     elif Conf.MONGO:
         from brokers import mongo
         return mongo.Mongo(list_key=list_key)

django_q/conf.py

@@ -1,4 +1,5 @@
 import logging
+from copy import deepcopy
 from signal import signal
 from multiprocessing import cpu_count, Queue
 
@@ -50,6 +51,9 @@ class Conf(object):
     # ORM broker
     ORM = conf.get('orm', None)
 
+    # Custom broker class
+    BROKER_CLASS = conf.get('broker_class', None)
+
     # Database Poll
     POLL = conf.get('poll', 0.2)
 
@@ -169,7 +173,7 @@ if not logger.handlers:
 
 # rollbar
 if Conf.ROLLBAR:
-    rollbar_conf = Conf.ROLLBAR
+    rollbar_conf = deepcopy(Conf.ROLLBAR)
     try:
         import rollbar
         rollbar.init(rollbar_conf.pop('access_token'), environment=rollbar_conf.pop('environment'), **rollbar_conf)
@@ -180,9 +184,6 @@ else:
     rollbar = None
 
 
-
-
-
 # get parent pid compatibility
 def get_ppid():
     if hasattr(os, 'getppid'):

django_q/tests/test_brokers.py

@@ -1,9 +1,11 @@
-from time import sleep
-import pytest
 import os
+from time import sleep
+
+import pytest
 import redis
-from django_q.conf import Conf
+
 from django_q.brokers import get_broker, Broker
+from django_q.conf import Conf
 from django_q.humanhash import uuid
 
 
@@ -47,6 +49,15 @@ def test_redis():
     Conf.DJANGO_REDIS = 'default'
 
 
+def test_custom():
+    Conf.BROKER_CLASS = 'brokers.redis_broker.Redis'
+    broker = get_broker()
+    assert broker.ping() is True
+    assert broker.info() is not None
+    assert broker.__class__.__name__ == 'Redis'
+    Conf.BROKER_CLASS = None
+
+
 def test_disque():
     Conf.DISQUE_NODES = ['127.0.0.1:7711']
     # check broker
@@ -132,15 +143,15 @@ def test_ironmq():
     broker.acknowledge(task[0])
     assert broker.dequeue() is None
     # Retry test
-    #Conf.RETRY = 1
+    # Conf.RETRY = 1
-    #broker.enqueue('test')
+    # broker.enqueue('test')
-    #assert broker.dequeue() is not None
-    #sleep(3)
     # assert broker.dequeue() is not None
-    #task = broker.dequeue()[0]
+    # sleep(3)
-    #assert len(task) > 0
+    # assert broker.dequeue() is not None
-    #broker.acknowledge(task[0])
+    # task = broker.dequeue()[0]
-    #sleep(3)
+    # assert len(task) > 0
+    # broker.acknowledge(task[0])
+    # sleep(3)
     # delete job
     task_id = broker.enqueue('test')
     broker.delete(task_id)

docs/brokers.rst

@@ -103,6 +103,36 @@ However for a medium message rate and scheduled tasks, this is the most convenie
 * Queue editable in Django Admin
 * See the :ref:`orm_configuration` configuration on how to set it up.
 
+
+
+Custom Broker
+-------------
+You can override the :class:`Broker` or any of its existing derived broker types.
+
+.. code-block:: python
+
+    # example Custom broker.py
+    from django_q.brokers import Broker
+
+    class CustomBroker(Broker):
+        def info(self):
+            return 'My Custom Broker'
+
+Using the :ref:`broker_class` configuration setting you can then instruct Django Q to use this instead of one of the existing brokers:
+
+.. code-block:: python
+
+    # example Custom broker class connection
+
+    Q_CLUSTER = {
+        'name': 'Custom',
+        'workers': 8,
+        'timeout': 60,
+        'broker_class: 'myapp.broker.CustomBroker'
+    }
+
+If you do write a custom broker for one of the many message queueing servers out there we don't support yet, please consider contributing it to the project.
+
 Reference
 ---------
 The :class:`Broker` class is used internally to communicate with the different types of brokers.

docs/configure.rst

@@ -292,6 +292,23 @@ mongo_db
 When using the MongoDB broker you can optionally provide a database name to use for the queues.
 Defaults to default database if available, otherwise ``django-q``
 
+.. _broker_class:
+
+broker_class
+~~~~~~~~~~~~
+You can use a custom broker class for your cluster workers::
+
+    # example Custom broker class connection
+
+    Q_CLUSTER = {
+        'name': 'Custom',
+        'workers': 8,
+        'timeout': 60,
+        'broker_class: 'myapp.broker.CustomBroker'
+    }
+
+Make sure your ``CustomBroker`` class inherits from either the base :class:`Broker` class or one of its children.
+
 .. _bulk:
 
 bulk