Skip to main content

Streaming Methods

Server-to-client streaming​

A hub method that is an async generator becomes a streaming method: each yield sends a StreamItem to the client, and a Completion closes the stream.

import asyncio


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

The client consumes it with stream():

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

Streaming semantics​

  • Items are delivered as they are produced; the generator is driven by a background task so other invocations are not blocked.
  • If the generator raises, the client receives a Completion with the error and its stream() iterator raises HubException.
  • If a non-generator method is invoked as a stream (or vice versa), the server replies with an error Completion.
  • Cancellation: when the client stops consuming early, it sends a CancelInvocation; the server cancels the running generator task.
class StreamingHub(Hub):
    async def infinite(self, n: int, cancellation_token: Any = None) -> ...:
        ...

Client-to-server upload streams​

A hub method can accept an upload stream — the client pushes a sequence of items that the server consumes.

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

Parameter annotation IncomingStream binds the upload stream for that invocation. The server reads items with async for, and the stream ends when the client sends its Completion (or is dropped, which surfaces the error).

Client side:

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 details​

  • Each invocation's upload stream has its own buffered queue (stream_buffer_capacity items).
  • StreamItem messages for a stream id are enqueued as they arrive.
  • A Completion (with or without error) closes the stream.
  • If the connection closes, all open streams are completed with the connection error.

Combining streams​

An invocation can combine both directions: a streaming method whose parameter is IncomingStream produces server-to-client items while consuming client items.