|
| 1 | +# Workflow Streams |
| 2 | + |
| 3 | +A durable publish/subscribe log hosted inside a Temporal Workflow. |
| 4 | + |
| 5 | +External code (activities, starters, other processes) publishes messages to |
| 6 | +named topics via **signals**; subscribers long-poll for new items via |
| 7 | +**updates**; a **query** exposes the current offset. The stream is backed by |
| 8 | +Temporal's durable execution, giving ordered, durable, exactly-once delivery |
| 9 | +with client-side batching, publisher dedup, continue-as-new survival, |
| 10 | +truncation, and ~1 MB response paging. |
| 11 | + |
| 12 | +It is well suited to durable event streams whose cost scales with durable |
| 13 | +batches rather than message count. Each poll round-trip costs ~100 ms of |
| 14 | +latency, so it is not intended for ultra-low-latency streaming. |
| 15 | + |
| 16 | +All APIs in this module are experimental and may change. |
| 17 | + |
| 18 | +## Workflow side |
| 19 | + |
| 20 | +Construct a `WorkflowStream` once in a `@WorkflowInit` constructor. The factory |
| 21 | +registers the publish signal, poll update, and offset query handlers, and a |
| 22 | +`@WorkflowInit` constructor runs before any handler dispatch, so polls and |
| 23 | +offset queries arriving with the first workflow task (e.g. from |
| 24 | +update-with-start) are accepted rather than rejected. |
| 25 | + |
| 26 | +```java |
| 27 | +public class MyInput { |
| 28 | + public int itemsProcessed; // your own workflow state |
| 29 | + public WorkflowStreamState streamState; |
| 30 | +} |
| 31 | + |
| 32 | +public class MyWorkflowImpl implements MyWorkflow { |
| 33 | + private final WorkflowStream stream; |
| 34 | + |
| 35 | + @WorkflowInit |
| 36 | + public MyWorkflowImpl(MyInput input) { |
| 37 | + stream = WorkflowStream.newInstance(input.streamState); |
| 38 | + } |
| 39 | + |
| 40 | + @Override |
| 41 | + public void execute(MyInput input) { |
| 42 | + // Optionally publish from workflow code: |
| 43 | + stream.topic("events").publish("hello from the workflow"); |
| 44 | + |
| 45 | + // Run your workflow; the stream serves external publishers and subscribers |
| 46 | + // for as long as the workflow is running. Block until your workflow's exit |
| 47 | + // condition is met (here, a `done` flag set elsewhere, e.g. by a signal). |
| 48 | + Workflow.await(() -> done); |
| 49 | + } |
| 50 | +} |
| 51 | +``` |
| 52 | + |
| 53 | +Constructing the stream at the top of the workflow method also works — signals |
| 54 | +received earlier are buffered by the SDK — but polls and offset queries are |
| 55 | +rejected until the stream exists, so prefer `@WorkflowInit`. |
| 56 | + |
| 57 | +For workflows that use continue-as-new, the stream's log and offsets must be |
| 58 | +carried across each boundary, since continue-as-new starts a fresh run with an |
| 59 | +empty history. This is a round-trip with two halves: |
| 60 | + |
| 61 | +- **Capture** the state when rolling over. Instead of calling |
| 62 | + `Workflow.continueAsNew` directly, call `stream.continueAsNew`. It drains |
| 63 | + pollers, waits for in-flight handlers, snapshots the current stream state, and |
| 64 | + hands it to your callback, which builds the argument list for the next run. |
| 65 | + The callback is where you assemble the full input — carry forward your own |
| 66 | + workflow state alongside the captured `state`: |
| 67 | + |
| 68 | + ```java |
| 69 | + stream.continueAsNew(state -> { |
| 70 | + MyInput next = new MyInput(); |
| 71 | + next.itemsProcessed = itemsProcessed; // your own state, carried across the boundary |
| 72 | + next.streamState = state; // the captured stream state |
| 73 | + return new Object[] {next}; |
| 74 | + }); |
| 75 | + ``` |
| 76 | + |
| 77 | +- **Restore** it on the next run. That `MyInput` arrives as the next run's |
| 78 | + input, and its `streamState` field is the value already passed to |
| 79 | + `WorkflowStream.newInstance` in the example above. It is `null` on a fresh |
| 80 | + start and non-null after a roll-over, so the stream rehydrates the log |
| 81 | + automatically. |
| 82 | + |
| 83 | +The `WorkflowStreamState` field is what gives the captured stream state |
| 84 | +somewhere to live between runs; the other fields on `MyInput` are your own and |
| 85 | +are threaded through the same way. |
| 86 | + |
| 87 | +## Publishing (client side) |
| 88 | + |
| 89 | +From an activity, use `fromActivity` to target the parent workflow: |
| 90 | + |
| 91 | +```java |
| 92 | +public void publishActivity() { |
| 93 | + try (WorkflowStreamClient client = WorkflowStreamClient.fromActivity()) { |
| 94 | + TopicHandle topic = client.topic("events"); |
| 95 | + for (int i = 0; i < 100; i++) { |
| 96 | + topic.publish("item " + i); |
| 97 | + } |
| 98 | + } // client.close() is called on completion, which flushes the remaining buffer |
| 99 | +} |
| 100 | +``` |
| 101 | + |
| 102 | +From a starter or any code with a `WorkflowClient`, use `newInstance` with an |
| 103 | +explicit workflow ID: |
| 104 | + |
| 105 | +```java |
| 106 | +try (WorkflowStreamClient client = WorkflowStreamClient.newInstance(workflowClient, workflowId)) { |
| 107 | + client.topic("events").publish("from outside", /* forceFlush */ true); |
| 108 | +} |
| 109 | +``` |
| 110 | + |
| 111 | +Items are buffered and flushed automatically every batch interval (default 2s), |
| 112 | +when the buffer reaches the max batch size, on `forceFlush`, on an explicit |
| 113 | +`flush()`, or on `close()`. |
| 114 | + |
| 115 | +## Subscribing |
| 116 | + |
| 117 | +There are two subscriber APIs over one shared poll engine: a non-blocking |
| 118 | +listener and a blocking iterator. Neither occupies a thread while a poll is |
| 119 | +blocked on the server — polling runs on a small executor shared by all of a |
| 120 | +client's subscriptions (2 daemon threads by default; see `pollExecutor`) — so |
| 121 | +many concurrent subscriptions do not mean many threads. Either way, the |
| 122 | +subscription ends cleanly when the workflow reaches a terminal state, |
| 123 | +automatically follows continue-as-new chains, recovers from truncation by |
| 124 | +restarting from the current base offset, and also ends when the owning |
| 125 | +`WorkflowStreamClient` is closed. |
| 126 | + |
| 127 | +Items carry the raw `io.temporal.api.common.v1.Payload`; decode at the call |
| 128 | +site with your data converter. Offsets are **global** (across all topics), not |
| 129 | +per-topic. |
| 130 | + |
| 131 | +### Listener (non-blocking) |
| 132 | + |
| 133 | +Pass a `WorkflowStreamListener` to `subscribe` to have items delivered on the |
| 134 | +poll executor. Callbacks are serialized (never invoked concurrently) and must |
| 135 | +not block; the `CompletionStage` returned by `onNext` is the backpressure |
| 136 | +boundary — return `null` (or a completed stage) to receive the next item |
| 137 | +immediately, or a pending stage to defer further delivery and polling until it |
| 138 | +completes: |
| 139 | + |
| 140 | +```java |
| 141 | +SubscribeOptions options = SubscribeOptions.newBuilder() |
| 142 | + .setTopics("events") // unset = all topics |
| 143 | + .build(); |
| 144 | +WorkflowStreamSubscriptionHandle handle = |
| 145 | + client.subscribe( |
| 146 | + options, |
| 147 | + new WorkflowStreamListener() { |
| 148 | + @Override |
| 149 | + public CompletionStage<Void> onNext(WorkflowStreamItem item) { |
| 150 | + String value = |
| 151 | + DefaultDataConverter.STANDARD_INSTANCE.fromPayload( |
| 152 | + item.getPayload(), String.class, String.class); |
| 153 | + System.out.printf( |
| 154 | + "offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value); |
| 155 | + return null; // or a pending stage to apply backpressure |
| 156 | + } |
| 157 | + |
| 158 | + @Override |
| 159 | + public void onCompleted() { |
| 160 | + System.out.println("stream ended"); |
| 161 | + } |
| 162 | + }); |
| 163 | +``` |
| 164 | + |
| 165 | +`handle.close()` stops the subscription before the next poll (without calling |
| 166 | +`onCompleted`); `handle.getDoneFuture()` completes when the subscription ends — |
| 167 | +normally on a clean end or close, exceptionally with the failure passed to |
| 168 | +`onError`. |
| 169 | + |
| 170 | +### Iterator (blocking) |
| 171 | + |
| 172 | +For synchronous consumers, `subscribe` without a listener returns a blocking, |
| 173 | +single-use subscription; the consuming thread blocks waiting for items while |
| 174 | +polling still runs on the shared executor: |
| 175 | + |
| 176 | +```java |
| 177 | +try (WorkflowStreamSubscription subscription = client.subscribe(options)) { |
| 178 | + for (WorkflowStreamItem item : subscription) { |
| 179 | + String value = |
| 180 | + DefaultDataConverter.STANDARD_INSTANCE.fromPayload( |
| 181 | + item.getPayload(), String.class, String.class); |
| 182 | + System.out.printf("offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value); |
| 183 | + } |
| 184 | +} |
| 185 | +``` |
| 186 | + |
| 187 | +`close()` stops it before the next poll; items already fetched still drain. An |
| 188 | +unrecoverable poll failure is rethrown from `hasNext()`. |
| 189 | + |
| 190 | +## Options |
| 191 | + |
| 192 | +| Option | Default | Meaning | |
| 193 | +| --- | --- | --- | |
| 194 | +| `batchInterval` | 2s | Automatic flush interval | |
| 195 | +| `maxBatchSize` | unset | Flush once the buffer reaches this size | |
| 196 | +| `maxRetryDuration` | 10m | Max time to retry a failed flush before `FlushTimeoutException`. Must be < the workflow's publisher TTL (15m) to preserve exactly-once delivery | |
| 197 | +| `payloadConverters` | standard set | Per-item serialization. Payload conversion only — the client's codec chain runs once on the envelope, never per item | |
| 198 | +| `pollExecutor` | 2 daemon threads, client-owned | Scheduler shared by the client's subscriptions. It runs the short update-admission and delivery steps and poll cooldowns — never held during the long poll itself. A user-supplied executor is never shut down by the client; supply a bigger pool for many subscriptions against slow workflows | |
| 199 | +| `SubscribeOptions.pollCooldown` | 100ms | Min interval between polls | |
| 200 | + |
| 201 | +## Cross-language protocol |
| 202 | + |
| 203 | +The handler names (`WorkflowStreamConstants.PUBLISH_SIGNAL_NAME`, |
| 204 | +`POLL_UPDATE_NAME`, `OFFSET_QUERY_NAME`), the JSON envelope field names, and |
| 205 | +the per-item payload encoding (base64 of the serialized |
| 206 | +`temporal.api.common.v1.Payload`) match other languages' packages |
| 207 | +exactly, so a Java publisher or subscriber interoperates with a workflow |
| 208 | +written in any of them and vice versa. The data converter codec chain |
| 209 | +(encryption, compression) runs once on the signal/update envelope — never per |
| 210 | +item — so payloads are not double-encoded. |
| 211 | + |
| 212 | +One Java-specific caveat: the protocol envelope types are serialized by the |
| 213 | +workflow's and client's *configured* data converter. The default Jackson JSON |
| 214 | +converter produces the wire-compatible snake_case field names (the types are |
| 215 | +annotated with `@JsonProperty`); if you configure a non-Jackson JSON converter, |
| 216 | +it must produce the same field names for cross-language interop. |
0 commit comments