samples: workflow_stream: add external-publisher scenario

Adds a third scenario covering the third publisher shape: a backend
service or scheduled job pushing events into a workflow it didn't
itself start. The earlier scenarios publish either from inside the
workflow or from one of its activities; this one uses
WorkflowStreamClient.create() externally.

HubWorkflow is a passive stream host — it does no work of its own and
just waits to be told to close, fitting the event-bus pattern. The
runner publishes a series of news headlines, runs a subscriber task
alongside, signals close, and exits when both tasks complete.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jssmith

workflow_stream/README.md

@@ -40,7 +40,21 @@ This directory has two scenarios sharing one Worker.
   disappear (page refresh, server restart, laptop closed) and resume
   later without missing events or seeing duplicates.
 
-`run_worker.py` registers both workflows and the activity.
+**Scenario 3 — external (non-Activity) publisher:**
+
+* `workflows/hub_workflow.py` — a passive workflow that does no work
+  of its own; it exists only to host a `WorkflowStream` and shut down
+  when signaled.
+* `run_external_publisher.py` — starts the hub, then publishes events
+  into it from a plain Python coroutine using
+  `WorkflowStreamClient.create(client, workflow_id)`. A subscriber
+  task runs alongside; when the publisher is done it signals
+  `HubWorkflow.close`, the workflow's run finishes, and the
+  subscriber's iterator exits normally. This is the shape that fits a
+  backend service or scheduled job pushing events into a workflow it
+  didn't itself start.
+
+`run_worker.py` registers all three workflows and the activity.
 
 ## Run it
 
@@ -52,6 +66,8 @@ uv run workflow_stream/run_worker.py
 uv run workflow_stream/run_publisher.py
 # or
 uv run workflow_stream/run_reconnecting_subscriber.py
+# or
+uv run workflow_stream/run_external_publisher.py
 ```
 
 Expected output on the basic publisher side:

workflow_stream/run_external_publisher.py

@@ -0,0 +1,91 @@
+"""External publisher: a non-Activity process pushes events into a workflow.
+
+The two earlier scenarios publish from inside the workflow itself
+(``OrderWorkflow``, ``PipelineWorkflow``) or from an Activity it runs
+(``charge_card``). This scenario shows the third shape: a backend
+service, scheduled job, or anything else with a Temporal ``Client``
+publishing into a *running* workflow it didn't start. Same factory as
+the subscribe path — :py:meth:`WorkflowStreamClient.create` — used for
+publishing instead.
+
+The script starts a ``HubWorkflow`` (which does no work of its own —
+it exists only to host the stream), then runs a publisher and a
+subscriber concurrently. When the publisher is done it signals
+``HubWorkflow.close``, the workflow's run finishes, and the
+subscriber's iterator exits normally.
+
+Run the worker first (``uv run workflow_stream/run_worker.py``), then::
+
+    uv run workflow_stream/run_external_publisher.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+
+from temporalio.client import Client
+from temporalio.contrib.workflow_stream import WorkflowStreamClient
+
+from workflow_stream.shared import (
+    TASK_QUEUE,
+    TOPIC_NEWS,
+    HubInput,
+    NewsEvent,
+)
+from workflow_stream.workflows.hub_workflow import HubWorkflow
+
+
+HEADLINES = [
+    "rates held",
+    "merger announced",
+    "outage resolved",
+    "earnings beat",
+    "regulator opens probe",
+]
+
+
+async def main() -> None:
+    client = await Client.connect("localhost:7233")
+
+    workflow_id = f"workflow-stream-hub-{uuid.uuid4().hex[:8]}"
+    handle = await client.start_workflow(
+        HubWorkflow.run,
+        HubInput(hub_id=workflow_id),
+        id=workflow_id,
+        task_queue=TASK_QUEUE,
+    )
+
+    async def publish_news() -> None:
+        # WorkflowStreamClient.create takes a Temporal client and a
+        # workflow id — the same factory used elsewhere for subscribing.
+        # The async context manager batches publishes and flushes on
+        # exit; we additionally call flush() before signaling close so
+        # we know the events landed before the workflow shuts down.
+        producer = WorkflowStreamClient.create(client, workflow_id)
+        async with producer:
+            for headline in HEADLINES:
+                producer.publish(TOPIC_NEWS, NewsEvent(headline=headline))
+                print(f"[publisher] sent: {headline}")
+                await asyncio.sleep(0.5)
+            await producer.flush()
+        # Tell the hub it can stop. The workflow's run() returns, and
+        # any in-flight subscribers see their async-for loop exit.
+        await handle.signal(HubWorkflow.close)
+        print("[publisher] signaled close")
+
+    async def consume_news() -> None:
+        consumer = WorkflowStreamClient.create(client, workflow_id)
+        async for item in consumer.subscribe(
+            [TOPIC_NEWS], result_type=NewsEvent
+        ):
+            print(f"[subscriber] offset={item.offset}: {item.data.headline}")
+
+    await asyncio.gather(publish_news(), consume_news())
+
+    result = await handle.result()
+    print(f"\nworkflow result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

workflow_stream/run_worker.py

@@ -8,6 +8,7 @@
 
 from workflow_stream.activities.payment_activity import charge_card
 from workflow_stream.shared import TASK_QUEUE
+from workflow_stream.workflows.hub_workflow import HubWorkflow
 from workflow_stream.workflows.order_workflow import OrderWorkflow
 from workflow_stream.workflows.pipeline_workflow import PipelineWorkflow
 
@@ -18,7 +19,7 @@ async def main() -> None:
     worker = Worker(
         client,
         task_queue=TASK_QUEUE,
-        workflows=[OrderWorkflow, PipelineWorkflow],
+        workflows=[HubWorkflow, OrderWorkflow, PipelineWorkflow],
         activities=[charge_card],
     )
     await worker.run()

workflow_stream/shared.py

@@ -13,6 +13,7 @@
 # Topics published by the workflow / activity.
 TOPIC_STATUS = "status"
 TOPIC_PROGRESS = "progress"
+TOPIC_NEWS = "news"
 
 
 @dataclass
@@ -45,6 +46,18 @@ class StageEvent:
     stage: str
 
 
+@dataclass
+class HubInput:
+    hub_id: str
+    # Carries stream state across continue-as-new. None on a fresh start.
+    stream_state: WorkflowStreamState | None = None
+
+
+@dataclass
+class NewsEvent:
+    headline: str
+
+
 T = TypeVar("T")
 
 

workflow_stream/workflows/hub_workflow.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from temporalio import workflow
+from temporalio.contrib.workflow_stream import WorkflowStream
+
+from workflow_stream.shared import HubInput
+
+
+@workflow.defn
+class HubWorkflow:
+    """Passive stream host: starts up, waits, closes when told.
+
+    Unlike OrderWorkflow or PipelineWorkflow, this workflow does no
+    work of its own — it exists only to host a ``WorkflowStream`` that
+    external publishers push events into and external subscribers read
+    from. The shape that fits a backend service or "event bus" pattern,
+    where the workflow owns durable state but the events come from
+    outside.
+    """
+
+    @workflow.init
+    def __init__(self, input: HubInput) -> None:
+        self.stream = WorkflowStream(prior_state=input.stream_state)
+        self._closed = False
+
+    @workflow.run
+    async def run(self, input: HubInput) -> str:
+        await workflow.wait_condition(lambda: self._closed)
+        return f"hub {input.hub_id} closed"
+
+    @workflow.signal
+    def close(self) -> None:
+        # Custom signal handler that does not read stream state, so the
+        # synchronous-handler race documented in the README does not
+        # apply.
+        self._closed = True