~12 min · management, user-id, multi-device
실제 사용자는 휴대폰, 노트북, 태블릿에서 동시에 연결해. 단순한 user_id → ws 맵은 두 번째 연결이 첫 번째를 조용히 덮어쓰는 순간 깨져. 일대다 인덱스로 바꿔 user_id → connection_id set 을 두고, 별도로 connection_id → WebSocket 을 관리해야 해.
연결할 때 무작위 ID 를 만들고 내부의 모든 곳에서 사용해. 클라이언트는 이 값을 알 필요가 없어. 구현 세부 사항을 드러내지 않으면서 특정 연결에만 메시지를 보내고, 연결별 속도 제한, 원격 측정, 세션 메타데이터도 붙일 수 있어.
import uuid
from typing import Dict, Set
class IdentifiedConnectionManager:
def __init__(self):
self.connections: Dict[str, WebSocket] = {} # conn_id -> ws
self.user_conns: Dict[str, Set[str]] = {} # user_id -> conn_ids
async def connect(self, ws: WebSocket, user_id: str) -> str:
await ws.accept()
conn_id = uuid.uuid4().hex[:8]
self.connections[conn_id] = ws
self.user_conns.setdefault(user_id, set()).add(conn_id)
return conn_id
def disconnect(self, conn_id: str, user_id: str):
self.connections.pop(conn_id, None)
conns = self.user_conns.get(user_id)
if conns:
conns.discard(conn_id)
if not conns:
self.user_conns.pop(user_id, None)
async def send_to_user(self, user_id: str, message: dict) -> int:
'''Send to ALL connections for a user (multi-device fan-out).'''
sent = 0
for conn_id in list(self.user_conns.get(user_id, ())):
ws = self.connections.get(conn_id)
if ws is None:
continue
try:
await ws.send_json(message)
sent += 1
except Exception:
self.disconnect(conn_id, user_id)
return sentuser_id 로 연결해. send_to_user(user_id, ...) 로 직접 메시지를 보내 둘 다 받는지 확인해. 하나를 닫아도 코드 변경 없이 나머지 하나가 계속 작동해야 해.아직 댓글이 없어요. 첫 댓글을 남겨보세요.