Use asyncio.Server as an async ctx manager in our TCP listener

This was added on py37, and ensures we don't silence crashes in our TCP
listener server

Author: gsalgado

setup.py

@@ -48,7 +48,6 @@
         "websockets>=8.1.0",
         "jsonschema==3.0.1",
         "mypy-extensions>=0.4.3,<0.5.0",
-        "typing_extensions>=3.7.4,<4.0.0",
         "ruamel.yaml==0.16.10",
         "argcomplete>=1.10.0,<2",
         "multiaddr>=0.0.8,<0.1.0",

trinity/server.py

@@ -1,8 +1,6 @@
 from abc import abstractmethod
 import asyncio
-import contextlib
 from typing import (
-    AsyncIterator,
     Generic,
     Tuple,
     Type,
@@ -103,21 +101,6 @@ def __init__(self,
     def _make_peer_pool(self) -> TPeerPool:
         ...
 
-    @contextlib.asynccontextmanager
-    async def tcp_listener(self) -> AsyncIterator[None]:
-        # TODO: Support IPv6 addresses as well.
-        tcp_listener = await asyncio.start_server(
-            self.receive_handshake,
-            host=BOUND_IP,
-            port=self.port,
-        )
-        try:
-            yield
-        finally:
-            self.logger.info("Server finished, closing TCP listener...")
-            tcp_listener.close()
-            await tcp_listener.wait_closed()
-
     async def run(self) -> None:
         self.logger.info("Running server...")
         self.logger.info(
@@ -129,9 +112,20 @@ async def run(self) -> None:
         self.logger.info('network: %s', self.network_id)
         self.logger.info('peers: max_peers=%s', self.max_peers)
 
-        async with self.tcp_listener():
-            self.manager.run_daemon_child_service(self.peer_pool)
-            await self.manager.wait_finished()
+        # TODO: Support IPv6 addresses as well.
+        tcp_listener = await asyncio.start_server(
+            self.receive_handshake,
+            host=BOUND_IP,
+            port=self.port,
+        )
+        try:
+            # mypy doesn't know that the Server returned above is an async contextmanager.
+            async with tcp_listener:  # type: ignore
+                self.manager.run_daemon_child_service(self.peer_pool)
+                await tcp_listener.serve_forever()
+        finally:
+            self.logger.info("TCP Listener finished, cancelling Server")
+            self.manager.cancel()
 
     async def receive_handshake(
             self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: