跳到主要内容

ASGI 部署

SignalRServer.asgi_app() 返回一个标准 ASGI 应用，为你的 hub 提供 negotiate / WebSocket / SSE / Long Polling / POST 端点。可以挂载到 uvicorn、FastAPI 或任何 ASGI 服务器。

独立 uvicorn​

import uvicorn

from aiosignalr.server import Hub, SignalRServer


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


server = SignalRServer(ChatHub)
app = server.asgi_app(path="/hub")

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

在 FastAPI 内​

from fastapi import FastAPI

from aiosignalr.server import Hub, SignalRServer


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


app = FastAPI()
server = SignalRServer(ChatHub)
app.mount("/hub", server.asgi_app(path="/hub"))

hub 现在位于 http://host:8000/hub。

挂载到现有 ASGI 树​

因为 asgi_app() 本身是普通 ASGI 应用，你可以用任何路由层组合它。 端点路径完全由 path 控制。

要求​

  • ASGI 服务器必须支持 WebSocket 升级（uvicorn 支持；部分生产代理需要开启 WebSocket 支持）。
  • 状态化重连要求底层 ASGI 服务器保持连接打开，并允许跨重连复用同一个连接 token（uvicorn 可以处理）。

生命周期说明​

返回的应用是无状态、进程内的。如果要在多 worker 或多机间共享 hub 状态， 请接入自定义生命周期管理器 —— 见 HubLifetimeManager。