varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2026-02-12 16:10:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

add grave that were removed by mistake

Browse Source
This commit is contained in:
Jinesh Jain
2025-07-17 20:55:32 +05:30
parent e3c9b4bd54
commit c9162beb2b
1 changed files with 12 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
libp2p/stream_muxer/mplex/mplex_stream.py
View File

@ -107,7 +107,7 @@ class MplexStream(IMuxedStream):
name: str name: str
stream_id: StreamID stream_id: StreamID
# NOTE: All methods used here are part of Mplex which is a derived # NOTE: All methods used here are part of `Mplex` which is a derived
# class of IMuxedConn. Ignoring this type assignment should not pose # class of IMuxedConn. Ignoring this type assignment should not pose
# any risk. # any risk.
muxed_conn: "Mplex" # type: ignore[assignment] muxed_conn: "Mplex" # type: ignore[assignment]
@ -117,7 +117,7 @@ class MplexStream(IMuxedStream):
rw_lock: ReadWriteLock rw_lock: ReadWriteLock
close_lock: trio.Lock close_lock: trio.Lock
# NOTE: dataIn is size of 8 in Go implementation. # NOTE: `dataIn` is size of 8 in Go implementation.
incoming_data_channel: "trio.MemoryReceiveChannel[bytes]" incoming_data_channel: "trio.MemoryReceiveChannel[bytes]"
event_local_closed: trio.Event event_local_closed: trio.Event
@ -175,8 +175,8 @@ class MplexStream(IMuxedStream):
async def read(self, n: int | None = None) -> bytes: async def read(self, n: int | None = None) -> bytes:
""" """
Read up to n bytes. Read possibly returns fewer than n bytes, if Read up to n bytes. Read possibly returns fewer than `n` bytes, if
there are not enough bytes in the Mplex buffer. If n is None, read there are not enough bytes in the Mplex buffer. If `n is None`, read
until EOF. until EOF.
:param n: number of bytes to read :param n: number of bytes to read
@ -185,8 +185,8 @@ class MplexStream(IMuxedStream):
async with self.rw_lock.read_lock(): async with self.rw_lock.read_lock():
if n is not None and n < 0: if n is not None and n < 0:
raise ValueError( raise ValueError(
"the number of bytes to read n must be non-negative or " "the number of bytes to read `n` must be non-negative or "
f"None to indicate read until EOF, got n={n}" f"`None` to indicate read until EOF, got n={n}"
) )
if self.event_reset.is_set(): if self.event_reset.is_set():
raise MplexStreamReset raise MplexStreamReset
@ -202,8 +202,8 @@ class MplexStream(IMuxedStream):
except trio.EndOfChannel: except trio.EndOfChannel:
raise MplexStreamEOF raise MplexStreamEOF
except trio.WouldBlock: except trio.WouldBlock:
# We know receive will be blocked here. Wait for data here with # We know `receive` will be blocked here. Wait for data here with
# receive and catch all kinds of errors here. # `receive` and catch all kinds of errors here.
try: try:
data = await self.incoming_data_channel.receive() data = await self.incoming_data_channel.receive()
self._buf.extend(data) self._buf.extend(data)
@ -213,12 +213,12 @@ class MplexStream(IMuxedStream):
if self.event_remote_closed.is_set(): if self.event_remote_closed.is_set():
raise MplexStreamEOF raise MplexStreamEOF
except trio.ClosedResourceError as error: except trio.ClosedResourceError as error:
# Probably incoming_data_channel is closed in reset when we are # Probably `incoming_data_channel` is closed in `reset` when
# waiting for receive. # we are waiting for `receive`.
if self.event_reset.is_set(): if self.event_reset.is_set():
raise MplexStreamReset raise MplexStreamReset
raise Exception( raise Exception(
"incoming_data_channel is closed but stream is not reset. " "`incoming_data_channel` is closed but stream is not reset."
"This should never happen." "This should never happen."
) from error ) from error
self._buf.extend(self._read_return_when_blocked()) self._buf.extend(self._read_return_when_blocked())
@ -256,7 +256,7 @@ class MplexStream(IMuxedStream):
flag = ( flag = (
HeaderTags.CloseInitiator if self.is_initiator else HeaderTags.CloseReceiver HeaderTags.CloseInitiator if self.is_initiator else HeaderTags.CloseReceiver
) )
# TODO: Raise when muxed_conn.send_message fails and Mplex isn't shutdown. # TODO: Raise when `muxed_conn.send_message` fails and `Mplex` isn't shutdown.
await self.muxed_conn.send_message(flag, None, self.stream_id) await self.muxed_conn.send_message(flag, None, self.stream_id)
_is_remote_closed: bool _is_remote_closed: bool