~12 min · fastapi, broadcast, exclude, background-task
실전 broadcast 는 세 모양. Room broadcast: room 의 모두에게. Sender 제외 room broadcast: 보낸 user 한테 echo back 회피. 특정 user 한테 직접 message: user_id → WebSocket 도 추적해야 함. Manager 에 셋 다 구현하면 대부분 needs 커버.
주기적 update (price ticker, system metric, heartbeat broadcast) 위해 lifespan startup 에 background task 띄워. manager.broadcast() 부르는 모든 task 가 어떤 클라 request 와도 독립적으로 돌아가. application 의 lifetime 동안 살아.
class ConnectionManager:
# ... rooms as before ...
user_map: Dict[str, WebSocket] = {}
async def broadcast(self, room: str, message: dict, *, exclude: WebSocket | None = None):
dead = []
for ws in self.rooms.get(room, set()):
if ws is exclude:
continue
try:
await ws.send_json(message)
except Exception:
dead.append(ws)
for ws in dead:
self.disconnect(ws, room)
async def send_to_user(self, user_id: str, message: dict):
ws = self.user_map.get(user_id)
if ws is None:
return False
try:
await ws.send_json(message)
return True
except Exception:
self.user_map.pop(user_id, None)
return Falsefrom contextlib import asynccontextmanager
import asyncio
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
task = asyncio.create_task(price_ticker())
yield
task.cancel()
async def price_ticker():
while True:
price = await get_latest_btc()
await manager.broadcast('btc-feed', {
'type': 'price.update',
'data': {'symbol': 'BTC', 'price': price},
})
await asyncio.sleep(1)
app = FastAPI(lifespan=lifespan)send_to_user 와 broadcast(..., exclude=...) 추가. chat endpoint 짜서 모든 message 가 sender 제외 room 에 broadcast; sender 는 private chat.ack 받게. 브라우저 탭 세 개로 sender 만 ack 보고 나머지 둘은 chat.message 보는 거 확인.아직 댓글이 없어요. 첫 댓글을 남겨보세요.