Publish events from a client to a Workflow Stream
Any process with a Temporal Client and target Workflow Id can publish by constructing a workflowstreams.NewClient(temporalClient, workflowID, workflowstreams.Options{}). Use it the same way as the Workflow-side handle: bind a topic, publish through it, and defer client.Close(ctx) to flush on scope exit. Options include BatchInterval (default unspecified) and MaxRetryDuration (default 10 minutes). Inside an Activity, use workflowstreams.NewClientFromActivity(ctx, workflowstreams.Options{}) to infer the Temporal Client and parent Workflow Id from the Activity context.
Subscribe to Workflow Stream with ack handshake code example
```go
func StreamChat(ctx context.Context, temporalClient client.Client, chatID string) (string, error) {
stream := workflowstreams.NewClient(temporalClient, chatID, workflowstreams.Options{})
dc := converter.GetDefaultDataConverter()
var output []string
render := func() {
// display the accumulated output
}
for item, err := range stream.Subscribe(ctx, workflowstreams.SubscribeOptions{Topics: []string{"delta", "retry", "close"}}) {
if err != nil {
return "", err
}
switch item.Topic {
case "retry":
output = output[:0]
render()
case "delta":
var delta TextDelta
if err := dc.FromPayload(item.Data, &delta); err != nil {
return "", err
}
output = append(output, delta.Text)
render()
case "close":
if err := temporalClient.SignalWorkflow(ctx, chatID, "", "subscriber-acknowledged-terminator", nil); err != nil {
return "", err
}
return strings.Join(output, ""), nil
}
}
return strings.Join(output, ""), nil
}
```
Control batch flushing in Workflow Streams with forceFlush and Flush
Pass true as the forceFlush argument on Publish() to wake the background flusher so the current buffer ships without waiting for the next interval. The call returns immediately and doesn't wait for delivery. Use it for latency-sensitive events like the first delta of a response. Call client.Flush(ctx) when you need a mid-stream barrier. Successful completion proves the Temporal server received all prior publications. Close() already flushes on exit, so explicit flush is only needed for barriers in the middle.
Subscribe to Workflow Stream events
Use workflowstreams.NewClient(temporalClient, workflowID, opts) from any process with a Temporal Client. Call client.Subscribe(ctx, workflowstreams.SubscribeOptions{Topics: []string{...}}) which returns an iter.Seq2 iterator yielding WorkflowStreamItem and error. Each item's Data is the raw payload; decode it with a payload converter. SubscribeOptions controls the subscription: Topics filters by name (empty or nil means all topics), FromOffset resumes from a stored offset (zero means beginning), and PollCooldown sets the minimum interval between polls. A single-topic convenience method streamClient.Topic("name").Subscribe(ctx, fromOffset) is equivalent to passing one name in Topics.
Subscribe to multiple topics with different payload types
Every item arrives as a raw *commonpb.Payload in item.Data, so a single subscription naturally consumes multiple topics with different payload types. Pass the topic names in SubscribeOptions.Topics or leave it empty for every topic. Dispatch on item.Topic and decode into the matching type using converter.GetDefaultDataConverter().FromPayload(item.Data, &evt).
Close a Workflow Stream properly
A subscriber's for...range loop doesn't know when the publisher is done. Use two approaches: 1) An in-band terminator where the Workflow publishes a sentinel event the subscriber recognizes and breaks on. 2) A brief overlap before Workflow returns. For fixed sleep approach, add workflow.Sleep(ctx, 30*time.Second) between the terminator and return. For acknowledgment handshake, the subscriber sends a Signal once it has the terminator and the Workflow waits up to a timeout using workflow.AwaitWithTimeout(). After the loop returns, call temporalClient.DescribeWorkflowExecution(ctx, workflowID, "") to inspect terminal status if needed.
Enable Workflow Streams in Go
Construct a WorkflowStream exactly once at the top of the Workflow function, before any blocking call. Use the library `go.temporal.io/sdk/contrib/workflowstreams`. Call `workflowstreams.NewWorkflowStream(ctx, input.StreamState)` where StreamState is nil on a fresh start and a *WorkflowStreamState after a Continue-As-New rollover. The constructor registers the publish Signal, subscribe Update, and offset Query handlers. Do not construct more than one WorkflowStream per Workflow, as this registers duplicate handlers.
Custom Continue-As-New parameters with Workflow Streams
To pass other Continue-As-New parameters such as a different task queue or custom publisher TTL, use the explicit recipe: Call stream.DetachPollers(), then workflow.Await(ctx, func() bool { return workflow.AllHandlersFinished(ctx) }), then state, err := stream.GetState(publisherTTL) with custom TTL (default 15 minutes). Set options such as task queue on context first with workflow.WithWorkflowTaskQueue(ctx, "other-tq"), then return workflow.NewContinueAsNewError(ctx, wfn, args).
Go Workflow Streams payload conversion
item.Data is always the raw payload. Decode it with a converter built from the same PayloadConverters used by the publisher. When publishers and subscribers both rely on defaults, converter.GetDefaultDataConverter() matches on both sides. If passing WithPayloadConverters on the Workflow side, build a matching converter.NewCompositeDataConverter(...) on the subscriber side. The codec chain runs once on the envelope; Payload codecs (encryption, compression) configured on the Temporal client run on the Signal or Update envelope carrying each batch, never per item.
Goroutine safety in Workflow Streams client publish
The client publish path is not goroutine-safe. The client buffer is mutated on the publish path and read from the background flusher. Do not call Publish() on the same Client from multiple goroutines without coordinating; route events to a single owner.
Go Workflow Streams example: Stream LLM output with activity
The Activity calls the model and publishes deltas as they arrive. The Workflow starts the Activity and waits for the consumer to acknowledge end-of-stream. The consumer subscribes, accumulates deltas, and clears state on RETRY before continuing. The Activity publishes a RETRY event when activity.GetInfo(ctx).Attempt > 1 to let the UI respond to failures. Termination uses an ack handshake: the consumer signals once it has the close event, so the Workflow returns as soon as confirmed. forceFlush is true only on the first delta and on RETRY sentinel where latency matters; subsequent deltas batch at the BatchInterval.
Stream deltas from Activity in Go with code example
```go
func StreamCompletion(ctx context.Context, prompt string) (string, error) {
attempt := activity.GetInfo(ctx).Attempt
streamClient, err := workflowstreams.NewClientFromActivity(ctx, workflowstreams.Options{
BatchInterval: 200 * time.Millisecond,
})
if err != nil {
return "", err
}
defer streamClient.Close(ctx)
deltas := streamClient.Topic("delta")
retry := streamClient.Topic("retry")
closeTopic := streamClient.Topic("close")
if attempt > 1 {
retry.Publish(RetryEvent{Attempt: attempt}, true)
}
var full []string
first := true
for token := range generateDeltas(prompt) {
deltas.Publish(TextDelta{Text: token}, first)
first = false
full = append(full, token)
}
closeTopic.Publish(struct{}{}, false)
return strings.Join(full, ""), nil
}
```
Workflow with Workflow Streams and ack handshake code example
```go
func ChatWorkflow(ctx workflow.Context, input ChatInput) (string, error) {
stream, err := workflowstreams.NewWorkflowStream(ctx, input.StreamState)
if err != nil {
return "", err
}
subscriberDone := false
ackCh := workflow.GetSignalChannel(ctx, "subscriber-acknowledged-terminator")
workflow.Go(ctx, func(ctx workflow.Context) {
ackCh.Receive(ctx, nil)
subscriberDone = true
})
ao := workflow.ActivityOptions{StartToCloseTimeout: 5 * time.Minute}
ctx = workflow.WithActivityOptions(ctx, ao)
var result string
if err := workflow.ExecuteActivity(ctx, StreamCompletion, input.Prompt).Get(ctx, &result); err != nil {
return "", err
}
_, _ = workflow.AwaitWithTimeout(ctx, 30*time.Second, func() bool { return subscriberDone })
return result, nil
}
```
Closing a stream with in-band terminator
A subscriber's loop or listener doesn't know when the publisher is done. One pattern combines an in-band terminator—the Workflow or its Activity publishes a sentinel event the subscriber recognizes and breaks on—with a brief overlap before the Workflow returns. Sleep between the terminator and the return so any in-flight poll has time to fetch the terminator before the Workflow exits: publish the terminator, then call `Workflow.sleep(Duration.ofSeconds(30))` before returning.
Heterogeneous topics example
```java
SubscribeOptions options =
SubscribeOptions.newBuilder().setTopics("status", "progress").build();
try (WorkflowStreamSubscription subscription = stream.subscribe(options)) {
for (WorkflowStreamItem item : subscription) {
switch (item.getTopic()) {
case "status":
StatusEvent status =
DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
item.getPayload(), StatusEvent.class, StatusEvent.class);
System.out.printf("[status] %s: %s%n", status.state, status.detail);
break;
case "progress":
ProgressEvent progress =
DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
item.getPayload(), ProgressEvent.class, ProgressEvent.class);
System.out.printf("[progress] %s%n", progress.message);
break;
}
}
}
```
This example shows subscribing to multiple topics and decoding items based on topic name.
Subscribe to multiple heterogeneous topics
Every item arrives as a raw `Payload` in `item.getPayload()`, so a single subscription naturally consumes multiple topics with different payload types. Pass the topic names to `SubscribeOptions.Builder.setTopics` or leave it unset for every topic, dispatch on `item.getTopic()`, and decode into the matching type. A single subscription over multiple topics avoids the cancellation race that two concurrent subscribers would create.
Non-blocking listener example
```java
import io.temporal.common.converter.DefaultDataConverter;
import io.temporal.workflowstreams.SubscribeOptions;
import io.temporal.workflowstreams.WorkflowStreamItem;
import io.temporal.workflowstreams.WorkflowStreamListener;
import io.temporal.workflowstreams.WorkflowStreamSubscriptionHandle;
import java.util.concurrent.CompletionStage;
SubscribeOptions options = SubscribeOptions.newBuilder().setTopics("status").build();
WorkflowStreamSubscriptionHandle handle =
streamClient.subscribe(
options,
new WorkflowStreamListener() {
@Override
public CompletionStage<Void> onNext(WorkflowStreamItem item) {
StatusEvent evt =
DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
item.getPayload(), StatusEvent.class, StatusEvent.class);
System.out.printf(
"offset=%d topic=%s state=%s%n", item.getOffset(), item.getTopic(), evt.state);
return null; // or a pending stage to apply backpressure
}
@Override
public void onCompleted() {
System.out.println("stream ended");
}
});
// The calling thread is free; wait on the handle when you need to.
handle.getDoneFuture().join();
```
This example shows how to subscribe using a non-blocking listener with callbacks.
Non-blocking listener subscriber API
Unique to the Java SDK, the non-blocking listener API inverts control. Pass a `WorkflowStreamListener` to `subscribe` and items are delivered as callbacks on the poll executor. `subscribe` returns a `WorkflowStreamSubscriptionHandle` immediately. This is the right shape when one process consumes many streams concurrently—all subscriptions share the client's small executor rather than each pinning a thread. Callbacks are serialized, never invoked concurrently, and must not block. The `CompletionStage` returned by `onNext` is the backpressure boundary.
Blocking iterator example
```java
import io.temporal.client.WorkflowClient;
import io.temporal.common.converter.DefaultDataConverter;
import io.temporal.workflowstreams.SubscribeOptions;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamItem;
import io.temporal.workflowstreams.WorkflowStreamSubscription;
public void watchOrder(WorkflowClient workflowClient, String orderId) {
SubscribeOptions options = SubscribeOptions.newBuilder().setTopics("status").build();
try (WorkflowStreamClient stream = WorkflowStreamClient.newInstance(workflowClient, orderId);
WorkflowStreamSubscription subscription = stream.subscribe(options)) {
for (WorkflowStreamItem item : subscription) {
StatusEvent evt =
DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
item.getPayload(), StatusEvent.class, StatusEvent.class);
System.out.printf("[%3d%%] %s: %s%n", evt.progress, evt.state, evt.detail);
if (evt.state.equals("completed")) {
break;
}
}
}
}
```
This example shows how to subscribe using a blocking iterator and decode items from their raw payloads.
Blocking iterator subscriber API
`client.subscribe(options)` without a listener returns a `WorkflowStreamSubscription`, a blocking, single-use subscription. The consuming thread iterates with a for-each loop and blocks waiting for items while polling still runs on a shared executor. `SubscribeOptions` controls the subscription with `setTopics` to filter by name (unset means all topics), `setFromOffset` to resume from a stored global offset (zero means the beginning), and `setPollCooldown` to set the minimum interval between polls (default 100 milliseconds).
forceFlush parameter controls batch latency
Pass `true` as the `forceFlush` argument on a publish call to wake the background flusher so the current buffer ships without waiting for the next interval. The flusher only runs while the client is open. The call returns immediately after appending to the buffer and signaling the flusher; it doesn't wait for delivery to the Workflow or to subscribers. Use it for latency-sensitive events like the first delta of a response or punctuated events like RETRY and STATUS_CHANGE.
Publish from Activity example
```java
import io.temporal.activity.Activity;
import io.temporal.workflowstreams.TopicHandle;
import io.temporal.workflowstreams.WorkflowStreamClient;
public class Delta {
public String text;
}
public void streamDeltas(String orderId) {
try (WorkflowStreamClient streamClient = WorkflowStreamClient.fromActivity()) {
TopicHandle deltas = streamClient.topic("delta");
for (Delta delta : generateDeltas(orderId)) {
deltas.publish(delta);
Activity.getExecutionContext().heartbeat(null);
}
// Buffer is flushed automatically on close().
}
}
```
This example shows how to publish from an Activity using WorkflowStreamClient.fromActivity().
Publish from an Activity using WorkflowStreamClient.fromActivity()
Inside an Activity scheduled by a Workflow, call `WorkflowStreamClient.fromActivity()` to infer the Temporal Client and parent Workflow Id from the Activity context without threading them through the Activity's input. For standalone Activities with no parent Workflow context, `fromActivity()` throws an `IllegalStateException`; fall back to the general pattern with `Activity.getExecutionContext().getWorkflowClient()` and the target Workflow Id threaded through the Activity's input.
Publish from a client example
```java
import io.temporal.client.WorkflowClient;
import io.temporal.workflowstreams.TopicHandle;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamClientOptions;
import java.time.Duration;
public void publishStatus(WorkflowClient workflowClient, String workflowId) {
WorkflowStreamClientOptions options =
WorkflowStreamClientOptions.newBuilder()
.setBatchInterval(Duration.ofMillis(200))
.build();
try (WorkflowStreamClient streamClient =
WorkflowStreamClient.newInstance(workflowClient, workflowId, options)) {
TopicHandle status = streamClient.topic("status");
status.publish(new StatusEvent("started", 0, ""));
// ...
// Buffer is flushed automatically on close().
}
}
```
This example shows how to create a WorkflowStreamClient, publish events, and let the buffer flush automatically.
Java topics have no per-topic type binding
Unlike Python and TypeScript SDKs, Java topics carry no per-topic type binding. A topic handle is bound only to a name; published values are `Object` and subscribers decode each item from its raw payload. To customize per-item serialization, pass `WorkflowStreamOptions.newBuilder().setPayloadConverters(...)` to `WorkflowStream.newInstance` and use matching converters on the subscriber side.
Workflow Streams library for Java
Workflow Streams is a durable event channel for workflows, provided via the `io.temporal:temporal-workflowstreams` contrib module. It enables outside observers to follow a Workflow's progress in real time by publishing events from Workflows and Activities, and subscribing to streams using either blocking iterators or non-blocking listeners.
WorkflowStream initialization example
```java
import io.temporal.workflow.WorkflowInit;
import io.temporal.workflowstreams.WorkflowStream;
import io.temporal.workflowstreams.WorkflowStreamState;
public class OrderWorkflowImpl implements OrderWorkflow {
private final WorkflowStream stream;
@WorkflowInit
public OrderWorkflowImpl(OrderInput input) {
stream = WorkflowStream.newInstance(input.streamState);
}
@Override
public void execute(OrderInput input) {
// ... rest of the workflow
}
}
```
This example shows how to construct a WorkflowStream in a @WorkflowInit constructor. Pass `null` on a fresh start and pass the carried `WorkflowStreamState` after a Continue-As-New rollover.
Enable Workflow Streams with @WorkflowInit
Enable Workflow Streams by constructing a `WorkflowStream` once via `WorkflowStream.newInstance()` in a `@WorkflowInit` constructor. This factory registers the stream's handlers on the current Workflow, and `@WorkflowInit` constructors run before any handler dispatch, so polls and offset Queries arriving with the first Workflow Task are accepted rather than rejected. Construct exactly one `WorkflowStream` per Workflow.
Publish to Workflow Streams from a Workflow example
```java
import io.temporal.activity.ActivityOptions;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInit;
import io.temporal.workflowstreams.WorkflowStream;
import io.temporal.workflowstreams.WorkflowTopicHandle;
import java.time.Duration;
public class StatusEvent {
public String state;
public int progress;
public String detail;
}
public class OrderWorkflowImpl implements OrderWorkflow {
private final WorkflowStream stream;
private final WorkflowTopicHandle status;
@WorkflowInit
public OrderWorkflowImpl(OrderInput input) {
stream = WorkflowStream.newInstance(input.streamState);
status = stream.topic("status");
}
@Override
public void execute(OrderInput input) {
status.publish(new StatusEvent("validating", 0, "checking inventory"));
activities.validateOrder(input.orderId);
status.publish(new StatusEvent("charging", 33, "authorizing payment"));
activities.chargePayment(input.orderId);
status.publish(new StatusEvent("shipping", 66, "dispatching to warehouse"));
activities.dispatchOrder(input.orderId);
status.publish(new StatusEvent("completed", 100, ""));
}
}
```
This example shows publishing status events through a topic handle at different points in the workflow execution.
Publish only from Activities in a stream pattern
When an Activity can retry, the consumer side has to account for it. A retried attempt is a fresh publisher, so its output appears in the stream alongside the output from the previous attempt. To keep this clean, the Activity should be the publisher because it owns the non-deterministic work. The Workflow processes only the Activity's return value, never reading its own stream. This keeps Workflow state independent of streamed output, letting retried Activity attempts surface to subscribers without polluting the Workflow's durable state.
LLM consumer with retry handling
```java
import io.temporal.client.WorkflowClient;
import io.temporal.common.converter.DefaultDataConverter;
import io.temporal.workflowstreams.SubscribeOptions;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamItem;
import io.temporal.workflowstreams.WorkflowStreamSubscription;
public String streamChat(WorkflowClient workflowClient, String chatId) {
StringBuilder output = new StringBuilder();
Runnable render = () -> {
// ... display the accumulated output
};
SubscribeOptions options =
SubscribeOptions.newBuilder().setTopics("delta", "retry", "close").build();
try (WorkflowStreamClient stream = WorkflowStreamClient.newInstance(workflowClient, chatId);
WorkflowStreamSubscription subscription = stream.subscribe(options)) {
for (WorkflowStreamItem item : subscription) {
switch (item.getTopic()) {
case "retry":
output.setLength(0);
render.run();
break;
case "delta":
TextDelta delta =
DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
item.getPayload(), TextDelta.class, TextDelta.class);
output.append(delta.text);
render.run();
break;
case "close":
workflowClient
.newUntypedWorkflowStub(chatId)
.signal("subscriberAcknowledgedTerminator");
return output.toString();
}
}
}
return output.toString();
}
```
This example shows a consumer that resets accumulated output on RETRY and signals the Workflow on close.
LLM workflow with acknowledgment handshake
```java
import io.temporal.activity.ActivityOptions;
import io.temporal.workflow.SignalMethod;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInit;
import io.temporal.workflow.WorkflowInterface;
import io.temporal.workflow.WorkflowMethod;
import io.temporal.workflowstreams.WorkflowStream;
import java.time.Duration;
@WorkflowInterface
public interface ChatWorkflow {
@WorkflowMethod
String complete(ChatInput input);
@SignalMethod
void subscriberAcknowledgedTerminator();
}
public class ChatWorkflowImpl implements ChatWorkflow {
private final LlmActivities activities =
Workflow.newActivityStub(
LlmActivities.class,
ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofMinutes(5)).build());
private boolean subscriberDone = false;
@WorkflowInit
public ChatWorkflowImpl(ChatInput input) {
WorkflowStream.newInstance(input.streamState);
}
@Override
public void subscriberAcknowledgedTerminator() {
subscriberDone = true;
}
@Override
public String complete(ChatInput input) {
String result = activities.streamCompletion(input.prompt);
Workflow.await(Duration.ofSeconds(30), () -> subscriberDone);
return result;
}
}
```
This example shows a Workflow that waits for the subscriber to acknowledge the terminal close event before returning.
LLM streaming example with retry handling
```java
import com.openai.client.OpenAIClient;
import io.temporal.activity.Activity;
import io.temporal.workflowstreams.TopicHandle;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamClientOptions;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
public class TextDelta {
public String text;
public TextDelta() {}
public TextDelta(String text) { this.text = text; }
}
public class RetryEvent {
public int attempt;
public RetryEvent() {}
public RetryEvent(int attempt) { this.attempt = attempt; }
}
public class CloseEvent {}
public class LlmActivitiesImpl implements LlmActivities {
@Override
public String streamCompletion(String prompt) {
WorkflowStreamClientOptions options =
WorkflowStreamClientOptions.newBuilder()
.setBatchInterval(Duration.ofMillis(200))
.build();
try (WorkflowStreamClient streamClient = WorkflowStreamClient.fromActivity(options)) {
TopicHandle deltas = streamClient.topic("delta");
TopicHandle retry = streamClient.topic("retry");
TopicHandle close = streamClient.topic("close");
int attempt = Activity.getExecutionContext().getInfo().getAttempt();
if (attempt > 1) {
retry.publish(new RetryEvent(attempt), /* forceFlush */ true);
}
OpenAIClient openai = OpenAIOkHttpClient.builder().fromEnv().maxRetries(0).build();
ChatCompletionCreateParams params =
ChatCompletionCreateParams.builder().model("gpt-4o-mini").addUserMessage(prompt).build();
StringBuilder full = new StringBuilder();
AtomicBoolean first = new AtomicBoolean(true);
try (StreamResponse<ChatCompletionChunk> stream =
openai.chat().completions().createStreaming(params)) {
stream.stream()
.forEach(
chunk ->
chunk.choices().stream()
.findFirst()
.flatMap(choice -> choice.delta().content())
.filter(text -> !text.isEmpty())
.ifPresent(
text -> {
deltas.publish(new TextDelta(text), first.getAndSet(false));
full.append(text);
}));
}
close.publish(new CloseEvent());
return full.toString();
}
}
}
```
This example shows an LLM Activity that publishes text deltas as they arrive, emits a RETRY event on retries, and closes the stream when done. forceFlush is true only on the first delta.
Cross-language Workflow Streams interoperability
The handler names, JSON envelope field names, and per-item payload encoding match the other SDKs' packages exactly, so a Java publisher or subscriber interoperates with a Workflow written in any of them and vice versa. One Java-specific caveat: the protocol envelope types are serialized by the Workflow's and client's configured data converter. The default Jackson JSON converter produces the wire-compatible snake_case field names; if you configure a non-Jackson JSON converter, it must produce the same field names for cross-language interop.
Codec chain runs once on envelope
Payload codecs (encryption, compression) configured on the Temporal client run on the Signal or Update envelope that carries each batch, never per item, so items are never double-encoded. `setPayloadConverters` handles only per-item serialization; its `PayloadConverter[]` type makes it impossible to slot a codec in per item. This ensures that encryption and compression are applied exactly once each direction.
Payload decoders are at call site
`item.getPayload()` is always the raw payload. Decode it with a converter built from the same payload converters used by the publisher. When publishers and subscribers both rely on the defaults, `DefaultDataConverter.STANDARD_INSTANCE` matches on both sides. If you pass `setPayloadConverters` on the Workflow side or the client side, build a matching converter on the subscriber side.
Poll executor sizing for slow Workflows
The default executor for polling has 2 daemon threads, is created lazily, and is owned (and shut down) by the client; a user-supplied executor is never shut down by the client. The executor runs the short update-admission and delivery steps and poll cooldowns, never the long poll itself, so a small pool serves many subscriptions. The known worst case for pool pressure is a backlogged Workflow pinning a thread in the update-admission call; supply a bigger pool via `setPollExecutor` when running many subscriptions against slow Workflows.
Listener callbacks must not block
Non-blocking listener callbacks run serialized on the client's poll executor, which drives every subscription on the client. Blocking in `onNext` stalls the client's other subscriptions. Hand slow work to your own executor and return the resulting `CompletionStage` for backpressure. For example, `CompletableFuture.runAsync(() -> render(item), renderExecutor)` returns a stage that completes when the work is done, and the next item is delivered only after it completes.
Publisher TTL and deduplication window
At each Continue-As-New, deduplication entries whose last-seen time is older than the Publisher TTL are dropped. The last-seen time is updated on each successful publish, not on each retry attempt. A publisher that returns after a longer pause may produce a duplicate. `stream.continueAsNew(...)` snapshots with a 15-minute default. To tune it, use the explicit recipe by calling `stream.detachPollers()`, `Workflow.await(() -> Workflow.isEveryHandlerFinished())`, then `stream.getState(publisherTtl)` with a custom duration.
Continue-As-New with Workflow Streams example
```java
public class WorkflowInput {
public int itemsProcessed;
public WorkflowStreamState streamState;
}
public class LongRunningWorkflowImpl implements LongRunningWorkflow {
private final WorkflowStream stream;
private int itemsProcessed;
@WorkflowInit
public LongRunningWorkflowImpl(WorkflowInput input) {
stream = WorkflowStream.newInstance(input.streamState);
itemsProcessed = input.itemsProcessed;
}
@Override
public void execute(WorkflowInput input) {
while (true) {
doOneIteration();
itemsProcessed++;
if (Workflow.getInfo().isContinueAsNewSuggested()) {
stream.continueAsNew(
state -> {
WorkflowInput next = new WorkflowInput();
next.itemsProcessed = itemsProcessed;
next.streamState = state;
return new Object[] {next};
});
}
}
}
}
```
This example shows how to carry both application state and stream state across a Continue-As-New boundary.
Closing a stream with acknowledgment handshake
The subscriber sends a Signal once it has the terminator; the Workflow waits up to a timeout, returning as soon as the ack arrives. Define a Signal method in the Workflow interface, set a flag in that method, and call `Workflow.await(Duration.ofSeconds(30), () -> subscriberDone)` to wait for the acknowledgment. This pattern provides an overlap so in-flight polls can deliver the terminator before the Workflow exits.
Event History displays workflow, activity, and timer summaries
In the Temporal Web UI Event History, individual events display their associated summaries when available. Workflow, Activity, and Timer summaries appear in purple text next to their corresponding events, providing immediate context without expanding the Event details. The summary is also prominently displayed when expanding the detailed view of an Event.
WorkflowStream basic setup example
from dataclasses import dataclass
from temporalio import workflow
from temporalio.contrib.workflow_streams import WorkflowStream
@dataclass
class OrderInput:
order_id: str
@workflow.defn
class OrderWorkflow:
@workflow.init
def __init__(self, input: OrderInput) -> None:
self.stream = WorkflowStream()
Import WorkflowStream from temporalio.contrib.workflow_streams
Workflow Streams functionality is available by importing WorkflowStream from the temporalio.contrib.workflow_streams module.
Enable WorkflowStream in @workflow.init method
WorkflowStream must be constructed from the @workflow.init method, not @workflow.run. Constructing it from @workflow.run raises a RuntimeError and would miss any publishes that arrived before the run body started executing. The stream's handlers have to be registered before the first publish Signal arrives.
Only one WorkflowStream per Workflow
Constructing more than one WorkflowStream on the same Workflow raises a RuntimeError.
Bind topic name to event type via stream.topic()
Bind a topic name to its event type once using self.stream.topic("name", type=Type). This returns a handle that records the per-stream binding from topic name to value type so call sites don't have to repeat the type on every publish and so subscribers reading the same handle decode to the matching type. The type= argument is optional and defaults to Any.
Publish events to workflow stream
Call publish() on a topic handle returned from stream.topic() to append events to the stream. The payload converter encodes each value, and the codec chain (encryption, compression, etc.) runs once on the Signal or Update envelope that carries the batch, not per item.
Workflow stream publish example
from dataclasses import dataclass
@dataclass
class StatusEvent:
state: str
progress: int = 0
detail: str = ""
@workflow.defn
class OrderWorkflow:
@workflow.init
def __init__(self, input: OrderInput) -> None:
self.stream = WorkflowStream()
self.status = self.stream.topic("status", type=StatusEvent)
@workflow.run
async def run(self, input: OrderInput) -> None:
self.status.publish(StatusEvent(state="validating", detail="checking inventory"))
await validate_order(input.order_id)
self.status.publish(StatusEvent(state="charging", progress=33, detail="authorizing payment"))
await charge_payment(input.order_id)
self.status.publish(StatusEvent(state="shipping", progress=66, detail="dispatching to warehouse"))
await dispatch_order(input.order_id)
self.status.publish(StatusEvent(state="completed", progress=100))
WorkflowStreamClient publish example
from datetime import timedelta
from temporalio.client import Client
from temporalio.contrib.workflow_streams import WorkflowStreamClient
async def publish_status(workflow_id: str) -> None:
temporal_client = await Client.connect("localhost:7233")
stream_client = WorkflowStreamClient.create(
temporal_client,
workflow_id=workflow_id,
batch_interval=timedelta(milliseconds=200),
)
async with stream_client:
status = stream_client.topic("status", type=StatusEvent)
status.publish(StatusEvent(state="started"))
...
# Buffer is flushed on context manager exit.
WorkflowStreamClient.from_within_activity() inside Activities
Inside an Activity scheduled by a Workflow, use WorkflowStreamClient.from_within_activity() to infer the Temporal Client and parent Workflow Id from the Activity context. This avoids threading them through the Activity's input. For standalone Activities started directly via Client.start_activity, there is no parent Workflow context to infer, so from_within_activity() raises an exception. Fall back to the general pattern with activity.client() and the target Workflow Id threaded through the Activity's input.
Activity stream publishing example
from temporalio import activity
from temporalio.contrib.workflow_streams import WorkflowStreamClient
@activity.defn
async def stream_deltas(order_id: str) -> None:
client = WorkflowStreamClient.from_within_activity()
async with client:
deltas = client.topic("delta", type=Delta)
for delta in generate_deltas(order_id):
deltas.publish(delta)
activity.heartbeat()
# Buffer is flushed on context manager exit.
force_flush parameter for latency-sensitive publishes
Pass force_flush=True on a publish to wake the background flusher so the current buffer ships without waiting for the next interval. Use it for latency-sensitive events like the first delta of a response so the user sees something fast, or punctuated events like RETRY and STATUS_CHANGE. The call returns immediately after appending to the buffer and signaling the flusher; it doesn't wait for delivery to the Workflow or subscribers. If the Workflow Stream client is not entered (not inside async with), force_flush=True queues the wake event but nothing ships until the context is entered or await client.flush() is called.
Subscribe to heterogeneous topics with RawValue
To consume multiple topics with different payload types, call client.subscribe() directly with a list of names (or subscribe([]) for every topic) and pass result_type=temporalio.common.RawValue so each item arrives as the underlying Payload wrapped in a RawValue. Dispatch on item.topic and decode the wrapped payload with the client's payload converter. A single iterator over multiple topics also avoids the cancellation race that two concurrent subscribers would create.
Heterogeneous topics subscription example
from temporalio.common import RawValue
converter = temporal_client.data_converter.payload_converter
async for item in stream.subscribe(["status", "progress"], result_type=RawValue):
if item.topic == "status":
evt = converter.from_payload(item.data.payload, StatusEvent)
print(f"[status] {evt.state}: {evt.detail}")
elif item.topic == "progress":
evt = converter.from_payload(item.data.payload, ProgressEvent)
print(f"[progress] {evt.message}")
Default result_type decoding behavior
Omitting result_type entirely or passing result_type=None decodes each item with the converter's default rules. For the stock JSON converter, that means a Python primitive, dict, or list. This works for fully homogeneous streams but not for the dispatch-by-topic pattern where each topic has its own concrete dataclass.
Close stream with in-band terminator
A subscriber's async for loop does not know when the publisher is done. A common pattern uses an in-band terminator: the Workflow or its Activity publishes a sentinel event the subscriber recognizes and breaks on. Each subscription decides what its own end-of-stream marker is.
Fixed sleep for stream closing overlap
Sleep between the terminator and the return so any in-flight poll has time to fetch the terminator before the Workflow exits. For example, at the end of @workflow.run: self.status.publish(StatusEvent(state="completed", progress=100)) followed by await workflow.sleep(timedelta(seconds=30)) before returning.
Acknowledgment handshake for stream closing
The subscriber sends a Signal once it has the terminator; the Workflow waits up to a timeout, returning as soon as the ack arrives. This pattern is more responsive than fixed sleep. The Workflow waits via workflow.wait_condition with a timeout as a fallback for when no subscriber is attached.
Acknowledgment handshake example
@workflow.signal
async def subscriber_acknowledged_terminator(self) -> None:
self.subscriber_done = True
@workflow.run
async def run(self, input: ChatInput) -> str:
...
try:
await workflow.wait_condition(
lambda: self.subscriber_done,
timeout=timedelta(seconds=30),
)
except TimeoutError:
pass # No subscriber attached; the run still completes cleanly.
return result