hypercorn/asgi/wsproto.py
async def asgi_send(self, message: dict) -> None:
        """Called by the ASGI instance to send a message."""
        if message["type"] == "websocket.accept" and self.state == ASGIWebsocketState.HANDSHAKE:
            headers = build_and_validate_headers(message.get("headers", []))
            raise_if_subprotocol_present(headers)
            headers.extend(self.response_headers())
            await self.asend(
                AcceptConnection(
                    extensions=[PerMessageDeflate()],
                    extra_headers=headers,
                    subprotocol=message.get("subprotocol"),
                )
            )
            self.state = ASGIWebsocketState.CONNECTED
            self.config.access_logger.access(
                self.scope, {"status": 101, "headers": []}, time() - self.start_time
            )
        elif (
            message["type"] == "websocket.http.response.start"
            and self.state == ASGIWebsocketState.HANDSHAKE
        ):
            self.response = message
            self.config.access_logger.access(self.scope, self.response, time() - self.start_time)
        elif message["type"] == "websocket.http.response.body" and self.state in {
            ASGIWebsocketState.HANDSHAKE,
            ASGIWebsocketState.RESPONSE,
        }:
            await self._asgi_send_rejection(message)
        elif message["type"] == "websocket.send" and self.state == ASGIWebsocketState.CONNECTED:
            data: Union[bytes, str]
            if message.get("bytes") is not None:
                await self.asend(BytesMessage(data=bytes(message["bytes"])))
            elif not isinstance(message["text"], str):
                raise TypeError(f"{message['text']} should be a str")
            else:
                await self.asend(TextMessage(data=message["text"]))
        elif message["type"] == "websocket.close" and self.state == ASGIWebsocketState.HANDSHAKE:
            await self.send_http_error(403)
            self.state = ASGIWebsocketState.HTTPCLOSED
        elif message["type"] == "websocket.close":
            await self.asend(CloseConnection(code=int(message["code"])))
            self.state = ASGIWebsocketState.CLOSED
        else:
            raise UnexpectedMessage(self.state, message["type"])
async def asgi_send(self, message: dict) -> None:
        """Called by the ASGI instance to send a message."""
        if message["type"] == "websocket.accept" and self.state == ASGIWebsocketState.HANDSHAKE:
            headers = build_and_validate_headers(message.get("headers", []))
            raise_if_subprotocol_present(headers)
            headers.extend(self.response_headers())
            await self.asend(
                AcceptConnection(
                    extensions=[PerMessageDeflate()],
                    extra_headers=headers,
                    subprotocol=message.get("subprotocol"),
                )
            )
            self.state = ASGIWebsocketState.CONNECTED
            self.config.access_logger.access(
                self.scope, {"status": 101, "headers": []}, time() - self.start_time
            )
        elif (
            message["type"] == "websocket.http.response.start"
            and self.state == ASGIWebsocketState.HANDSHAKE
        ):
            self.response = message
            self.config.access_logger.access(self.scope, self.response, time() - self.start_time)
        elif message["type"] == "websocket.http.response.body" and self.state in {
            ASGIWebsocketState.HANDSHAKE,
            ASGIWebsocketState.RESPONSE,
        }:
            await self._asgi_send_rejection(message)
        elif message["type"] == "websocket.send" and self.state == ASGIWebsocketState.CONNECTED:
            data: Union[bytes, str]
            if message.get("bytes") is not None:
                await self.asend(BytesMessage(data=bytes(message["bytes"])))
            elif not isinstance(message["text"], str):
                raise TypeError(f"{message['text']} should be a str")
            else:
                await self.asend(TextMessage(data=message["text"]))
        elif message["type"] == "websocket.close" and self.state == ASGIWebsocketState.HANDSHAKE:
            await self.send_http_error(403)
            self.state = ASGIWebsocketState.HTTPCLOSED
        elif message["type"] == "websocket.close":
            await self.asend(CloseConnection(code=int(message["code"])))
            self.state = ASGIWebsocketState.CLOSED
        else:
            raise UnexpectedMessage(self.state, message["type"])
async def asgi_send ( self , message : dict ) -> None : if message [ "type" ] == "websocket.accept" and self . state == ASGIWebsocketState . HANDSHAKE : headers = build_and_validate_headers ( message . get ( "headers" , [ ] ) ) raise_if_subprotocol_present ( headers ) headers . extend ( self . response_headers ( ) ) await self . asend ( AcceptConnection ( extensions = [ PerMessageDeflate ( ) ] , extra_headers = headers , subprotocol = message . get ( "subprotocol" ) , ) ) self . state = ASGIWebsocketState . CONNECTED self . config . access_logger . access ( self . scope , { "status" : 101 , "headers" : [ ] } , time ( ) - self . start_time ) elif ( message [ "type" ] == "websocket.http.response.start" and self . state == ASGIWebsocketState . HANDSHAKE ) : self . response = message self . config . access_logger . access ( self . scope , self . response , time ( ) - self . start_time ) elif message [ "type" ] == "websocket.http.response.body" and self . state in { ASGIWebsocketState . HANDSHAKE , ASGIWebsocketState . RESPONSE , } : await self . _asgi_send_rejection ( message ) elif message [ "type" ] == "websocket.send" and self . state == ASGIWebsocketState . CONNECTED : data : Union [ bytes , str ] if message . get ( "bytes" ) is not None : await self . asend ( BytesMessage ( data = bytes ( message [ "bytes" ] ) ) ) elif not isinstance ( message [ "text" ] , str ) : raise TypeError ( f"{message['text']} should be a str" ) else : await self . asend ( TextMessage ( data = message [ "text" ] ) ) elif message [ "type" ] == "websocket.close" and self . state == ASGIWebsocketState . HANDSHAKE : await self . send_http_error ( 403 ) self . state = ASGIWebsocketState . HTTPCLOSED elif message [ "type" ] == "websocket.close" : await self . asend ( CloseConnection ( code = int ( message [ "code" ] ) ) ) self . state = ASGIWebsocketState . CLOSED else : raise UnexpectedMessage ( self . state , message [ "type" ] )
Called by the ASGI instance to send a message.
Called by the ASGI instance to send a message .