almet/umap
1
0
Fork 0
mirror of https://github.com/umap-project/umap.git synced 2025-04-29 03:42:37 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

wip(sync): try a better pattern to unsubscribe to pubsub channels

Browse source 
When publishing a "STOP", this would unsubscribe every listener of the
channel.
This commit is contained in:
Yohan Boniface 2025-01-20 19:13:41 +01:00
parent 7e42331533
commit 82342ea00f
1 changed files with 9 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
umap/sync/app.py
View file

@ -63,6 +63,7 @@ class Peer:
self.username = username or "" self.username = username or ""
self.room_id = room_id self.room_id = room_id
self.is_authenticated = False self.is_authenticated = False
self._subscriptions = []
async def get_peers(self): async def get_peers(self):
peers = await self.client.hgetall(self.room_id) peers = await self.client.hgetall(self.room_id)
@ -73,17 +74,20 @@ class Peer:
async def reader(pubsub): async def reader(pubsub):
await pubsub.subscribe(channel_name) await pubsub.subscribe(channel_name)
while True: while True:
if pubsub.connection is None:
# It has been unsubscribed/closed.
break
try: try:
message = await pubsub.get_message(ignore_subscribe_messages=True) message = await pubsub.get_message(ignore_subscribe_messages=True)
except Exception as err: except Exception as err:
print(err) print(err)
break break
if message is not None: if message is not None:
if message["data"].decode() == "STOP":
break
await self.send(message["data"].decode()) await self.send(message["data"].decode())
await asyncio.sleep(0.001) # Be nice with the server
async with self.client.pubsub() as pubsub: async with self.client.pubsub() as pubsub:
self._subscriptions.append(pubsub)
asyncio.create_task(reader(pubsub)) asyncio.create_task(reader(pubsub))
async def listen(self): async def listen(self):
@ -95,10 +99,11 @@ class Peer:
async def disconnect(self): async def disconnect(self):
await self.client.hdel(self.room_id, self.peer_id) await self.client.hdel(self.room_id, self.peer_id)
for pubsub in self._subscriptions:
await pubsub.unsubscribe()
await pubsub.close()
await self.send_peers_list() await self.send_peers_list()
await self.client.aclose() await self.client.aclose()
await self.client.publish(self.room_id, "STOP")
await self.client.publish(self.peer_id, "STOP")
async def send_peers_list(self): async def send_peers_list(self):
message = ListPeersResponse(peers=await self.get_peers()) message = ListPeersResponse(peers=await self.get_peers())