PR feedback for MplexStream.read

This commit is contained in:
mhchia 2019-08-16 10:21:51 +08:00 committed by Kevin Mai-Husan Chia
parent 8699568d43
commit c5f32bf431
2 changed files with 8 additions and 5 deletions

View File

@ -75,7 +75,8 @@ class Mplex(IMuxedConn):
async def read_buffer(self, stream_id: int) -> bytes:
"""
Read a message from stream_id's buffer, check raw connection for new messages
Read a message from stream_id's buffer, check raw connection for new messages.
`StreamNotFound` is raised when stream `stream_id` is not found in `Mplex`.
:param stream_id: stream id of stream to read from
:return: message read
"""
@ -86,6 +87,7 @@ class Mplex(IMuxedConn):
async def read_buffer_nonblocking(self, stream_id: int) -> Optional[bytes]:
"""
Read a message from `stream_id`'s buffer, non-blockingly.
`StreamNotFound` is raised when stream `stream_id` is not found in `Mplex`.
"""
if stream_id not in self.buffers:
raise StreamNotFound(f"stream {stream_id} is not found")

View File

@ -46,14 +46,15 @@ class MplexStream(IMuxedStream):
:param n: number of bytes to read
:return: bytes actually read
"""
# TODO: Handle `StreamNotFound` raised in `self.mplex_conn.read_buffer`.
# TODO: Add exceptions and handle/raise them in this class.
if n < 0 and n != -1:
raise ValueError(f"the number of bytes to read ``n`` must be positive or -1 to indicate read until EOF")
raise ValueError(
f"the number of bytes to read `n` must be positive or -1 to indicate read until EOF"
)
# If the buffer is empty at first, blocking wait for data.
if len(self._buf) == 0:
self._buf.extend(await self.mplex_conn.read_buffer(self.stream_id))
# Sanity check: `self._buf` should never be empty here.
if self._buf is None or len(self._buf) == 0:
raise Exception("`self._buf` should never be empty here")
# FIXME: If `n == -1`, we should blocking read until EOF, instead of returning when
# no message is available.