WebSocket and Real-time

Message Handler Compilation

def handle_message(msg: dict) -> dict:
    event_type = msg['type']
    if event_type == 'update':
        return process_update(msg['data'])
    elif event_type == 'delete':
        return process_delete(msg['data'])
    return {'error': 'unknown type'}

FastAPI WebSockets

@app.websocket("/ws")
async def websocket_endpoint(websocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_json()
        result = handle_message(data)  # compiled
        await websocket.send_json(result)

Socket.io

@sio.on('compute')
def on_compute(sid, data):
    result = compiled_compute(data)
    sio.emit('result', result, room=sid)