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)))