聊天服务器示例

这是 examples/chat_server.py 中自带的可运行示例。它演示了生命周期钩子、 分组与广播。

服务器

import asyncio

from aiosignalr.server import Hub, SignalRServer


class ChatHub(Hub):
    """Broadcasts messages to every connected client."""

    async def on_connected(self) -> None:
        await self.clients.all_.send("system", f"{self.context.connection_id} joined")
        await self.groups.add("general")

    async def on_disconnected(self, exception: BaseException | None) -> None:
        await self.clients.all_.send("system", f"{self.context.connection_id} left")

    async def send_message(self, text: str) -> None:
        await self.clients.group("general").send("message", self.context.connection_id, text)

    async def whisper(self, connection_id: str, text: str) -> None:
        await self.clients.client(connection_id).send("message", self.context.connection_id, text)


async def main() -> None:
    server = SignalRServer(ChatHub)
    host, port = await server.serve("127.0.0.1", 8080, path="/hub")
    print(f"Chat hub listening on http://{host}:{port}/hub")
    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        pass
    finally:
        await server.close()


if __name__ == "__main__":
    asyncio.run(main())

运行它：

uv run python examples/chat_server.py
# Chat hub listening on http://127.0.0.1:8080/hub

客户端

import asyncio

from aiosignalr.client import HubConnection


async def main() -> None:
    conn = HubConnection()
    conn.on("system", lambda text: print("[system]", text))
    conn.on("message", lambda sender, text: print(f"[{sender}] {text}"))

    await conn.start("http://127.0.0.1:8080/hub")

    await conn.send("send_message", "hello everyone")

    # 保持客户端存活并响应消息。
    await asyncio.sleep(5)
    await conn.stop()


asyncio.run(main())

本示例覆盖的能力

• on_connected / on_disconnected 生命周期钩子。
• 分组成员（groups.add）与分组广播（clients.group("general").send）。
• 向所有人广播（clients.all_.send）。
• 通过连接 id 定向发送（clients.client(connection_id).send）。