Async: await socket closure (#701)

Additionally, add warning about known performance issue for aync driver with
Python 3.8.

Author: robsdedude

docs/source/async_api.rst

@@ -14,6 +14,11 @@ Async API Documentation
 
 .. versionadded:: 5.0
 
+.. warning::
+    There are known issue with Python 3.8 and the async driver where it
+    gradually slows down. Generally, it's recommended to use the latest
+    supported version of Python for best performance, stability, and security.
+
 ******************
 AsyncGraphDatabase
 ******************

neo4j/_async/io/_bolt.py

@@ -658,7 +658,7 @@ async def close(self):
                 pass
         log.debug("[#%04X]  C: <CLOSE>", self.local_port)
         try:
-            self.socket.close()
+            await self.socket.close()
         except OSError:
             pass
         finally:

neo4j/_async_compat/network/_bolt_socket.py

@@ -136,8 +136,9 @@ async def sendall(self, data):
         io_fut = self._writer.drain()
         return await self._wait_for_io(io_fut)
 
-    def close(self):
+    async def close(self):
         self._writer.close()
+        await self._writer.wait_closed()
 
     @classmethod
     async def _connect_secure(cls, resolved_address, timeout, keep_alive, ssl):
@@ -210,7 +211,7 @@ async def _connect_secure(cls, resolved_address, timeout, keep_alive, ssl):
             log.debug("[#0000]  C: <TIMEOUT> %s", resolved_address)
             log.debug("[#0000]  C: <CLOSE> %s", resolved_address)
             if s:
-                cls.close_socket(s)
+                await cls.close_socket(s)
             raise ServiceUnavailable(
                 "Timed out trying to establish connection to {!r}".format(
                     resolved_address))
@@ -296,15 +297,15 @@ async def _handshake(self, resolved_address):
         return self, agreed_version, handshake, data
 
     @classmethod
-    def close_socket(cls, socket_):
+    async def close_socket(cls, socket_):
         if isinstance(socket_, socket):
             try:
                 socket_.shutdown(SHUT_RDWR)
                 socket_.close()
             except OSError:
                 pass
         else:
-            socket_.close()
+            await socket_.close()
 
     @classmethod
     async def connect(cls, address, *, timeout, custom_resolver, ssl_context,
@@ -339,12 +340,12 @@ async def connect(cls, address, *, timeout, custom_resolver, ssl_context,
                 log.debug("[#%04X]  C: <CONNECTION FAILED> %s", local_port,
                           err_str)
                 if s:
-                    cls.close_socket(s)
+                    await cls.close_socket(s)
                 errors.append(error)
                 failed_addresses.append(resolved_address)
             except Exception:
                 if s:
-                    cls.close_socket(s)
+                    await cls.close_socket(s)
                 raise
         if not errors:
             raise ServiceUnavailable(