log exceptions in channel reader

README.rst

@@ -117,8 +117,9 @@ Simple consumer
         # Declaring queue
         declare_ok = await channel.queue_declare('helo')
         consume_ok = await channel.basic_consume(
-            declare_ok.queue, on_message, no_ack=True
+            declare_ok.queue, on_message, no_ack=True,
         )
+        print("Consumer started")
 
 
     loop = asyncio.get_event_loop()
@@ -563,6 +564,8 @@ RPC server
 .. code-block:: python
 
     import asyncio
+    import functools
+    from contextlib import suppress
     import aiormq
     import aiormq.abc
 
@@ -576,46 +579,55 @@ RPC server
             return fib(n-1) + fib(n-2)
 
 
-    async def on_message(message:aiormq.abc.DeliveredMessage):
-        n = int(message.body.decode())
+    async def on_message(num, message: aiormq.types.DeliveredMessage):
+        try:
+            n = int(message.body.decode())
 
-        print(f" [.] fib({n})")
-        response = str(fib(n)).encode()
+            print(f" [.] fib({n}) from {num} channel")
 
-        await message.channel.basic_publish(
-            response, routing_key=message.header.properties.reply_to,
-            properties=aiormq.spec.Basic.Properties(
-                correlation_id=message.header.properties.correlation_id
-            ),
+            fib_n = await asyncio.run_in_executor(None, fib, n)
+            response = str(fib(n)).encode()
 
-        )
+            await message.channel.basic_publish(
+                response, routing_key=message.header.properties.reply_to,
+                properties=aiormq.spec.Basic.Properties(
+                    correlation_id=message.header.properties.correlation_id
+                ),
 
-        await message.channel.basic_ack(message.delivery.delivery_tag)
-        print('Request complete')
+            )
+        except Exception as exc:
+            print('Message exception:', exc)
+        finally:
+            await message.channel.basic_ack(message.delivery.delivery_tag)
+        print(f'Request from {num} channel complete')
 
 
     async def main():
         # Perform connection
         connection = await aiormq.connect("amqp://guest:guest@localhost/")
 
-        # Creating a channel
-        channel = await connection.channel()
+        for i in range(10):
+            # Creating a channel
+            channel = await connection.channel()
 
-        # Declaring queue
-        declare_ok = await channel.queue_declare('rpc_queue')
+            # Declaring queue
+            declare_ok = await channel.queue_declare('rpc_queue')
 
-        # Start listening the queue with name 'hello'
-        await channel.basic_consume(declare_ok.queue, on_message)
+            # Start listening the queue with name 'hello'
+            # with several consumers simultaneously
+            on_message = functools.partial(on_message, i)
+            await channel.basic_consume(declare_ok.queue, on_message)
+            print(f'Consumer {i} started')
+
+        print(" [x] Awaiting RPC requests")
+        with suppress(asyncio.CancelledError):
+            await connection.closing
 
+        # waiting for the connections closing to release recources
 
     loop = asyncio.get_event_loop()
     loop.create_task(main())
 
-    # we enter a never-ending loop that waits for data
-    # and runs callbacks whenever necessary.
-    print(" [x] Awaiting RPC requests")
-    loop.run_forever()
-
 
 RPC client
 **********

aiormq/channel.py

@@ -279,7 +279,10 @@ async def _on_deliver(self, frame: spec.Basic.Deliver) -> None:
         consumer = self.consumers.get(frame.consumer_tag)
         if consumer is not None:
             # noinspection PyAsyncCall
-            self.create_task(consumer(message))
+            try:
+                self.create_task(consumer(message))
+            except Exception:
+                log.exception('Unhandled consumer exception')
 
     async def _on_get(
         self, frame: Union[spec.Basic.GetOk, spec.Basic.GetEmpty],
@@ -438,7 +441,7 @@ async def _reader(self) -> None:
             except asyncio.CancelledError:
                 return
             except Exception as e:  # pragma: nocover
-                log.debug("Channel reader exception %r", exc_info=e)
+                log.exception("Channel reader exception")
                 await self._cancel_tasks(e)
                 raise
 
@@ -453,6 +456,7 @@ async def _on_close(self, exc: Optional[ExceptionType] = None) -> None:
                 ),
                 timeout=self.connection.connection_tune.heartbeat or None,
             )
+
         self.connection.channels.pop(self.number, None)
 
     async def basic_get(

aiormq/connection.py

@@ -145,16 +145,16 @@ async def get_frame(self) -> ReceivedFrame:
 
         async with self.lock:
             try:
+                if self.reader is None:
+                    raise ConnectionError
+
                 frame_header = await self.reader.readexactly(1)
 
                 if frame_header == b"\0x00":
                     raise AMQPFrameError(
                         await self.reader.read(),
                     )
 
-                if self.reader is None:
-                    raise ConnectionError
-
                 frame_header += await self.reader.readexactly(6)
 
                 if not self.started and frame_header.startswith(b"AMQP"):