Handle Stream.wait_closed() windows bug gracefully

Author: 1st1

asyncpg/compat.py

@@ -79,3 +79,14 @@ def current_asyncio_task(loop):
 else:
     def current_asyncio_task(loop):
         return asyncio.Task.current_task(loop)
+
+
+async def wait_closed(stream):
+    # Not all asyncio versions have StreamWriter.wait_closed().
+    if hasattr(stream, 'wait_closed'):
+        try:
+            await stream.wait_closed()
+        except ConnectionResetError:
+            # On Windows wait_closed() sometimes propagates
+            # ConnectionResetError which is totally unnecessary.
+            pass

asyncpg/connect_utils.py

@@ -611,8 +611,7 @@ async def _negotiate_ssl_connection(host, port, conn_factory, *, loop, ssl,
         sock = sock.dup()  # Must come before tr.close()
     finally:
         writer.close()
-        if hasattr(writer, 'wait_closed'):
-            await writer.wait_closed()
+        await compat.wait_closed(writer)
 
     try:
         return await conn_factory(sock=sock)  # Must come after tr.close()

asyncpg/connection.py

@@ -1223,8 +1223,7 @@ async def _cancel(self, waiter):
                 waiter.set_result(None)
             if w is not None:
                 w.close()
-                if hasattr(w, 'wait_closed'):
-                    await w.wait_closed()
+                await compat.wait_closed(w)
 
     def _cancel_current_command(self, waiter):
         self._cancellations.add(self._loop.create_task(self._cancel(waiter)))