Function model_stream

src/py/reactpy/reactpy/backend/starlette.py:141–170

(socket: WebSocket)

    options: Options, app: Starlette, component: RootComponentConstructor
) -> None:
    async def model_stream(socket: WebSocket) -> None:
        await socket.accept()
        send, recv = _make_send_recv_callbacks(socket)

        pathname = "/" + socket.scope["path_params"].get("path", "")
        pathname = pathname[len(options.url_prefix) :] or "/"
        search = socket.scope["query_string"].decode()

        try:
            await serve_layout(
                Layout(
                    ConnectionContext(
                        component(),
                        value=Connection(
                            scope=socket.scope,
                            location=Location(pathname, f"?{search}" if search else ""),
                            carrier=socket,
                        ),
                    )
                ),
                send,
                recv,
            )
        except BaseExceptionGroup as egroup:
            for e in egroup.exceptions:
                if isinstance(e, WebSocketDisconnect):
                    logger.info(f"WebSocket disconnect: {e.code}")
                    break
            else:  # nocov
                raise

    app.add_websocket_route(str(STREAM_PATH), model_stream)
    app.add_websocket_route(f"{STREAM_PATH}/{{path:path}}", model_stream)