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

Apply PR feedback:

Browse Source 
- fix await stream close/reset
- make `_handle_dead_peer` a sync function
This commit is contained in:
NIC619 2019-11-05 14:27:06 +08:00
parent d36e323703
commit eeb87848af
No known key found for this signature in database
GPG Key ID: 570C35F5C2D51B17
1 changed files with 4 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
libp2p/pubsub/pubsub.py
View File

@ -253,10 +253,10 @@ class Pubsub:
try: try:
await self.continuously_read_stream(stream) await self.continuously_read_stream(stream)
except StreamEOF as error: except StreamEOF as error:
stream.close() await stream.close()
logger.debug("fail to read from peer %s, error=%s", peer_id, error) logger.debug("fail to read from peer %s, error=%s", peer_id, error)
except (ParseError, IncompleteReadError, StreamReset) as error: except (ParseError, IncompleteReadError, StreamReset) as error:
stream.reset() await stream.reset()
logger.debug("read corrupted data from peer %s, error=%s", peer_id, error) logger.debug("read corrupted data from peer %s, error=%s", peer_id, error)
async def _handle_new_peer(self, peer_id: ID) -> None: async def _handle_new_peer(self, peer_id: ID) -> None:
@ -281,7 +281,7 @@ class Pubsub:
logger.debug("added new peer %s", peer_id) logger.debug("added new peer %s", peer_id)
async def _handle_dead_peer(self, peer_id: ID) -> None: def _handle_dead_peer(self, peer_id: ID) -> None:
del self.peers[peer_id] del self.peers[peer_id]
for topic in self.peer_topics: for topic in self.peer_topics:
@ -300,14 +300,9 @@ class Pubsub:
pubsub protocols we support pubsub protocols we support
""" """
while True: while True:
peer_id: ID = await self.peer_queue.get() peer_id: ID = await self.peer_queue.get()
# Add Peer # Add Peer
asyncio.ensure_future(self._handle_new_peer(peer_id)) asyncio.ensure_future(self._handle_new_peer(peer_id))
# Force context switch
await asyncio.sleep(0)
async def handle_dead_peer_queue(self) -> None: async def handle_dead_peer_queue(self) -> None:
""" """
@ -315,14 +310,9 @@ class Pubsub:
remove peer info from pubsub and pubsub router. remove peer info from pubsub and pubsub router.
""" """
while True: while True:
peer_id: ID = await self.dead_peer_queue.get() peer_id: ID = await self.dead_peer_queue.get()
# Remove Peer # Remove Peer
asyncio.ensure_future(self._handle_dead_peer(peer_id)) self._handle_dead_peer(peer_id)
# Force context switch
await asyncio.sleep(0)
def handle_subscription( def handle_subscription(
self, origin_id: ID, sub_message: rpc_pb2.RPC.SubOpts self, origin_id: ID, sub_message: rpc_pb2.RPC.SubOpts