INNER CODE UNIT · Python

websocket_endpoint

YichengYang-Ethan/oracle3 · oracle3/dashboard/server.py:358

    async def websocket_endpoint(websocket: WebSocket):
        from starlette.websockets import WebSocketDisconnect

        await websocket.accept()
        ws_clients.add(websocket)
        try:
            while True:
                try:
                    state = _serialize_snapshot(engine)
                    await websocket.send_json(state)
                except Exception:
                    logger.debug('WebSocket send failed', exc_info=True)
                    break
                await asyncio.sleep(2.0)
        except WebSocketDisconnect:
            pass
        finally:
            ws_clients.discard(websocket)

View source record →