Function model_stream

src/py/reactpy/reactpy/backend/sanic.py:163–194

(
        request: request.Request[Any, Any],
        socket: WebSocketConnection,
        path: str = "",
    )

    options: Options,
) -> None:
    async def model_stream(
        request: request.Request[Any, Any],
        socket: WebSocketConnection,
        path: str = "",
    ) -> None:
        asgi_app = getattr(request.app, "_asgi_app", None)
        scope = asgi_app.transport.scope if asgi_app else {}
        if not scope:  # nocov
            logger.warning("No scope. Sanic may not be running with an ASGI server")

        send, recv = _make_send_recv_callbacks(socket)
        await serve_layout(
            Layout(
                ConnectionContext(
                    constructor(),
                    value=Connection(
                        scope=scope,
                        location=Location(
                            pathname=f"/{path[len(options.url_prefix):]}",
                            search=(
                                f"?{request.query_string}"
                                if request.query_string
                                else ""
                            ),
                        ),
                        carrier=_SanicCarrier(request, socket),
                    ),
                )
            ),
            send,
            recv,
        )

    api_blueprint.add_websocket_route(
        model_stream,