Skip to main content

Invocations & Streaming

This page covers everything a client can send: invoke, send, stream and client_stream.

invoke() — call and wait for a result​

result = await connection.invoke("Add", 40, 2)
  • Sends an Invocation message with a fresh invocation id and waits for the matching Completion.
  • Raises HubException if the server reports an error.
  • Raises ConnectionClosed if the connection drops before the result arrives (unless stateful reconnect replays the call).
  • Arguments can be any JSON / MessagePack-serializable values.

send() — fire-and-forget​

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

Sends an Invocation with no invocation id. No result is expected and no Completion is awaited.

stream() — consume a streaming method​

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

stream() returns an async iterator:

  • Yields each StreamItem as it arrives.
  • Raises HubException if the stream fails.
  • Ends when the server sends a Completion.

If you abandon the iterator early (e.g. break or aclose()), the client sends a CancelInvocation so the server can stop producing items:

iterator = await connection.stream("Counter", 1000)
it = iterator.__aiter__()
first = await it.__anext__()
await it.aclose()          # the server is told to stop

client_stream() — upload a stream to the server​

If a server method accepts an upload-stream parameter (stream: IncomingStream), the client can send a stream of items with client_stream(). It is an async context manager:

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) uploads one item (a StreamItem message).
  • On exit, complete() is called automatically (sends a Completion), and the manager awaits the server's result.
  • stream.result holds the final return value.

Invocation ids​

The client allocates monotonically increasing invocation ids (1, 2, 3, …) for every non-fire-and-forget call, keeping track of the pending calls in an internal dictionary. Ids are unique per connection, so the server and client can correlate StreamItem / Completion / CancelInvocation messages unambiguously.

Pitfalls​

  • Do not call invoke/stream/send from a state that is not CONNECTED — they raise ConnectionClosed.
  • Arguments and results are serialized by the negotiated protocol; use plain values (str, int, float, bool, None, dict, list) for maximum compatibility.
  • With stateful reconnect enabled, an in-flight invoke survives a transport drop because the Invocation message is replayed after the reconnect.