ASGI server

ASGI server.

Source code in pycrdt_websocket/asgi_server.py

class ASGIServer:
    """ASGI server."""

    def __init__(
        self,
        websocket_server: WebsocketServer,
        on_connect: Callable[[dict[str, Any], dict[str, Any]], Awaitable[bool] | bool]
        | None = None,
        on_disconnect: Callable[[dict[str, Any]], Awaitable[None] | None] | None = None,
    ):
        """Initialize the object.

        Arguments:
            websocket_server: An instance of WebsocketServer.
            on_connect: An optional callback to call when connecting the WebSocket.
                If the callback returns True, the WebSocket is not accepted.
            on_disconnect: An optional callback called when disconnecting the WebSocket.
        """
        self._websocket_server = websocket_server
        self._on_connect = on_connect
        self._on_disconnect = on_disconnect

    async def __call__(
        self,
        scope: dict[str, Any],
        receive: Callable[[], Awaitable[dict[str, Any]]],
        send: Callable[[dict[str, Any]], Awaitable[None]],
    ):
        msg = await receive()
        if msg["type"] == "websocket.connect":
            if self._on_connect is not None:
                close = self._on_connect(msg, scope)
                if isawaitable(close):
                    close = await close
                if close:
                    return

            await send({"type": "websocket.accept"})
            websocket = ASGIWebsocket(receive, send, scope["path"], self._on_disconnect)
            await self._websocket_server.serve(websocket)

__init__(websocket_server, on_connect=None, on_disconnect=None)

Initialize the object.

Parameters:

Name	Type	Description	Default
websocket_server	WebsocketServer	An instance of WebsocketServer.	required
on_connect	Callable[[dict[str, Any], dict[str, Any]], Awaitable[bool] | bool] | None	An optional callback to call when connecting the WebSocket. If the callback returns True, the WebSocket is not accepted.	None
on_disconnect	Callable[[dict[str, Any]], Awaitable[None] | None] | None	An optional callback called when disconnecting the WebSocket.	None

Source code in pycrdt_websocket/asgi_server.py

def __init__(
    self,
    websocket_server: WebsocketServer,
    on_connect: Callable[[dict[str, Any], dict[str, Any]], Awaitable[bool] | bool]
    | None = None,
    on_disconnect: Callable[[dict[str, Any]], Awaitable[None] | None] | None = None,
):
    """Initialize the object.

    Arguments:
        websocket_server: An instance of WebsocketServer.
        on_connect: An optional callback to call when connecting the WebSocket.
            If the callback returns True, the WebSocket is not accepted.
        on_disconnect: An optional callback called when disconnecting the WebSocket.
    """
    self._websocket_server = websocket_server
    self._on_connect = on_connect
    self._on_disconnect = on_disconnect