StarMirror/py-libp2p
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Clean up peer record if pubsub stream fail

Browse Source
This commit is contained in:
NIC619 2019-11-05 15:09:53 +08:00
parent eeb87848af
commit 93ef36bd86
No known key found for this signature in database
GPG Key ID: 570C35F5C2D51B17
1 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
libp2p/pubsub/pubsub.py
View File

@ -248,16 +248,17 @@ class Pubsub:
""" """
peer_id = stream.muxed_conn.peer_id peer_id = stream.muxed_conn.peer_id
# Error handling pattern reference:
# https://github.com/libp2p/go-libp2p-pubsub/blob/534fe2f382d8dd75dab89ddb0760542546c9f24e/comm.go#L38-L46 # noqa: E501
try: try:
await self.continuously_read_stream(stream) await self.continuously_read_stream(stream)
except StreamEOF as error: except (StreamEOF, StreamReset, ParseError, IncompleteReadError) as error:
await stream.close() logger.debug(
logger.debug("fail to read from peer %s, error=%s", peer_id, error) "fail to read from peer %s, error=%s,"
except (ParseError, IncompleteReadError, StreamReset) as error: "closing the stream and remove the peer from record",
peer_id,
error,
)
await stream.reset() await stream.reset()
logger.debug("read corrupted data from peer %s, error=%s", peer_id, error) self._handle_dead_peer(peer_id)
async def _handle_new_peer(self, peer_id: ID) -> None: async def _handle_new_peer(self, peer_id: ID) -> None:
try: try:
@ -282,6 +283,8 @@ class Pubsub:
logger.debug("added new peer %s", peer_id) logger.debug("added new peer %s", peer_id)
def _handle_dead_peer(self, peer_id: ID) -> None: def _handle_dead_peer(self, peer_id: ID) -> None:
if peer_id not in self.peers:
return
del self.peers[peer_id] del self.peers[peer_id]
for topic in self.peer_topics: for topic in self.peer_topics: