WebSocket

Bases: Protocol

WebSocket.

Websocket 实例可以使用异步迭代器接收消息，直到连接关闭：

async for message in websocket:
    ...

或者通过直接调用 recv()：

message = await websocket.recv()

发送消息使用 send()：

await websocket.send(message)

Source code in ypy_websocket/websocket.py

class Websocket(Protocol):
    """WebSocket.

    Websocket 实例可以使用异步迭代器接收消息，直到连接关闭：

    ```py
    async for message in websocket:
        ...
    ```

    或者通过直接调用 `recv()`：

    ```py
    message = await websocket.recv()
    ```

    发送消息使用 `send()`：

    ```py
    await websocket.send(message)
    ```
    """

    @property
    def path(self) -> str:
        """WebSocket 路径"""
        ...

    def __aiter__(self):
        return self

    async def __anext__(self) -> bytes:
        try:
            message = await self.recv()
        except Exception:
            raise StopAsyncIteration()

        return message

    async def send(self, message: bytes) -> None:
        """发送消息。

        Arguments:
            message: 要发送的消息。
        """
        ...

    async def recv(self) -> bytes:
        """收到一条消息。

        Returns:
            收到的消息。
        """
        ...

path: str property

WebSocket 路径

recv() async

收到一条消息。

Returns:

Type	Description
bytes	收到的消息。

Source code in ypy_websocket/websocket.py

async def recv(self) -> bytes:
    """收到一条消息。

    Returns:
        收到的消息。
    """
    ...

send(message) async

发送消息。

Parameters:

Name	Type	Description	Default
message	bytes	要发送的消息。	required

Source code in ypy_websocket/websocket.py

async def send(self, message: bytes) -> None:
    """发送消息。

    Arguments:
        message: 要发送的消息。
    """
    ...