연결 관리자 패턴

처리기 하나로는 부족해

에코 서버는 연결한 클라이언트 하나와만 대화해. 채팅, 전체 전송, 접속 상태처럼 더 흥미로운 기능은 서버가 여러 연결을 동시에 알아야 하지. 이때 연결 관리자 를 둬. 열린 소켓을 추적하면서 방, 사용자, 채널처럼 애플리케이션이 중요하게 여기는 값으로 찾을 수 있게 하는 작은 단일 객체야.

가장 작은 관리자

방 이름을 WebSocket 객체 집합에 연결하는 딕셔너리만으로 대부분의 상황을 처리할 수 있어. 개인 메시지가 필요하면 사용자 ID 추적을 더하고, 관측 정보가 필요하면 연결 시각이나 IP 같은 메타데이터를 더해. 필요가 생길 때마다 점진적으로 키우면 돼.

전체 전송 중 죽은 연결 처리하기

방의 연결 집합을 순회할 때는 네트워크 단절을 아직 감지하지 못한 반쯤 죽은 소켓이 섞여 있을 수 있어. 모든 send 호출을 감싸고, 죽은 참조를 모은 다음, 반복이 끝난 뒤 제거해. 순회 중인 집합을 바로 바꾸면 안 돼.

Code

가장 작은 연결 관리자·python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, Set

app = FastAPI()

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

    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        self.rooms.setdefault(room, set()).add(websocket)

    def disconnect(self, websocket: WebSocket, room: str):
        members = self.rooms.get(room)
        if not members:
            return
        members.discard(websocket)
        if not members:
            self.rooms.pop(room, None)  # garbage collect empty rooms

    async def broadcast(self, room: str, message: dict):
        dead = []
        for ws in self.rooms.get(room, set()):
            try:
                await ws.send_json(message)
            except Exception:
                dead.append(ws)
        for ws in dead:
            self.disconnect(ws, room)

manager = ConnectionManager()

@app.websocket('/ws/{room}')
async def chat(websocket: WebSocket, room: str):
    await manager.connect(websocket, room)
    try:
        async for msg in websocket.iter_json():
            await manager.broadcast(room, {
                'type': 'chat.message',
                'data': msg,
            })
    except WebSocketDisconnect:
        manager.disconnect(websocket, room)
        await manager.broadcast(room, {
            'type': 'user.left',
            'data': {'room': room},
        })

Exercise

위 연결 관리자를 만들고 /ws/general 에 브라우저 탭 세 개를 열어. 한 탭에서 입력하면 다른 두 탭에 메시지가 나타나는지 확인해. 탭 하나를 강제로 종료하면 나머지 둘이 user.left 전체 메시지를 받고, 마지막 탭을 닫으면 방 딕셔너리가 정리되는지도 확인해.