Refactor: Mplex, Swarm, and BasicHost
#303
Conversation
|
|
||
| async def _handle_new_streams(self) -> None: | ||
| # TODO: Break the loop when anything wrong in the connection. | ||
| while True: |
There was a problem hiding this comment.
does it make sense to put this while block inside a try block and then put await self.close() inside the finally: clause?
There was a problem hiding this comment.
It seems pretty convenient but is it also possible that we catch unwanted exceptions? In the latest code I catch only the MuxedConnUnavailable in this loop.
There was a problem hiding this comment.
On second thought, I'm thinking it might be possibly an easier pattern for the shutting down logic in Mplex in #304 , will investigate more. Thanks for bringing this up.
| @@ -34,17 +33,13 @@ class Mplex(IMuxedConn): | |||
| next_channel_id: int | |||
| streams: Dict[StreamID, MplexStream] | |||
| streams_lock: asyncio.Lock | |||
| new_stream_queue: "asyncio.Queue[IMuxedStream]" | |||
| shutdown: asyncio.Event | |||
|
|
|||
| _tasks: List["asyncio.Future[Any]"] | |||
|
|
|||
| # TODO: `generic_protocol_handler` should be refactored out of mplex conn. | |||
There was a problem hiding this comment.
is this comment still relevant?
There was a problem hiding this comment.
Ah I removed this line in #305 , but I can also fix it here. Good catch!
There was a problem hiding this comment.
if you removed it elsewhere, we can leave it here -- just want to keep master as clean as possible :)
|
Big thanks for reviewing! I will solve the conflicts first and then get this ready again tmr. |
Swarm#281.generic_protocol_handlerout ofMplex. It makes protocol negotiation abstracted out ofMplex, which makes testing easier. For example, we can test with aMplexStreamwithout the initial protocol negotiation.generic_protocol_handleris no longer registered by default inSwarm. Now, it is implemented inBasicHost, and is registered as thestream_handlerinSwarm.SwarmConn, which is a wrapper ofMuxedConn, has the reference to Swarm.