INNER CODE UNIT · Python

port

ImFeH2/huddol · core/src/fora/adapters/websocket/server.py:364

    def port(self) -> int:
        return self._port

    def add_websocket_route(
        self, path: str, handler: SocketHandler, *, max_msg_size: int
    ) -> None:
        if self._thread.is_alive():
            raise RuntimeError("Register routes before starting the server")

        async def connected(request: web.Request) -> web.StreamResponse:
            self._authenticate(request, websocket=True)
            connection = web.WebSocketResponse(
                max_msg_size=max_msg_size,
                compress=False,
                heartbeat=20 if path == "/ws" else None,
            )
            if (
                path == "/ws"

View source record →