Skip to content

Commit 08eae4d

Browse files
brianstrauchclaude
brianstrauch
and
claude
authored
Sample code for Workflow Streams (#783)
* Sample code for Workflow Streams Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Use SDK 1.37.0-SNAPSHOT from the Central snapshots repository SDK main now publishes 1.37.0-SNAPSHOT, and snapshots moved from oss.sonatype.org to central.sonatype.com when OSSRH was decommissioned. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Add non-blocking listener scenario to Workflow Streams sample Scenario 2 subscribes with a WorkflowStreamListener instead of iterating: two order workflows are consumed concurrently on one shared executor thread, with CompletionStage-based backpressure and done futures instead of a parked thread per subscription. Later scenarios renumbered, and the stale SDK snapshot note updated for the Central snapshots repository. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Use released SDK 1.37.0 instead of snapshot The temporal-workflowstreams module is now in the tagged 1.37.0 release on Maven Central, so drop the -SNAPSHOT pin and the snapshots repository. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 7b3efa6 commit 08eae4d

29 files changed

Lines changed: 1498 additions & 5 deletions

‎.github/CODEOWNERS‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,4 +7,6 @@
77
# For each one, we add the owning team, as well as
88
# @temporalio/sdk, so the SDK team can continue to
99
# manage repo-wide concerns
10-
/springai/ @temporalio/ai-sdk @temporalio/sdk
10+
/springai/ @temporalio/sdk @temporalio/ai-sdk
11+
/core/src/main/java/io/temporal/samples/workflowstreams/ @temporalio/sdk @temporalio/ai-sdk
12+
/core/src/test/java/io/temporal/samples/workflowstreams/ @temporalio/sdk @temporalio/ai-sdk

‎README.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,8 @@ Load client configuration from TOML files with programmatic overrides.
117117

118118
#### API demonstrations
119119

120+
- [**Workflow Streams**](/core/src/main/java/io/temporal/samples/workflowstreams): Demonstrates a durable publish/subscribe log hosted inside a workflow, using the experimental `temporal-workflowstreams` module.
121+
120122
- [**Async Untyped Child Workflow**](/core/src/main/java/io/temporal/samples/asyncuntypedchild): Demonstrates how to invoke an untyped child workflow async, that can complete after parent workflow is already completed.
121123

122124
- [**Updatable Timer**](/core/src/main/java/io/temporal/samples/updatabletimer): Demonstrates the use of a helper class which relies on `Workflow.await` to implement a blocking sleep that can be updated at any moment.

‎build.gradle‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,15 +21,12 @@ subprojects {
2121
ext {
2222
otelVersion = '1.30.1'
2323
otelVersionAlpha = "${otelVersion}-alpha"
24-
javaSDKVersion = '1.36.1'
24+
javaSDKVersion = '1.37.0'
2525
camelVersion = '3.22.1'
2626
jarVersion = '1.0.0'
2727
}
2828

2929
repositories {
30-
maven {
31-
url "https://oss.sonatype.org/content/repositories/snapshots/"
32-
}
3330
mavenCentral()
3431
}
3532

‎core/build.gradle‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ dependencies {
22
// Temporal SDK
33
implementation "io.temporal:temporal-sdk:$javaSDKVersion"
44
implementation "io.temporal:temporal-opentracing:$javaSDKVersion"
5+
implementation "io.temporal:temporal-workflowstreams:$javaSDKVersion"
56
testImplementation("io.temporal:temporal-testing:$javaSDKVersion")
67

78
// Environment configuration
@@ -27,6 +28,7 @@ dependencies {
2728
implementation 'io.jaegertracing:jaeger-client:1.8.1'
2829

2930
// Used in samples
31+
implementation "com.openai:openai-java:4.39.1"
3032
implementation group: 'commons-configuration', name: 'commons-configuration', version: '1.10'
3133
implementation group: 'io.cloudevents', name: 'cloudevents-core', version: '4.0.1'
3234
implementation group: 'io.cloudevents', name: 'cloudevents-api', version: '4.0.1'

‎core/src/main/java/io/temporal/samples/workflowstreams/ExternalPublisher.java‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
package io.temporal.samples.workflowstreams;
2+
3+
import io.temporal.client.WorkflowClient;
4+
import io.temporal.client.WorkflowOptions;
5+
import io.temporal.samples.workflowstreams.Shared.HubInput;
6+
import io.temporal.samples.workflowstreams.Shared.NewsEvent;
7+
import io.temporal.workflowstreams.TopicHandle;
8+
import io.temporal.workflowstreams.WorkflowStreamClient;
9+
import io.temporal.workflowstreams.WorkflowStreamItem;
10+
import io.temporal.workflowstreams.WorkflowStreamSubscription;
11+
import java.util.UUID;
12+
13+
/**
14+
* Scenario 4: external (non-activity) publisher. The hub workflow does no work of its own; it just
15+
* hosts the stream. A separate process publishes news into it using the same client factory used to
16+
* subscribe, then signals the workflow to close. Here the publisher and a subscriber run as two
17+
* threads.
18+
*/
19+
public class ExternalPublisher {
20+
21+
private static final String[] HEADLINES = {
22+
"markets open higher", "new bridge opens downtown", "local team wins championship",
23+
};
24+
25+
/** The sentinel the publisher sends last so the subscriber knows to stop. */
26+
private static final String DONE_HEADLINE = "-- end of feed --";
27+
28+
public static void main(String[] args) throws InterruptedException {
29+
WorkflowClient client = Shared.newWorkflowClient();
30+
31+
String workflowId = "workflow-streams-hub-" + UUID.randomUUID();
32+
HubWorkflow workflow =
33+
client.newWorkflowStub(
34+
HubWorkflow.class,
35+
WorkflowOptions.newBuilder()
36+
.setWorkflowId(workflowId)
37+
.setTaskQueue(Shared.TASK_QUEUE)
38+
.build());
39+
WorkflowClient.start(workflow::host, new HubInput("newsroom"));
40+
System.out.println("Started workflow: " + workflowId);
41+
42+
Thread subscriber =
43+
new Thread(
44+
() -> {
45+
try (WorkflowStreamClient stream =
46+
WorkflowStreamClient.newInstance(client, workflowId);
47+
WorkflowStreamSubscription subscription =
48+
stream.topic(Shared.TOPIC_NEWS).subscribe(0)) {
49+
for (WorkflowStreamItem item : subscription) {
50+
NewsEvent evt = Shared.decode(item, NewsEvent.class);
51+
if (evt.headline.equals(DONE_HEADLINE)) {
52+
return;
53+
}
54+
System.out.printf("[subscriber] %s%n", evt.headline);
55+
}
56+
}
57+
});
58+
59+
Thread publisher =
60+
new Thread(
61+
() -> {
62+
try (WorkflowStreamClient producer =
63+
WorkflowStreamClient.newInstance(client, workflowId)) {
64+
TopicHandle news = producer.topic(Shared.TOPIC_NEWS);
65+
for (String headline : HEADLINES) {
66+
news.publish(new NewsEvent(headline));
67+
System.out.printf("[publisher] sent: %s%n", headline);
68+
try {
69+
Thread.sleep(500);
70+
} catch (InterruptedException e) {
71+
Thread.currentThread().interrupt();
72+
return;
73+
}
74+
}
75+
// Force-flush the sentinel and wait for the server to confirm delivery
76+
// before signaling the workflow to close.
77+
news.publish(new NewsEvent(DONE_HEADLINE), /* forceFlush */ true);
78+
producer.flush();
79+
}
80+
workflow.close();
81+
System.out.println("[publisher] signaled close");
82+
});
83+
84+
subscriber.start();
85+
publisher.start();
86+
subscriber.join();
87+
publisher.join();
88+
System.exit(0);
89+
}
90+
}

‎core/src/main/java/io/temporal/samples/workflowstreams/HubWorkflow.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
package io.temporal.samples.workflowstreams;
2+
3+
import io.temporal.samples.workflowstreams.Shared.HubInput;
4+
import io.temporal.workflow.SignalMethod;
5+
import io.temporal.workflow.WorkflowInterface;
6+
import io.temporal.workflow.WorkflowMethod;
7+
8+
/**
9+
* Scenario 4: does no work of its own; it exists only to host the stream for an external publisher
10+
* and shuts down on a close signal.
11+
*/
12+
@WorkflowInterface
13+
public interface HubWorkflow {
14+
@WorkflowMethod
15+
String host(HubInput input);
16+
17+
@SignalMethod
18+
void close();
19+
}

‎core/src/main/java/io/temporal/samples/workflowstreams/HubWorkflowImpl.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
package io.temporal.samples.workflowstreams;
2+
3+
import io.temporal.samples.workflowstreams.Shared.HubInput;
4+
import io.temporal.workflow.Workflow;
5+
import io.temporal.workflow.WorkflowInit;
6+
import io.temporal.workflowstreams.WorkflowStream;
7+
8+
public class HubWorkflowImpl implements HubWorkflow {
9+
10+
private boolean closed;
11+
12+
@WorkflowInit
13+
public HubWorkflowImpl(HubInput input) {
14+
WorkflowStream.newInstance(input.streamState);
15+
}
16+
17+
@Override
18+
public String host(HubInput input) {
19+
Workflow.await(() -> closed);
20+
21+
// The publisher publishes its own terminator into the stream before signaling close
22+
// (see ExternalPublisher). Hold the run open briefly so subscribers' final poll
23+
// delivers any items still in the log.
24+
Workflow.sleep(OrderWorkflowImpl.DRAIN_DELAY);
25+
return "hub " + input.hubId + " closed";
26+
}
27+
28+
@Override
29+
public void close() {
30+
closed = true;
31+
}
32+
}

‎core/src/main/java/io/temporal/samples/workflowstreams/Llm.java‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
package io.temporal.samples.workflowstreams;
2+
3+
import io.temporal.client.WorkflowClient;
4+
import io.temporal.client.WorkflowOptions;
5+
import io.temporal.samples.workflowstreams.Shared.LlmInput;
6+
import io.temporal.samples.workflowstreams.Shared.RetryEvent;
7+
import io.temporal.samples.workflowstreams.Shared.TextDelta;
8+
import io.temporal.workflowstreams.SubscribeOptions;
9+
import io.temporal.workflowstreams.WorkflowStreamClient;
10+
import io.temporal.workflowstreams.WorkflowStreamItem;
11+
import io.temporal.workflowstreams.WorkflowStreamSubscription;
12+
import java.util.UUID;
13+
14+
/**
15+
* Scenario 6: LLM token streaming. The workflow hosts the stream while an activity makes the
16+
* streaming OpenAI call and republishes each token delta. On a retry the activity emits a
17+
* RetryEvent and this subscriber rewinds the terminal and re-renders. Run {@link LlmWorker} with
18+
* {@code OPENAI_API_KEY} set before running this.
19+
*/
20+
public class Llm {
21+
22+
/**
23+
* ANSI escapes to save the cursor position and to restore it while clearing everything below, so
24+
* a retry can re-render the completion from scratch.
25+
*/
26+
private static final String ANSI_SAVE = "\u001b[s";
27+
28+
private static final String ANSI_RESTORE_AND_CLEAR = "\u001b[u\u001b[J";
29+
30+
public static void main(String[] args) {
31+
String prompt = args.length > 0 ? args[0] : "In one short paragraph, explain what Temporal is.";
32+
33+
WorkflowClient client = Shared.newWorkflowClient();
34+
35+
String workflowId = "workflow-streams-llm-" + UUID.randomUUID();
36+
LlmWorkflow workflow =
37+
client.newWorkflowStub(
38+
LlmWorkflow.class,
39+
WorkflowOptions.newBuilder()
40+
.setWorkflowId(workflowId)
41+
.setTaskQueue(Shared.LLM_TASK_QUEUE)
42+
.build());
43+
WorkflowClient.start(workflow::complete, new LlmInput(prompt, null));
44+
System.out.println("Started workflow: " + workflowId);
45+
46+
try (WorkflowStreamClient stream = WorkflowStreamClient.newInstance(client, workflowId);
47+
WorkflowStreamSubscription subscription =
48+
stream.subscribe(
49+
SubscribeOptions.newBuilder()
50+
.setTopics(Shared.TOPIC_DELTA, Shared.TOPIC_RETRY, Shared.TOPIC_COMPLETE)
51+
.build())) {
52+
System.out.print(ANSI_SAVE);
53+
for (WorkflowStreamItem item : subscription) {
54+
if (item.getTopic().equals(Shared.TOPIC_RETRY)) {
55+
RetryEvent evt = Shared.decode(item, RetryEvent.class);
56+
System.out.print(ANSI_RESTORE_AND_CLEAR);
57+
System.out.printf("[retry attempt %d] resetting output%n%n", evt.attempt);
58+
System.out.print(ANSI_SAVE);
59+
} else if (item.getTopic().equals(Shared.TOPIC_DELTA)) {
60+
TextDelta evt = Shared.decode(item, TextDelta.class);
61+
System.out.print(evt.text);
62+
System.out.flush();
63+
} else if (item.getTopic().equals(Shared.TOPIC_COMPLETE)) {
64+
System.out.println();
65+
break;
66+
}
67+
}
68+
}
69+
System.exit(0);
70+
}
71+
}

‎core/src/main/java/io/temporal/samples/workflowstreams/LlmActivities.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package io.temporal.samples.workflowstreams;
2+
3+
import io.temporal.activity.ActivityInterface;
4+
import io.temporal.activity.ActivityMethod;
5+
import io.temporal.samples.workflowstreams.Shared.LlmInput;
6+
7+
@ActivityInterface
8+
public interface LlmActivities {
9+
/**
10+
* Streams an LLM completion to the parent workflow's stream and returns the accumulated full
11+
* text.
12+
*/
13+
@ActivityMethod
14+
String streamCompletion(LlmInput input);
15+
}

‎core/src/main/java/io/temporal/samples/workflowstreams/LlmActivitiesImpl.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package io.temporal.samples.workflowstreams;
2+
3+
import com.openai.client.OpenAIClient;
4+
import com.openai.client.okhttp.OpenAIOkHttpClient;
5+
import com.openai.core.http.StreamResponse;
6+
import com.openai.models.chat.completions.ChatCompletionChunk;
7+
import com.openai.models.chat.completions.ChatCompletionCreateParams;
8+
import io.temporal.activity.Activity;
9+
import io.temporal.samples.workflowstreams.Shared.LlmInput;
10+
import io.temporal.samples.workflowstreams.Shared.RetryEvent;
11+
import io.temporal.samples.workflowstreams.Shared.TextComplete;
12+
import io.temporal.samples.workflowstreams.Shared.TextDelta;
13+
import io.temporal.workflowstreams.TopicHandle;
14+
import io.temporal.workflowstreams.WorkflowStreamClient;
15+
import io.temporal.workflowstreams.WorkflowStreamClientOptions;
16+
import java.time.Duration;
17+
18+
/**
19+
* Calls OpenAI with streaming enabled and republishes each token delta to the workflow stream. The
20+
* accumulated text is published on the complete topic and returned as the activity result. Because
21+
* the activity owns the non-deterministic OpenAI call, the workflow stays deterministic.
22+
*
23+
* <p>Retries are disabled on the OpenAI client so transient failures surface as Temporal activity
24+
* retries instead. On a retry (attempt &gt; 1) it publishes a RetryEvent so subscribers can reset
25+
* partially rendered output.
26+
*/
27+
public class LlmActivitiesImpl implements LlmActivities {
28+
29+
static final String DEFAULT_MODEL = "gpt-4o-mini";
30+
31+
@Override
32+
public String streamCompletion(LlmInput input) {
33+
WorkflowStreamClientOptions options =
34+
WorkflowStreamClientOptions.newBuilder().setBatchInterval(Duration.ofMillis(200)).build();
35+
try (WorkflowStreamClient streamClient = WorkflowStreamClient.fromActivity(options)) {
36+
TopicHandle deltas = streamClient.topic(Shared.TOPIC_DELTA);
37+
TopicHandle complete = streamClient.topic(Shared.TOPIC_COMPLETE);
38+
TopicHandle retry = streamClient.topic(Shared.TOPIC_RETRY);
39+
40+
int attempt = Activity.getExecutionContext().getInfo().getAttempt();
41+
if (attempt > 1) {
42+
retry.publish(new RetryEvent(attempt), /* forceFlush */ true);
43+
}
44+
45+
String model = input.model != null && !input.model.isEmpty() ? input.model : DEFAULT_MODEL;
46+
47+
// Reads OPENAI_API_KEY from the environment.
48+
OpenAIClient openai = OpenAIOkHttpClient.builder().fromEnv().maxRetries(0).build();
49+
ChatCompletionCreateParams params =
50+
ChatCompletionCreateParams.builder().model(model).addUserMessage(input.prompt).build();
51+
52+
StringBuilder full = new StringBuilder();
53+
try (StreamResponse<ChatCompletionChunk> stream =
54+
openai.chat().completions().createStreaming(params)) {
55+
stream.stream()
56+
.forEach(
57+
chunk ->
58+
chunk.choices().stream()
59+
.findFirst()
60+
.flatMap(choice -> choice.delta().content())
61+
.filter(text -> !text.isEmpty())
62+
.ifPresent(
63+
text -> {
64+
deltas.publish(new TextDelta(text));
65+
full.append(text);
66+
}));
67+
}
68+
69+
String fullText = full.toString();
70+
complete.publish(new TextComplete(fullText), /* forceFlush */ true);
71+
return fullText;
72+
}
73+
}
74+
}

0 commit comments

Comments
 (0)