Skip to content

Add support for streaming - #157

Merged
joefreeman merged 35 commits into
mainfrom
streaming
Aug 30, 2026
Merged

Add support for streaming#157
joefreeman merged 35 commits into
mainfrom
streaming

Conversation

@joefreeman

Copy link
Copy Markdown
Collaborator

This adds support for streaming values from a producer to consumers.

A stream can be created explicitly with cf.stream(...), where the argument is a generator. Or implicitly by having the task yield values. A stream can be configured with a buffer size and a timeout. By default there's no buffer or timeout, which means the producer will produce one item at a time to meet demand from consumers. If a buffer is configured, the producer will eagerly produce items to fill the buffer (or continuously if the buffer is set to None). If a timeout is configured then the stream will be closed if no items are produced for that interval - this could be either because items aren't being produced or because items aren't being consumed.

A @task can be configured with streams=cf.Streams(...) (plural) to configure default buffer/timeout for streams (including an implicitly created stream).

A stream can be sliced or partitioned to select the items to be received, and then iterated to consume them - so a stream can be consumed in parallel by multiple consumers.

Item values are persisted to the orchestration server (either as raw values or references to blobs). Executions that have streams that are successfully exhausted (i.e., closed by the producer) can be cached/memoised.

@joefreeman
joefreeman merged commit 154ebdd into main Aug 30, 2026
8 checks passed
@joefreeman
joefreeman deleted the streaming branch August 30, 2026 19:53
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant