Remove all errors and asyncio debug warnings on shutdown

aiormq/base.py

@@ -1,5 +1,6 @@
 import abc
 import asyncio
+import logging
 from contextlib import suppress
 from functools import wraps
 from typing import Any, Callable, Coroutine, Optional, Set, TypeVar, Union
@@ -59,6 +60,8 @@ async def reject_all(self, exception: Optional[ExceptionType]) -> None:
                 tasks.append(future)
             elif isinstance(future, asyncio.Future):
                 future.set_exception(exception or Exception)
+            else:
+                raise ValueError(future)
 
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
@@ -128,7 +131,13 @@ async def __closer(self, exc: Optional[ExceptionType]) -> None:
             await self._on_close(exc)
 
         with suppress(Exception):
-            await self._cancel_tasks(exc)
+            try:
+                await self._cancel_tasks(exc)
+            except BaseException:
+                logging.exception(
+                    "Exception when cancelling %r...", self.__class__,
+                )
+                raise
 
     async def close(
         self, exc: Optional[ExceptionType] = asyncio.CancelledError,

aiormq/channel.py

@@ -436,7 +436,7 @@ async def _reader(self) -> None:
 
                 await self.rpc_frames.put(frame)
             except asyncio.CancelledError:
-                return
+                raise
             except Exception as e:  # pragma: nocover
                 log.debug("Channel reader exception %r", exc_info=e)
                 await self._cancel_tasks(e)

aiormq/connection.py

@@ -145,16 +145,18 @@ async def get_frame(self) -> ReceivedFrame:
 
         async with self.lock:
             try:
-                frame_header = await self.reader.readexactly(1)
+                try:
+                    frame_header = await self.reader.readexactly(1)
+                except asyncio.IncompleteReadError as e:
+                    if e.partial:
+                        raise
+                    raise ConnectionClosed(0, "socket closed")
 
                 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"):
@@ -170,9 +172,12 @@ async def get_frame(self) -> ReceivedFrame:
 
                 frame_payload = await self.reader.readexactly(frame_length + 1)
             except asyncio.IncompleteReadError as e:
-                raise AMQPFrameError(
-                    "Server connection unexpectedly closed",
-                ) from e
+                if e.partial:
+                    raise AMQPFrameError(
+                        "Server connection unexpectedly closed",
+                    ) from e
+                raise ConnectionClosed(0, "socket closed")
+
         return pamqp.frame.unmarshal(frame_header + frame_payload)
 
     async def __anext__(self) -> ReceivedFrame:
@@ -265,6 +270,7 @@ def __init__(
         self.__close_reply_text: str = "normally closed"
         self.__close_class_id: int = 0
         self.__close_method_id: int = 0
+        self.__close_event: asyncio.Event = asyncio.Event()
         self.__update_secret_lock: asyncio.Lock = asyncio.Lock()
         self.__update_secret_future: Optional[asyncio.Future] = None
         self.__connection_unblocked: asyncio.Event = asyncio.Event()
@@ -578,6 +584,11 @@ async def __reader(self, frame_receiver: FrameReceiver) -> None:
                 if isinstance(frame, CHANNEL_CLOSE_RESPONSES):
                     self.channels[channel] = None
 
+                if self.__close_event.is_set():
+                    # Methods should be discarded after sending close
+                    #  https://bit.ly/3BCtywe
+                    log.warning("Ignoring frame %r after close", frame)
+                    break
                 await ch.frames.put((weight, frame))
         except asyncio.CancelledError as e:
             if self.is_connection_was_stuck:
@@ -693,6 +704,7 @@ async def _on_close(
         ex: Optional[ExceptionType] = ConnectionClosed(0, "normal closed"),
     ) -> None:
         log.debug("Closing connection %r cause: %r", self, ex)
+        self.__close_event.set()
         if not self._reader_task.done():
             self._reader_task.cancel()
         if not self._writer_task.done():