WebSockets Advanced

from justapi import JustAPIApp
from starlette.websockets import WebSocket, WebSocketDisconnect

app = JustAPIApp()

class ConnectionManager:
    def __init__(self):
        self.connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.connections.remove(websocket)

    async def broadcast(self, message: str):
        for conn in self.connections:
            await conn.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Echo: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

from collections import defaultdict

rooms: dict[str, list[WebSocket]] = defaultdict(list)

@app.websocket("/ws/{room}")
async def room_websocket(websocket: WebSocket, room: str):
    await websocket.accept()
    rooms[room].append(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            for conn in rooms[room]:
                if conn != websocket:
                    await conn.send_text(data)
    except WebSocketDisconnect:
        rooms[room].remove(websocket)