Pika connection parameters can be passed through CELERYT_PIKA_OPTIONS…

… option

tcelery/__init__.py

@@ -23,6 +23,9 @@ def setup_nonblocking_producer(celery_app=None, io_loop=None,
 
     def connect():
         broker_url = celery_app.connection().as_uri(include_password=True)
-        NonBlockingTaskProducer.conn_pool.connect(broker_url, on_ready)
+        options = celery_app.conf.get('CELERYT_PIKA_OPTIONS', {})
+        NonBlockingTaskProducer.conn_pool.connect(broker_url,
+                                                  options=options,
+                                                  callback=on_ready)
 
     io_loop.add_callback(connect)

tcelery/connection.py

@@ -17,15 +17,19 @@ def __init__(self):
         self.channel = None
         self.connection = None
 
-    def connect(self, url, callback=None):
+    def connect(self, url, options=None, callback=None):
         purl = urlparse(url)
         credentials = pika.PlainCredentials(purl.username, purl.password)
         virtual_host = purl.path[1:]
         host = purl.hostname
         port = purl.port
 
-        params = pika.ConnectionParameters(host, port, virtual_host,
-                                           credentials)
+        options = options or {}
+        options = {k.lstrip('DEFAULT_').lower(): v for k, v in options.items()}
+        options.update(host=host, port=port, virtual_host=virtual_host,
+                       credentials=credentials)
+
+        params = pika.ConnectionParameters(**options)
         self.connection = TornadoConnection(
                 params, on_open_callback=partial(self.on_connect, callback))
         self.connection.add_on_close_callback(self.on_closed)
@@ -81,11 +85,12 @@ def __init__(self, limit):
         self._connections = []
         self._connection = None
 
-    def connect(self, broker_url, callback=None):
+    def connect(self, broker_url, options=None, callback=None):
         self._on_ready = callback
         for _ in range(self._limit):
             conn = Connection()
-            conn.connect(broker_url, partial(self._on_connect, conn))
+            conn.connect(broker_url, options=options,
+                         callback=partial(self._on_connect, conn))
 
     def _on_connect(self, connection):
         self._connections.append(connection)