refactor: remove presence debounce for instant user departures
refactor: simplify websocket connection and error handling in rpc view
This commit is contained in:
parent
2e886c7bc1
commit
bdc8f149dc
@ -6,6 +6,14 @@
|
|||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
## Version 1.6.0 - 2025-12-17
|
||||||
|
|
||||||
|
Removes presence debounce to make user departures instant. Simplifies websocket connection and error handling in RPC views for improved reliability.
|
||||||
|
|
||||||
|
**Changes:** 2 files, 98 lines
|
||||||
|
**Languages:** Python (98 lines)
|
||||||
|
|
||||||
## Version 1.5.0 - 2025-12-17
|
## Version 1.5.0 - 2025-12-17
|
||||||
|
|
||||||
remove umami analytics script
|
remove umami analytics script
|
||||||
|
|||||||
@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "Snek"
|
name = "Snek"
|
||||||
version = "1.5.0"
|
version = "1.6.0"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
#license = { file = "LICENSE", content-type="text/markdown" }
|
#license = { file = "LICENSE", content-type="text/markdown" }
|
||||||
description = "Snek Chat Application by Molodetz"
|
description = "Snek Chat Application by Molodetz"
|
||||||
|
|||||||
@ -8,9 +8,6 @@ from snek.system.service import BaseService
|
|||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
from snek.system.model import now
|
from snek.system.model import now
|
||||||
|
|
||||||
PRESENCE_DEBOUNCE_SECONDS = 3.0
|
|
||||||
|
|
||||||
|
|
||||||
class SocketService(BaseService):
|
class SocketService(BaseService):
|
||||||
|
|
||||||
class Socket:
|
class Socket:
|
||||||
@ -48,7 +45,6 @@ class SocketService(BaseService):
|
|||||||
self.users = {}
|
self.users = {}
|
||||||
self.subscriptions = {}
|
self.subscriptions = {}
|
||||||
self.last_update = str(datetime.now())
|
self.last_update = str(datetime.now())
|
||||||
self._departure_tasks = {}
|
|
||||||
|
|
||||||
async def user_availability_service(self):
|
async def user_availability_service(self):
|
||||||
logger.info("User availability update service started.")
|
logger.info("User availability update service started.")
|
||||||
@ -90,10 +86,6 @@ class SocketService(BaseService):
|
|||||||
logger.info(f"Added socket for user {s.user['username']}")
|
logger.info(f"Added socket for user {s.user['username']}")
|
||||||
|
|
||||||
is_first_connection = False
|
is_first_connection = False
|
||||||
if user_uid in self._departure_tasks:
|
|
||||||
self._departure_tasks[user_uid].cancel()
|
|
||||||
del self._departure_tasks[user_uid]
|
|
||||||
|
|
||||||
if not self.users.get(user_uid):
|
if not self.users.get(user_uid):
|
||||||
self.users[user_uid] = set()
|
self.users[user_uid] = set()
|
||||||
is_first_connection = True
|
is_first_connection = True
|
||||||
@ -147,25 +139,7 @@ class SocketService(BaseService):
|
|||||||
if user_uid in self.users:
|
if user_uid in self.users:
|
||||||
self.users[user_uid].discard(s)
|
self.users[user_uid].discard(s)
|
||||||
if len(self.users[user_uid]) == 0:
|
if len(self.users[user_uid]) == 0:
|
||||||
await self._schedule_departure(user_uid, user_nick, user_color)
|
|
||||||
|
|
||||||
async def _schedule_departure(self, user_uid, user_nick, user_color):
|
|
||||||
if user_uid in self._departure_tasks:
|
|
||||||
return
|
|
||||||
|
|
||||||
async def delayed_departure():
|
|
||||||
try:
|
|
||||||
await asyncio.sleep(PRESENCE_DEBOUNCE_SECONDS)
|
|
||||||
if user_uid in self.users and len(self.users[user_uid]) == 0:
|
|
||||||
await self._broadcast_presence("departed", user_uid, user_nick, user_color)
|
await self._broadcast_presence("departed", user_uid, user_nick, user_color)
|
||||||
if user_uid in self._departure_tasks:
|
|
||||||
del self._departure_tasks[user_uid]
|
|
||||||
except asyncio.CancelledError:
|
|
||||||
pass
|
|
||||||
except Exception as ex:
|
|
||||||
logger.warning(f"Error in departure broadcast: {ex}")
|
|
||||||
|
|
||||||
self._departure_tasks[user_uid] = asyncio.create_task(delayed_departure())
|
|
||||||
|
|
||||||
async def _broadcast_presence(self, event_type, user_uid, user_nick, user_color):
|
async def _broadcast_presence(self, event_type, user_uid, user_nick, user_color):
|
||||||
if not user_uid or not user_nick:
|
if not user_uid or not user_nick:
|
||||||
|
|||||||
@ -627,27 +627,24 @@ class RPCView(BaseView):
|
|||||||
|
|
||||||
ws = web.WebSocketResponse()
|
ws = web.WebSocketResponse()
|
||||||
await ws.prepare(self.request)
|
await ws.prepare(self.request)
|
||||||
user_uid = self.request.session.get("uid") if self.request.session.get("logged_in") else None
|
if self.request.session.get("logged_in"):
|
||||||
|
await self.services.socket.add(ws, self.request.session.get("uid"))
|
||||||
try:
|
|
||||||
if user_uid:
|
|
||||||
await self.services.socket.add(ws, user_uid)
|
|
||||||
async for subscription in self.services.channel_member.find(
|
async for subscription in self.services.channel_member.find(
|
||||||
user_uid=user_uid,
|
user_uid=self.request.session.get("uid"),
|
||||||
deleted_at=None,
|
deleted_at=None,
|
||||||
is_banned=False,
|
is_banned=False,
|
||||||
):
|
):
|
||||||
await self.services.socket.subscribe(
|
await self.services.socket.subscribe(
|
||||||
ws, subscription["channel_uid"], user_uid
|
ws, subscription["channel_uid"], self.request.session.get("uid")
|
||||||
)
|
)
|
||||||
if not scheduled and self.request.app.uptime_seconds < 5:
|
if not scheduled and self.request.app.uptime_seconds < 5:
|
||||||
await schedule(
|
await schedule(
|
||||||
user_uid,
|
self.request.session.get("uid"),
|
||||||
0,
|
0,
|
||||||
{"event": "refresh", "data": {"message": "Finishing deployment"}},
|
{"event": "refresh", "data": {"message": "Finishing deployment"}},
|
||||||
)
|
)
|
||||||
await schedule(
|
await schedule(
|
||||||
user_uid,
|
self.request.session.get("uid"),
|
||||||
15,
|
15,
|
||||||
{"event": "deployed", "data": {"uptime": self.request.app.uptime}},
|
{"event": "deployed", "data": {"uptime": self.request.app.uptime}},
|
||||||
)
|
)
|
||||||
@ -658,13 +655,12 @@ class RPCView(BaseView):
|
|||||||
try:
|
try:
|
||||||
await rpc(msg.json())
|
await rpc(msg.json())
|
||||||
except Exception as ex:
|
except Exception as ex:
|
||||||
|
print("XXXXXXXXXX Deleting socket", ex, flush=True)
|
||||||
logger.exception(ex)
|
logger.exception(ex)
|
||||||
|
await self.services.socket.delete(ws)
|
||||||
break
|
break
|
||||||
elif msg.type == web.WSMsgType.ERROR:
|
elif msg.type == web.WSMsgType.ERROR:
|
||||||
break
|
pass
|
||||||
elif msg.type == web.WSMsgType.CLOSE:
|
elif msg.type == web.WSMsgType.CLOSE:
|
||||||
break
|
pass
|
||||||
finally:
|
|
||||||
await self.services.socket.delete(ws)
|
|
||||||
|
|
||||||
return ws
|
return ws
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user