taolib.testing.config_center.server.websocket.presence

用户在线状态追踪模块。

基于 Redis HASH 实现的分布式在线状态管理，支持多实例部署。

Attributes

logger

Classes

RedisPresenceTracker	基于 Redis HASH 的在线状态追踪实现。
InMemoryPresenceTracker	内存在线状态追踪实现（用于测试）。

Module Contents

taolib.testing.config_center.server.websocket.presence.logger

class taolib.testing.config_center.server.websocket.presence.RedisPresenceTracker(redis_client: redis.asyncio.Redis, *, presence_ttl: int = 120)

基于 Redis HASH 的在线状态追踪实现。

_redis

_ttl = 120

_key(user_id: str) → str

async set_online(user_id: str, instance_id: str) → None

标记用户在线。

async set_offline(user_id: str, instance_id: str) → None

标记用户离线（若该实例上无更多连接）。

async get_status(user_id: str) → taolib.testing.config_center.server.websocket.models.UserPresence | None

获取用户在线状态。

async get_all_online() → list[taolib.testing.config_center.server.websocket.models.UserPresence]

获取所有在线用户（通过 SCAN 避免阻塞）。

async refresh(user_id: str) → None

刷新在线状态 TTL（心跳时调用）。

class taolib.testing.config_center.server.websocket.presence.InMemoryPresenceTracker

内存在线状态追踪实现（用于测试）。

_state: dict[str, taolib.testing.config_center.server.websocket.models.UserPresence]

_instances: dict[str, set[str]]

async set_online(user_id: str, instance_id: str) → None

async set_offline(user_id: str, instance_id: str) → None

async get_status(user_id: str) → taolib.testing.config_center.server.websocket.models.UserPresence | None

async get_all_online() → list[taolib.testing.config_center.server.websocket.models.UserPresence]

async refresh(user_id: str) → None