Fix connection leaks in broker and transport access

flower/api/tasks.py

@@ -396,7 +396,9 @@ async def get(self):
         if app.transport == 'amqp' and app.options.broker_api:
             http_api = app.options.broker_api
 
-        broker = Broker(app.capp.connection().as_uri(include_password=True),
+        with app.capp.connection() as conn:
+            broker_uri = conn.as_uri(include_password=True)
+        broker = Broker(broker_uri,
                         http_api=http_api, broker_options=self.capp.conf.broker_transport_options,
                         broker_use_ssl=self.capp.conf.broker_use_ssl)
 

flower/app.py

@@ -64,6 +64,7 @@ def __init__(self, options=None, capp=None, events=None,
             max_workers_in_memory=self.options.max_workers,
             max_tasks_in_memory=self.options.max_tasks)
         self.started = False
+        self._transport = None
 
     def start(self):
         self.events.start()
@@ -93,7 +94,10 @@ def stop(self):
 
     @property
     def transport(self):
-        return getattr(self.capp.connection().transport, 'driver_type', None)
+        if self._transport is None:
+            with self.capp.connection() as conn:
+                self._transport = getattr(conn.transport, 'driver_type', None)
+        return self._transport
 
     @property
     def workers(self):

flower/command.py

@@ -173,7 +173,8 @@ def print_banner(app, ssl):
     else:
         logger.info("Visit me via unix socket file: %s", options.unix_socket)
 
-    logger.info('Broker: %s', app.connection().as_uri())
+    with app.connection() as conn:
+        logger.info('Broker: %s', conn.as_uri())
     logger.info(
         'Registered tasks: \n%s',
         pformat(sorted(app.tasks.keys()))

flower/views/broker.py

@@ -17,8 +17,12 @@ async def get(self):
         if app.transport == 'amqp' and app.options.broker_api:
             http_api = app.options.broker_api
 
+        with app.capp.connection(connect_timeout=1.0) as conn:
+            broker_uri = conn.as_uri(include_password=True)
+            broker_url = conn.as_uri()
+
         try:
-            broker = Broker(app.capp.connection(connect_timeout=1.0).as_uri(include_password=True),
+            broker = Broker(broker_uri,
                             http_api=http_api, broker_options=self.capp.conf.broker_transport_options,
                             broker_use_ssl=self.capp.conf.broker_use_ssl)
         except NotImplementedError as exc:
@@ -31,5 +35,5 @@ async def get(self):
             logger.error("Unable to get queues: '%s'", e)
 
         self.render("broker.html",
-                    broker_url=app.capp.connection().as_uri(),
+                    broker_url=broker_url,
                     queues=queues)

flower/views/workers.py

@@ -69,9 +69,11 @@ async def get(self):
         if json:
             self.write(dict(data=list(workers.values())))
         else:
+            with self.application.capp.connection() as conn:
+                broker_url = conn.as_uri()
             self.render("workers.html",
                         workers=workers,
-                        broker=self.application.capp.connection().as_uri(),
+                        broker=broker_url,
                         autorefresh=1 if self.application.options.auto_refresh else 0)
 
     @classmethod