~12 min · management, user-id, multi-device
실전 user 가 동시에 폰, 노트북, 아이패드에서 connect. 순진한 user_id → ws 맵은 그 순간 깨짐 — 두 번째 connection 이 첫 번째 조용히 덮어. 답은 one-to-many 인덱스: user_id → connection_id set, 별도 connection_id → WebSocket.
Connect 시 random ID 생성, 내부에서 모든 곳에 사용. 클라가 알 필요 없어. '이 connection 한테 broadcast' 가능해지면서 구현 detail 노출 안 함, connection 별 rate limit, telemetry, session metadata 추가 가능.
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 로 두 브라우저 (다른 창) 에서 connect. send_to_user(user_id, ...) 로 직접 message. 둘 다 받는지 확인. 하나 닫으면 — 다른 거 코드 변경 없이 계속 작동해야 함.아직 댓글이 없어요. 첫 댓글을 남겨보세요.