ConnectionManager 클래스

딕셔너리 대신 클래스를 쓰는 이유

트랙 3 의 가장 작은 관리자는 집합을 담은 딕셔너리였어. 실전 애플리케이션에는 세 차원이 더 필요해. 언제 어느 IP 와 세션에서 연결했는지 담는 연결별 메타데이터, 지금 사용자 X 의 연결을 찾는 사용자별 조회, 방·사용자·버전별 현재 연결 수를 보여 주는 관측 정보 야. 여러 모듈에 흩뿌리지 말고 클래스 하나에 모아.

세 가지 인덱스

거의 언제나 방 → 연결 집합, user_id → 연결, 연결 → 메타데이터라는 세 방향 조회가 필요해. 작은 딕셔너리 세 개면 되지만 모든 작업이 셋을 일관되게 갱신해야 해. 하나라도 빠뜨리면 운영 환경의 ‘오래된 연결’ 버그가 거기서 생겨.

Code

운영 환경을 닮은 연결 관리자·python

from fastapi import WebSocket
from typing import Dict, Set, Optional
import time

class ConnectionManager:
    def __init__(self):
        self.rooms: Dict[str, Set[WebSocket]] = {}
        self.user_map: Dict[str, WebSocket] = {}
        self.ws_meta: Dict[WebSocket, dict] = {}

    async def connect(self, ws: WebSocket, user_id: str, room: str):
        await ws.accept()
        self.rooms.setdefault(room, set()).add(ws)
        self.user_map[user_id] = ws
        self.ws_meta[ws] = {
            'user_id': user_id,
            'room': room,
            'connected_at': time.time(),
            'ip': ws.client.host if ws.client else None,
        }

    def disconnect(self, ws: WebSocket):
        meta = self.ws_meta.pop(ws, None)
        if meta is None:
            return
        room = meta['room']
        members = self.rooms.get(room)
        if members:
            members.discard(ws)
            if not members:
                self.rooms.pop(room, None)
        existing = self.user_map.get(meta['user_id'])
        if existing is ws:
            self.user_map.pop(meta['user_id'], None)

    async def send_to_user(self, user_id: str, message: dict) -> bool:
        ws = self.user_map.get(user_id)
        if ws is None:
            return False
        try:
            await ws.send_json(message)
            return True
        except Exception:
            self.disconnect(ws)
            return False

    async def broadcast(self, room: str, message: dict, *, exclude: Optional[WebSocket] = None):
        dead = []
        for ws in list(self.rooms.get(room, ())):
            if ws is exclude:
                continue
            try:
                await ws.send_json(message)
            except Exception:
                dead.append(ws)
        for ws in dead:
            self.disconnect(ws)

    @property
    def total_connections(self) -> int:
        return len(self.ws_meta)

    def room_users(self, room: str) -> list[str]:
        return [
            self.ws_meta[ws]['user_id']
            for ws in self.rooms.get(room, ())
            if ws in self.ws_meta
        ]

Exercise

/admin/connections HTTP 엔드포인트를 추가해 {rooms, total, users_per_room} 을 돌려줘. 시험 클라이언트 세 개를 연결한 채 방문해 숫자가 맞는지 확인해. 하나를 강제로 종료하고 새로 고침했을 때 카운트가 이전 값에 머물지 않고 몇 초 안에 갱신되어야 해.

딕셔너리 대신 클래스를 쓰는 이유

세 가지 인덱스

Code

External links

Exercise

Progress

댓글 0