Skip to content

[Brook] Added the Activity and client stream handles over Core. #154

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
moetemp wants to merge 9 commits into from
Closed

[Brook] Added the Activity and client stream handles over Core. #154

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
cae0e2a
Added the Activity and client stream handles over Core.
moetemp Oct 13, 2026
e9d47b7
Merged p2e-provider-plugin into p3b-outside-handles.
moetemp Oct 14, 2026
09236ce
Said that SUPERSEDED withdraws one producer's records, not others'.
moetemp Oct 14, 2026
b9d2eb6
Removed run pinning from stream handles, so each read follows the chain.
moetemp Oct 14, 2026
8fecaa9
Merged p2e-provider-plugin into p3b-outside-handles.
moetemp Oct 14, 2026
59bc227
Let a handle name its chain by a run, and read it to the chain's end.
moetemp Oct 14, 2026
253c338
Merged p2e-provider-plugin into p3b-outside-handles.
moetemp Oct 14, 2026
6d17ae7
Merged p2e-provider-plugin into p3b-outside-handles.
moetemp Oct 14, 2026
5358116
Merged p2e-provider-plugin into p3b-outside-handles.
moetemp Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog/added/pirouetting-platypus.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Add experimental `temporalio.contrib.streams` so Activities and clients can append ordered records to a Workflow's stream in Redis, and outside readers can follow it from a cursor.
25 changes: 24 additions & 1 deletion temporalio/contrib/streams/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,18 @@

A store is configured as a plugin:
``Client.connect(..., plugins=[RedisStreams(...)])`` registers it on the
client and on every Worker built from that client.
client and on every Worker built from that client. Each
context then reaches a stream the same way:

- An Activity reaches the stream of the Workflow that scheduled it with
:func:`activity_handle`, and writes as itself: its Activity id with the
scheduling run id, and its Temporal attempt.
- Any process holding a client reaches a Workflow's stream with
:func:`get_stream_handle`, and writes with a producer id and an attempt of
its own.

Every read follows the Workflow's run chain across Continue-as-New and ends
when the chain closes. A reader can't pin one run.

The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in
:mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body``
Expand All @@ -55,6 +66,13 @@
StreamStorageError,
StreamUnsupportedError,
)
from temporalio.contrib.streams._handles import (
ActivityStreamHandle,
StreamHandle,
StreamProducer,
activity_handle,
get_stream_handle,
)
from temporalio.contrib.streams._plugin import StreamStorePlugin
from temporalio.contrib.streams._record import (
Cursor,
Expand All @@ -66,6 +84,7 @@
from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic

__all__ = [
"ActivityStreamHandle",
"BEGINNING",
"Cursor",
"DEFAULT_TOPIC",
Expand All @@ -75,9 +94,11 @@
"StreamCursorError",
"StreamError",
"StreamExpiredError",
"StreamHandle",
"StreamNotFoundError",
"StreamOutcomeUnknownError",
"StreamOwnerKind",
"StreamProducer",
"StreamProducerError",
"StreamRecord",
"StreamRecordError",
Expand All @@ -88,5 +109,7 @@
"StreamTopic",
"StreamUnsupportedError",
"Supersession",
"activity_handle",
"get_stream_handle",
"topic",
]
Loading
Loading