fix recreate binding on failure (#1516)

src/socketio/async_aiopika_manager.py

@@ -101,26 +101,26 @@ class AsyncAioPikaManager(AsyncPubSubManager):  # pragma: no cover
                 raise asyncio.CancelledError()
 
     async def _listen(self):
-        async with (await self._connection()) as connection:
+        retry_sleep = 1
-            channel = await self._channel(connection)
+        while True:
-            await channel.set_qos(prefetch_count=1)
+            try:
-            exchange = await self._exchange(channel)
+                async with (await self._connection()) as connection:
-            queue = await self._queue(channel, exchange)
+                    channel = await self._channel(connection)
-
+                    await channel.set_qos(prefetch_count=1)
-            retry_sleep = 1
+                    exchange = await self._exchange(channel)
-            while True:
+                    queue = await self._queue(channel, exchange)
-                try:
+
                     async with queue.iterator() as queue_iter:
                         async for message in queue_iter:
                             async with message.process():
                                 yield message.body
                                 retry_sleep = 1
-                except aio_pika.AMQPException:
+            except aio_pika.AMQPException:
-                    self._get_logger().error(
+                self._get_logger().error(
-                        'Cannot receive from rabbitmq... '
+                    'Cannot receive from rabbitmq... '
-                        'retrying in {} secs'.format(retry_sleep))
+                    'retrying in {} secs'.format(retry_sleep))
-                    await asyncio.sleep(retry_sleep)
+                await asyncio.sleep(retry_sleep)
-                    retry_sleep = min(retry_sleep * 2, 60)
+                retry_sleep = min(retry_sleep * 2, 60)
-                except aio_pika.exceptions.ChannelInvalidStateError:
+            except aio_pika.exceptions.ChannelInvalidStateError:
-                    # aio_pika raises this exception when the task is cancelled
+                # aio_pika raises this exception when the task is cancelled
-                    raise asyncio.CancelledError()
+                raise asyncio.CancelledError()

src/socketio/kombu_manager.py

@@ -115,10 +115,10 @@ class KombuManager(PubSubManager):  # pragma: no cover
                     break
 
     def _listen(self):
-        reader_queue = self._queue()
         retry_sleep = 1
         while True:
             try:
+                reader_queue = self._queue()
                 with self._connection() as connection:
                     with connection.SimpleQueue(reader_queue) as queue:
                         while True: