From fa81edac46e84cdad9263d696ff478e762a5667a Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 01:35:12 -0700 Subject: [PATCH 1/5] Added the Jetty v2 examples and ran them on three lanes. --- tests/contrib/streams/examples/README.md | 10 ++ .../examples/migrate_redis_to_native.py | 103 ++++++++++++++++++ .../contrib/streams/examples/test_examples.py | 81 ++++++++++++++ .../contrib/streams/examples/two_backends.py | 89 +++++++++++++++ .../streams/examples/workflow_reader.py | 68 ++++++++++++ 5 files changed, 351 insertions(+) create mode 100644 tests/contrib/streams/examples/migrate_redis_to_native.py create mode 100644 tests/contrib/streams/examples/two_backends.py create mode 100644 tests/contrib/streams/examples/workflow_reader.py diff --git a/tests/contrib/streams/examples/README.md b/tests/contrib/streams/examples/README.md index 9564002ee..1b001933d 100644 --- a/tests/contrib/streams/examples/README.md +++ b/tests/contrib/streams/examples/README.md @@ -13,3 +13,13 @@ Each file is a complete program for one section of the proposal. `test_examples. | Worker setup | [`worker_setup.py`](worker_setup.py) | The proposal's Nexus examples are a later stage, so they have no file here. + +## Jetty v2 + +These add Workflow reads, the native backend, and two backends in one Worker to the same API. `test_examples.py` runs each one on three lanes. The memory and Redis lanes use a memory Store with the backend name `native` in place of the server. The native lane uses `NativeStreamPlugin` against a server with the Stream service, and it runs only when `TEMPORAL_STREAMS_NATIVE_TARGET` is set. + +| Section | File | +|---|---| +| One Worker with Redis and native, one Stream on each | [`two_backends.py`](two_backends.py) | +| A Workflow reads another Workflow's Stream | [`workflow_reader.py`](workflow_reader.py) | +| Old Streams stay on Redis, new ones go to native | [`migrate_redis_to_native.py`](migrate_redis_to_native.py) | diff --git a/tests/contrib/streams/examples/migrate_redis_to_native.py b/tests/contrib/streams/examples/migrate_redis_to_native.py new file mode 100644 index 000000000..2ed793cdd --- /dev/null +++ b/tests/contrib/streams/examples/migrate_redis_to_native.py @@ -0,0 +1,103 @@ +"""Jetty v2, "Move from Redis to native Streams". + +Old Streams stay on Redis, and new ones go to the server. Step 1 runs the old code on a Worker +with only ``RedisStreamPlugin``. Step 2 deploys new code on a Worker with both plugins. The new +code picks ``native`` behind ``workflow.patched``, so a run that started on the old code replays +the same and goes on with its Redis Stream. Each handle carries its backend, so the client reads +each Stream where it is. Redis stays the default, since a change of the default changes the +commands of a replayed create. +""" + +from __future__ import annotations + +import uuid +from datetime import timedelta +from typing import Any + +from temporalio import workflow +from temporalio.client import Client, WorkflowHandle +from temporalio.contrib import streams +from temporalio.contrib.streams import workflow as stream_workflow +from temporalio.contrib.streams.native import NativeStreamPlugin +from temporalio.contrib.streams.redis import RedisStreamPlugin +from temporalio.worker import Worker + + +@streams.defn(name="events") +class Events: + tokens = streams.topic(str) + + +class _Chat: + def __init__(self) -> None: + self._more: list[str] = [] + + @workflow.signal + def more(self, text: str) -> None: + self._more.append(text) + + async def chat(self, stream: stream_workflow.WorkflowStream[Events]) -> str | None: + await stream.topic(Events.tokens).publish("hello") + await workflow.wait_condition(lambda: bool(self._more)) + await stream.topic(Events.tokens).publish(*self._more) + return stream.backend + + +@workflow.defn(name="Chat") +class ChatBeforeTheMove(_Chat): + @workflow.run + async def run(self) -> str | None: + return await self.chat(await stream_workflow.create_stream(Events)) + + +@workflow.defn(name="Chat") +class Chat(_Chat): + @workflow.run + async def run(self) -> str | None: + # A run from before the move has no patch marker, so it keeps its Redis Stream. + backend = "native" if workflow.patched("streams-on-native") else None + return await self.chat( + await stream_workflow.create_stream(Events, backend=backend) + ) + + +async def connect_before(target_host: str, redis_url: str) -> Client: + return await Client.connect(target_host, plugins=[RedisStreamPlugin(redis_url)]) + + +async def connect_during(target_host: str, redis_url: str) -> Client: + return await Client.connect( + target_host, plugins=[RedisStreamPlugin(redis_url), NativeStreamPlugin()] + ) + + +async def read(client: Client, handle: WorkflowHandle[Any, str | None]) -> list[str]: + backend = await handle.result() + stream = streams.StreamClient(client).workflow_stream( + handle, Events, backend=backend + ) + return [ + f"{backend}: {item.data}" + async for item in stream.subscribe() + if isinstance(item, streams.Record) + ] + + +async def main(before: Client, during: Client, task_queue: str) -> list[str]: + # Step 1: the old code starts a chat on Redis. + async with Worker(before, task_queue=task_queue, workflows=[ChatBeforeTheMove]): + old = await before.start_workflow( + ChatBeforeTheMove.run, id=f"chat-{uuid.uuid4()}", task_queue=task_queue + ) + stream = streams.StreamClient(before).workflow_stream(old, Events) + async for _ in stream.subscribe(create_wait=timedelta(seconds=30)): + # The old chat wrote its first Record, so it runs on the old code. + break + # Step 2: the new code goes on with the old chat, and starts a new one on the server. + async with Worker(during, task_queue=task_queue, workflows=[Chat]): + await old.signal(ChatBeforeTheMove.more, "old chat goes on") + new = await during.start_workflow( + Chat.run, id=f"chat-{uuid.uuid4()}", task_queue=task_queue + ) + await new.signal(Chat.more, "new chat") + return [*await read(during, old), *await read(during, new)] diff --git a/tests/contrib/streams/examples/test_examples.py b/tests/contrib/streams/examples/test_examples.py index ec4939196..f6377e113 100644 --- a/tests/contrib/streams/examples/test_examples.py +++ b/tests/contrib/streams/examples/test_examples.py @@ -2,16 +2,26 @@ import uuid +import pytest + from temporalio.client import Client from temporalio.contrib import streams +from temporalio.contrib.streams.native import NativeStreamPlugin +from temporalio.contrib.streams.redis import RedisStreamPlugin +from temporalio.contrib.streams.testing import MemoryStreamPlugin +from temporalio.plugin import SimplePlugin +from tests.contrib.streams.conftest import NATIVE_NAMESPACE, NATIVE_TARGET from tests.contrib.streams.examples import ( activity_streams_tokens, agent_publishes_turns, client_creates_stream, + migrate_redis_to_native, resume_from_offset, turn_scoped_stream, + two_backends, ui_backend_reads, worker_setup, + workflow_reader, ) @@ -89,3 +99,74 @@ async def test_worker_setup_on_redis(client: Client, redis_url: str): ] migrating = await worker_setup.connect_while_migrating(target, redis_url, redis_url) assert await worker_setup.create_on_second_backend(migrating) == "redis-b" + + +# The Jetty v2 examples run with a native stand-in, a memory Store with the backend name +# `native`, and on a server with the Stream service when TEMPORAL_STREAMS_NATIVE_TARGET is set. +LANES = ["memory", "redis", "native"] + + +def lane_plugins( + lane: str, request: pytest.FixtureRequest +) -> tuple[SimplePlugin, SimplePlugin]: + """The plugins of a lane: one named `redis` and one named `native`.""" + if lane == "memory": + return MemoryStreamPlugin(name="redis"), MemoryStreamPlugin(name="native") + redis_url = request.getfixturevalue("redis_url") + redis = RedisStreamPlugin(redis_url, key_prefix=f"t-{uuid.uuid4()}") + if lane == "redis": + return redis, MemoryStreamPlugin(name="native") + return redis, NativeStreamPlugin() + + +async def lane_client(lane: str, client: Client, plugins: list[SimplePlugin]) -> Client: + if lane != "native": + config = client.config() + config["plugins"] = plugins + return Client(**config) + if not NATIVE_TARGET: + pytest.skip( + "TEMPORAL_STREAMS_NATIVE_TARGET names no server with the Stream service" + ) + return await Client.connect( + NATIVE_TARGET, namespace=NATIVE_NAMESPACE, plugins=plugins + ) + + +@pytest.mark.parametrize("lane", LANES) +async def test_two_backends(lane: str, client: Client, request: pytest.FixtureRequest): + two = await lane_client(lane, client, list(lane_plugins(lane, request))) + assert await two_backends.main(two, task_queue()) == { + "audit": ["started", "summarized"], + "live": ["hello", "there", "summary"], + } + + +@pytest.mark.parametrize("lane", LANES) +async def test_workflow_reader( + lane: str, client: Client, request: pytest.FixtureRequest +): + # Each lane reads from one backend: the memory Store, Redis or the server. + plugin = {"memory": 0, "redis": 0, "native": 1}[lane] + one = await lane_client(lane, client, [lane_plugins(lane, request)[plugin]]) + assert await workflow_reader.main(one, task_queue()) == [ + "find", + "the", + "answer", + "agent said FIND THE ANSWER", + ] + + +@pytest.mark.parametrize("lane", LANES) +async def test_migrate_redis_to_native( + lane: str, client: Client, request: pytest.FixtureRequest +): + redis, native = lane_plugins(lane, request) + before = await lane_client(lane, client, [redis]) + during = await lane_client(lane, client, [redis, native]) + assert await migrate_redis_to_native.main(before, during, task_queue()) == [ + "redis: hello", + "redis: old chat goes on", + "native: hello", + "native: new chat", + ] diff --git a/tests/contrib/streams/examples/two_backends.py b/tests/contrib/streams/examples/two_backends.py new file mode 100644 index 000000000..337a19e75 --- /dev/null +++ b/tests/contrib/streams/examples/two_backends.py @@ -0,0 +1,89 @@ +"""Jetty v2, "Two backends in one Worker". + +The client has ``RedisStreamPlugin`` and ``NativeStreamPlugin``, and the Worker takes both from +it. Each Stream handle carries its backend name, so every call goes to the backend that keeps +the Stream. A create with no ``backend=`` goes to the default one, here Redis. Tests swap a +plugin for ``MemoryStreamPlugin(name="native")`` with no change to Workflow code. +""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass +from datetime import timedelta + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.contrib import streams +from temporalio.contrib.streams import workflow as stream_workflow +from temporalio.contrib.streams.native import NativeStreamPlugin +from temporalio.contrib.streams.redis import RedisStreamPlugin +from temporalio.worker import Worker + + +@streams.defn(name="events") +class Events: + tokens = streams.topic(str) + + +@dataclass +class SummarizeInput: + audit: streams.ActivityStream[Events] + live: streams.ActivityStream[Events] + + +@activity.defn +async def summarize(input: SummarizeInput) -> str: + # One Activity writes to both Streams. Each write goes to its own backend. + await input.live.topic(Events.tokens).publish("summary") + await input.audit.topic(Events.tokens).publish("summarized") + return "done" + + +@workflow.defn +class TwoStreams: + @workflow.run + async def run(self) -> str: + # The audit log stays on the default backend, Redis. The live view is on the server. + audit = await stream_workflow.create_stream(Events, name="audit") + live = await stream_workflow.create_stream( + Events, name="live", backend="native" + ) + await audit.topic(Events.tokens).publish("started") + await live.topic(Events.tokens).publish("hello", "there") + return await workflow.execute_activity( + summarize, + SummarizeInput( + streams.ActivityStream.from_handle(audit), + streams.ActivityStream.from_handle(live), + ), + start_to_close_timeout=timedelta(seconds=30), + ) + + +async def connect(target_host: str, redis_url: str) -> Client: + return await Client.connect( + target_host, plugins=[RedisStreamPlugin(redis_url), NativeStreamPlugin()] + ) + + +async def main(client: Client, task_queue: str) -> dict[str, list[str]]: + async with Worker( + client, task_queue=task_queue, workflows=[TwoStreams], activities=[summarize] + ): + handle = await client.start_workflow( + TwoStreams.run, id=f"two-{uuid.uuid4()}", task_queue=task_queue + ) + await handle.result() + stream_client = streams.StreamClient(client) + read: dict[str, list[str]] = {} + for name, backend in (("audit", "redis"), ("live", "native")): + stream = stream_client.workflow_stream( + handle, Events, name=name, backend=backend + ) + read[name] = [ + item.data + async for item in stream.subscribe() + if isinstance(item, streams.Record) + ] + return read diff --git a/tests/contrib/streams/examples/workflow_reader.py b/tests/contrib/streams/examples/workflow_reader.py new file mode 100644 index 000000000..f5a70b2b1 --- /dev/null +++ b/tests/contrib/streams/examples/workflow_reader.py @@ -0,0 +1,68 @@ +"""Jetty v2, "A Workflow reads another Workflow's Stream". + +An orchestrator Workflow follows the turns of an agent Workflow. It gets the agent's Stream as +an argument and reads it with ``subscribe``, an async iterator in Workflow code. Each page is in +History, so a replay reads the same items. On a native Stream a read at the end waits on the +server. On Redis it waits on a Local Activity, then on a durable timer. The Workflow code is +the same on both. +""" + +from __future__ import annotations + +import uuid +from datetime import timedelta + +from temporalio import workflow +from temporalio.client import Client +from temporalio.contrib import streams +from temporalio.contrib.streams import workflow as stream_workflow +from temporalio.worker import Worker + + +@streams.defn(name="events") +class Events: + tokens = streams.topic(str) + + +@workflow.defn +class Agent: + @workflow.run + async def run(self, prompt: str) -> str: + stream = await stream_workflow.create_stream(Events) + for word in prompt.split(): + # Each word goes out in its own Workflow Task, so the follower waits for some. + await workflow.sleep(timedelta(milliseconds=200)) + await stream.topic(Events.tokens).publish(word) + return prompt.upper() + + +@workflow.defn +class Follower: + @workflow.run + async def run(self, agent: stream_workflow.WorkflowStream[Events]) -> list[str]: + seen = [] + async for item in agent.subscribe(create_wait=timedelta(seconds=30)): + if isinstance(item, streams.Record): + seen.append(item.data) + elif isinstance(item, streams.Closed): + # The agent's Stream closes with the agent's result. + seen.append(f"agent said {item.result}") + return seen + + +async def main(client: Client, task_queue: str) -> list[str]: + async with Worker(client, task_queue=task_queue, workflows=[Agent, Follower]): + agent = await client.start_workflow( + Agent.run, + "find the answer", + id=f"agent-{uuid.uuid4()}", + task_queue=task_queue, + ) + # The handle has the first run id, so the client builds the Stream with no call. + stream = streams.StreamClient(client).workflow_stream(agent, Events) + return await client.execute_workflow( + Follower.run, + stream_workflow.WorkflowStream.from_handle(stream), + id=f"follower-{uuid.uuid4()}", + task_queue=task_queue, + ) From cba68d54f744422e1472375a8d529927414de8d2 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 02:24:19 -0700 Subject: [PATCH 2/5] Read a moving chat by the backend lookup in the migration example. --- .../examples/migrate_redis_to_native.py | 25 +++++++++---------- .../contrib/streams/examples/test_examples.py | 7 +++--- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/tests/contrib/streams/examples/migrate_redis_to_native.py b/tests/contrib/streams/examples/migrate_redis_to_native.py index 2ed793cdd..75e648c3d 100644 --- a/tests/contrib/streams/examples/migrate_redis_to_native.py +++ b/tests/contrib/streams/examples/migrate_redis_to_native.py @@ -36,29 +36,26 @@ def __init__(self) -> None: def more(self, text: str) -> None: self._more.append(text) - async def chat(self, stream: stream_workflow.WorkflowStream[Events]) -> str | None: + async def chat(self, stream: stream_workflow.WorkflowStream[Events]) -> None: await stream.topic(Events.tokens).publish("hello") await workflow.wait_condition(lambda: bool(self._more)) await stream.topic(Events.tokens).publish(*self._more) - return stream.backend @workflow.defn(name="Chat") class ChatBeforeTheMove(_Chat): @workflow.run - async def run(self) -> str | None: - return await self.chat(await stream_workflow.create_stream(Events)) + async def run(self) -> None: + await self.chat(await stream_workflow.create_stream(Events)) @workflow.defn(name="Chat") class Chat(_Chat): @workflow.run - async def run(self) -> str | None: + async def run(self) -> None: # A run from before the move has no patch marker, so it keeps its Redis Stream. backend = "native" if workflow.patched("streams-on-native") else None - return await self.chat( - await stream_workflow.create_stream(Events, backend=backend) - ) + await self.chat(await stream_workflow.create_stream(Events, backend=backend)) async def connect_before(target_host: str, redis_url: str) -> Client: @@ -71,16 +68,18 @@ async def connect_during(target_host: str, redis_url: str) -> Client: ) -async def read(client: Client, handle: WorkflowHandle[Any, str | None]) -> list[str]: - backend = await handle.result() +async def read(client: Client, handle: WorkflowHandle[Any, None]) -> list[str]: + await handle.result() + # A UI that reads chats from both sides of the move looks on native first, then Redis. stream = streams.StreamClient(client).workflow_stream( - handle, Events, backend=backend + handle, Events, backend=("native", "redis") ) - return [ - f"{backend}: {item.data}" + tokens = [ + item.data async for item in stream.subscribe() if isinstance(item, streams.Record) ] + return [f"{stream.backend}: {token}" for token in tokens] async def main(before: Client, during: Client, task_queue: str) -> list[str]: diff --git a/tests/contrib/streams/examples/test_examples.py b/tests/contrib/streams/examples/test_examples.py index a71bf6f75..85ede5678 100644 --- a/tests/contrib/streams/examples/test_examples.py +++ b/tests/contrib/streams/examples/test_examples.py @@ -9,7 +9,6 @@ from temporalio.contrib.streams.native import NativeStreamPlugin from temporalio.contrib.streams.redis import RedisStreamPlugin from temporalio.contrib.streams.testing import MemoryStreamPlugin -from temporalio.plugin import SimplePlugin from tests.contrib.streams.conftest import NATIVE_NAMESPACE, NATIVE_TARGET from tests.contrib.streams.examples import ( activity_streams_tokens, @@ -109,7 +108,7 @@ async def test_worker_setup_on_redis(client: Client, redis_url: str): def lane_plugins( lane: str, request: pytest.FixtureRequest -) -> tuple[SimplePlugin, SimplePlugin]: +) -> tuple[streams.StreamPlugin, streams.StreamPlugin]: """The plugins of a lane: one named `redis` and one named `native`.""" if lane == "memory": return MemoryStreamPlugin(name="redis"), MemoryStreamPlugin(name="native") @@ -120,7 +119,9 @@ def lane_plugins( return redis, NativeStreamPlugin() -async def lane_client(lane: str, client: Client, plugins: list[SimplePlugin]) -> Client: +async def lane_client( + lane: str, client: Client, plugins: list[streams.StreamPlugin] +) -> Client: if lane != "native": config = client.config() config["plugins"] = plugins From efb043ad891a413e02798492b221c483a1cbd321 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 02:42:22 -0700 Subject: [PATCH 3/5] Skipped the native example lane before building its plugin. --- tests/contrib/streams/examples/test_examples.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/contrib/streams/examples/test_examples.py b/tests/contrib/streams/examples/test_examples.py index 85ede5678..948450144 100644 --- a/tests/contrib/streams/examples/test_examples.py +++ b/tests/contrib/streams/examples/test_examples.py @@ -112,6 +112,8 @@ def lane_plugins( """The plugins of a lane: one named `redis` and one named `native`.""" if lane == "memory": return MemoryStreamPlugin(name="redis"), MemoryStreamPlugin(name="native") + if lane == "native" and not NATIVE_TARGET: + pytest.skip("TEMPORAL_STREAMS_NATIVE_TARGET names no server with the Stream service") redis_url = request.getfixturevalue("redis_url") redis = RedisStreamPlugin(redis_url, key_prefix=f"t-{uuid.uuid4()}") if lane == "redis": From d3f9c3922fe61c4c400590d9a527ac00fcbff359 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 02:42:47 -0700 Subject: [PATCH 4/5] Formatted the example lane skip. --- tests/contrib/streams/examples/test_examples.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/contrib/streams/examples/test_examples.py b/tests/contrib/streams/examples/test_examples.py index 948450144..25c4f07cd 100644 --- a/tests/contrib/streams/examples/test_examples.py +++ b/tests/contrib/streams/examples/test_examples.py @@ -113,7 +113,9 @@ def lane_plugins( if lane == "memory": return MemoryStreamPlugin(name="redis"), MemoryStreamPlugin(name="native") if lane == "native" and not NATIVE_TARGET: - pytest.skip("TEMPORAL_STREAMS_NATIVE_TARGET names no server with the Stream service") + pytest.skip( + "TEMPORAL_STREAMS_NATIVE_TARGET names no server with the Stream service" + ) redis_url = request.getfixturevalue("redis_url") redis = RedisStreamPlugin(redis_url, key_prefix=f"t-{uuid.uuid4()}") if lane == "redis": From c0fcc8da93c9c5cfd0cda4ab95a0ff35f1021041 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 14:23:47 -0700 Subject: [PATCH 5/5] Named the two example Streams with `streams.with_name`. --- .../contrib/streams/examples/two_backends.py | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/tests/contrib/streams/examples/two_backends.py b/tests/contrib/streams/examples/two_backends.py index 337a19e75..e58152487 100644 --- a/tests/contrib/streams/examples/two_backends.py +++ b/tests/contrib/streams/examples/two_backends.py @@ -26,6 +26,11 @@ class Events: tokens = streams.topic(str) +# Two Streams of one definition, under their own names. +Audit = streams.with_name(Events, "audit") +Live = streams.with_name(Events, "live") + + @dataclass class SummarizeInput: audit: streams.ActivityStream[Events] @@ -45,10 +50,8 @@ class TwoStreams: @workflow.run async def run(self) -> str: # The audit log stays on the default backend, Redis. The live view is on the server. - audit = await stream_workflow.create_stream(Events, name="audit") - live = await stream_workflow.create_stream( - Events, name="live", backend="native" - ) + audit = await stream_workflow.create_stream(Audit) + live = await stream_workflow.create_stream(Live, backend="native") await audit.topic(Events.tokens).publish("started") await live.topic(Events.tokens).publish("hello", "there") return await workflow.execute_activity( @@ -77,10 +80,11 @@ async def main(client: Client, task_queue: str) -> dict[str, list[str]]: await handle.result() stream_client = streams.StreamClient(client) read: dict[str, list[str]] = {} - for name, backend in (("audit", "redis"), ("live", "native")): - stream = stream_client.workflow_stream( - handle, Events, name=name, backend=backend - ) + for name, definition, backend in ( + ("audit", Audit, "redis"), + ("live", Live, "native"), + ): + stream = stream_client.workflow_stream(handle, definition, backend=backend) read[name] = [ item.data async for item in stream.subscribe()