跳到主要内容

快速开始：服务端

本教程构建一个 Python SignalR hub 并通过 WebSocket 提供出来。

1. 定义 hub​

hub 是 Hub 的子类。每个公共方法都可被客户端调用； 返回值会作为结果回发；用 yield 的方法会成为流式方法。

import asyncio

from aiosignalr.server import Hub, SignalRServer


class ChatHub(Hub):
    async def on_connected(self) -> None:
        await self.clients.all_.send(
            "message", f"User {self.context.connection_id} joined"
        )

    async def echo(self, text: str) -> str:
        return text

    async def counter(self, n: int):
        for i in range(n):
            yield i
            await asyncio.sleep(0.01)


async def main() -> None:
    server = SignalRServer(ChatHub)
    await server.serve("127.0.0.1", 8080, path="/hub")


asyncio.run(main())

特殊参数​

可以通过参数名注入连接上下文与取消能力：

  • context / hub_context → HubContext， 含 connection_id、请求头和查询串。
  • cancellation_token / cancellationToken → 一个 asyncio.Event， 客户端取消调用时会被置位。
class MyHub(Hub):
    async def slow_operation(self, n: int, cancellation_token: Any = None) -> int:
        for i in range(n):
            await asyncio.sleep(0.05)
            if cancellation_token is not None and cancellation_token.is_set():
                return -1
        return n

2. 选择宿主​

SignalRServer.serve() 运行一个独立的 aiohttp 应用：

await server.serve("127.0.0.1", 8080, path="/hub")

也可以把服务器挂载到现有 ASGI 应用（uvicorn / FastAPI）：

app = server.asgi_app(path="/hub")
# 把 app 传给 uvicorn.run(app, ...) 或 FastAPI

详见 部署。

3. 连接客户端​

现在用任何 SignalR 客户端连接 hub。以 aiosignalr 为例：

import asyncio

from aiosignalr.client import HubConnection


async def main() -> None:
    conn = HubConnection()
    conn.on("message", lambda text: print("server says:", text))
    await conn.start("http://127.0.0.1:8080/hub")

    print(await conn.invoke("echo", "ping"))      # "ping"
    async for item in await conn.stream("counter", 3):
        print("counter:", item)

    await conn.stop()


asyncio.run(main())

下一步​