跳到主要内容

流式方法

服务器到客户端的流式​

异步生成器方法就是流式方法：每次 yield 向客户端发送一个 StreamItem， Completion 关闭流。

import asyncio


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

客户端用 stream() 消费：

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

流式语义​

  • 数据项生产即投递；生成器由后台任务驱动，不阻塞其他调用。
  • 生成器抛异常时，客户端收到带错误的 Completion，其 stream() 迭代器抛 HubException。
  • 以流式方式调用非生成器方法（或反之），服务器回复错误 Completion。
  • 取消：客户端提前停止消费时发送 CancelInvocation；服务器取消正在运行的 生成器任务。
class StreamingHub(Hub):
    async def infinite(self, n: int, cancellation_token: Any = None) -> ...:
        ...

客户端上传流​

hub 方法可以接受上传流 —— 客户端推送一串数据，服务器逐个消费。

from aiosignalr.server import IncomingStream


class UploadHub(Hub):
    async def upload_sum(self, stream: IncomingStream) -> int:
        total = 0
        async for item in stream:
            total += int(item)
        return total

参数注解 IncomingStream 绑定该调用的上传流。服务器用 async for 读取数据项， 客户端发送 Completion（或连接断开，会透出错误）时流结束。

客户端侧：

async with connection.client_stream("upload_sum") as stream:
    await stream.send_item(1)
    await stream.send_item(2)
print("sum:", stream.result)  # 3

IncomingStream 细节​

  • 每次调用的上传流有独立的缓冲队列（stream_buffer_capacity 个数据项）。
  • 流 id 对应的 StreamItem 消息到达即入队。
  • Completion（带或不带错误）关闭流。
  • 连接关闭时，所有打开的流以连接错误结束。

组合流​

一次调用可以同时使用两个方向：参数是 IncomingStream 的流式方法，既能消费 客户端数据项，又能向客户端产出数据。