独立 aiohttp 服务器

SignalRServer.serve() 运行一个基于 aiohttp 的完全自包含 HTTP + WebSocket 服务器，无需外部 ASGI 服务器。

基本用法

import asyncio

from aiosignalr.server import Hub, SignalRServer


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


async def main() -> None:
    server = SignalRServer(ChatHub)
    host, port = await server.serve("127.0.0.1", 8080, path="/hub")
    print(f"listening on http://{host}:{port}/hub")


asyncio.run(main())

serve() 做了什么

1. 构建 aiohttp 应用，注册 {path}/negotiate 与 {path} （WebSocket 升级、SSE、Long Polling、HTTP POST）路由。
2. 绑定 TCP socket。传 port=0 让操作系统挑选空闲端口；返回实际绑定的 (host, port)。
3. 一直运行直到你关闭它。

底层 aiohttp 应用

create_application() 直接返回 aiohttp 的 web.Application，因此可以自定义 中间件或挂载到更大的 aiohttp 应用：

app = server.create_application(path="/hub")
# app.middlewares.append(...)

端点行为

路径 / 方法	用途
POST {path}/negotiate	协商（支持 negotiateVersion 与 useStatefulReconnect）
GET {path}（WebSocket 升级）	全双工 WebSocket 传输
GET {path}（Accept: text/event-stream）	Server-Sent Events 传输
GET {path}	Long Polling 传输
POST {path}	客户端到服务器的 HTTP POST
DELETE {path}	优雅关闭连接

关闭

await server.close()

close() 清理 aiohttp runner。在信号处理器中调用是安全的。

配置

所有旋钮都在 ServerOptions 中：

server = SignalRServer(
    ChatHub,
    options=ServerOptions(
        keep_alive_interval=10.0,
        client_timeout_interval=20.0,
        enable_detailed_errors=True,
        allow_stateful_reconnects=True,
    ),
)