跳到主要内容

调用与流式

本页覆盖客户端能发送的所有方式：invoke、send、stream 与 client_stream。

invoke() — 调用并等待结果​

result = await connection.invoke("Add", 40, 2)
  • 发送带新调用 id 的 Invocation 消息，等待匹配的 Completion。
  • 服务器报错时抛 HubException。
  • 结果到达前连接断开则抛 ConnectionClosed （除非状态化重连重放了这次调用）。
  • 参数可以是任意可序列化为 JSON / MessagePack 的值。

send() — 即发即弃​

await connection.send("Notify", "hello")

发送不带调用 id 的 Invocation，不期待结果，也不等待 Completion。

stream() — 消费流式方法​

async for item in await connection.stream("Counter", 5):
    print(item)

stream() 返回一个异步迭代器：

  • 逐个产出 StreamItem。
  • 流失败时抛 HubException。
  • 服务器发送 Completion 时结束。

如果提前放弃迭代器（如 break 或 aclose()），客户端会发送 CancelInvocation，让服务器停止生产数据：

iterator = await connection.stream("Counter", 1000)
it = iterator.__aiter__()
first = await it.__anext__()
await it.aclose()          # 通知服务器停止

client_stream() — 向服务器上传流​

如果服务器方法接受上传流参数（stream: IncomingStream），客户端可用 client_stream() 发送一串数据。它是异步上下文管理器：

async with connection.client_stream("Upload") as stream:
    await stream.send_item(1)
    await stream.send_item(2)
    await stream.send_item(3)
print("sum:", stream.result)
  • send_item(item) 上传一个数据项（StreamItem 消息）。
  • 退出时自动调用 complete()（发送 Completion），并等待服务器结果。
  • stream.result 保存最终返回值。

调用 id​

客户端为每个非即发即弃的调用分配单调递增的调用 id（1、2、3 …）， 在内部字典中跟踪挂起调用。每个连接内 id 唯一，因此服务器与客户端能无歧义地 关联 StreamItem / Completion / CancelInvocation 消息。

常见陷阱​

  • 不要在非 CONNECTED 状态调用 invoke/stream/send —— 会抛 ConnectionClosed。
  • 参数与结果由协商好的协议序列化；为最大兼容性请使用普通值 （str、int、float、bool、None、dict、list）。
  • 开启状态化重连后，在途的 invoke 能扛过传输断开，因为 Invocation 消息会在重连后被重放。