add the ability to disable RabbitMQ queue durability

awx/main/dispatch/control.py

@@ -4,7 +4,8 @@ import socket
 from django.conf import settings
 
 from awx.main.dispatch import get_local_queuename
-from kombu import Connection, Queue, Exchange, Producer, Consumer
+from awx.main.dispatch.kombu import Connection
+from kombu import Queue, Exchange, Producer, Consumer
 
 logger = logging.getLogger('awx.main.dispatch')
 

awx/main/dispatch/kombu.py

@@ -0,0 +1,42 @@
+from amqp.exceptions import PreconditionFailed
+from django.conf import settings
+from kombu.connection import Connection as KombuConnection
+from kombu.transport import pyamqp
+
+import logging
+
+logger = logging.getLogger('awx.main.dispatch')
+
+
+__all__ = ['Connection']
+
+
+class Connection(KombuConnection):
+
+    def __init__(self, *args, **kwargs):
+        super(Connection, self).__init__(*args, **kwargs)
+        class _Channel(pyamqp.Channel):
+
+            def queue_declare(self, queue, *args, **kwargs):
+                kwargs['durable'] = settings.BROKER_DURABILITY
+                try:
+                    return super(_Channel, self).queue_declare(queue, *args, **kwargs)
+                except PreconditionFailed as e:
+                    if "inequivalent arg 'durable'" in getattr(e, 'reply_text', None):
+                        logger.error(
+                            'queue {} durability is not {}, deleting and recreating'.format(
+
+                                queue,
+                                kwargs['durable']
+                            )
+                        )
+                        self.queue_delete(queue)
+                    return super(_Channel, self).queue_declare(queue, *args, **kwargs)
+
+        class _Connection(pyamqp.Connection):
+            Channel = _Channel
+
+        class _Transport(pyamqp.Transport):
+            Connection = _Connection
+
+        self.transport_cls = _Transport

awx/main/dispatch/publish.py

@@ -4,7 +4,9 @@ import sys
 from uuid import uuid4
 
 from django.conf import settings
-from kombu import Connection, Exchange, Producer
+from kombu import Exchange, Producer
+
+from awx.main.dispatch.kombu import Connection
 
 logger = logging.getLogger('awx.main.dispatch')