diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index 4e7dedf10b..2053f8ac79 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -40,7 +40,7 @@ public class MyWorkflowImpl implements MyWorkflow { @Override public void execute(MyInput input) { // Optionally publish from workflow code: - stream.topic("events").publish("hello from the workflow"); + stream.topic("events").publish("hello from the workflow"); // Run your workflow; the stream serves external publishers and subscribers // for as long as the workflow is running. Block until your workflow's exit @@ -91,7 +91,7 @@ From an activity, use `fromActivity` to target the parent workflow: ```java public void publishActivity() { try (WorkflowStreamClient client = WorkflowStreamClient.fromActivity()) { - TopicHandle topic = client.topic("events"); + TopicHandle topic = client.topic("events", String.class); for (int i = 0; i < 100; i++) { topic.publish("item " + i); } @@ -124,13 +124,19 @@ automatically follows continue-as-new chains, recovers from truncation by restarting from the current base offset, and also ends when the owning `WorkflowStreamClient` is closed. -Items carry the raw `io.temporal.api.common.v1.Payload`. Use -`WorkflowStreamClient.decodeItem` to decode them with the configured stream -item converter. Transfer conversion and payload conversion apply to each item. -Payload codecs apply only once to the surrounding Temporal signal or update -envelope, never to an individual item. Configure matching payload converters -on the workflow and client sides. Offsets are **global** (across all topics), -not per-topic. +Typed subscriptions decode each item with the stream client's configured item +converter before delivery. Every `WorkflowStreamItem` exposes the decoded +value through `getValue()` and retains the raw +`io.temporal.api.common.v1.Payload` through `getPayload()`. Calling +`client.subscribe(options)` or `client.topic(name)` without a type is the raw +escape hatch and yields `WorkflowStreamItem`. + +Transfer conversion and payload conversion apply to each typed item. Payload +codecs apply only once to the surrounding Temporal signal or update envelope, +never to an individual item. Configure matching payload converters on the +workflow and client sides. A topic may have multiple typed views; Java does not +register or enforce one type for a topic name. Offsets are **global** (across +all topics), not per-topic. ### Listener (non-blocking) @@ -148,12 +154,13 @@ SubscribeOptions options = SubscribeOptions.newBuilder() WorkflowStreamSubscriptionHandle handle = client.subscribe( options, - new WorkflowStreamListener() { + String.class, + new WorkflowStreamListener() { @Override - public CompletionStage onNext(WorkflowStreamItem item) { - String value = client.decodeItem(item, String.class); + public CompletionStage onNext(WorkflowStreamItem item) { System.out.printf( - "offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value); + "offset=%d topic=%s value=%s%n", + item.getOffset(), item.getTopic(), item.getValue()); return null; // or a pending stage to apply backpressure } @@ -176,10 +183,12 @@ single-use subscription; the consuming thread blocks waiting for items while polling still runs on the shared executor: ```java -try (WorkflowStreamSubscription subscription = client.subscribe(options)) { - for (WorkflowStreamItem item : subscription) { - String value = client.decodeItem(item, String.class); - System.out.printf("offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value); +try (WorkflowStreamSubscription subscription = + client.subscribe(options, String.class)) { + for (WorkflowStreamItem item : subscription) { + System.out.printf( + "offset=%d topic=%s value=%s%n", + item.getOffset(), item.getTopic(), item.getValue()); } } ``` diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/TopicHandle.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/TopicHandle.java index 09807319e3..54e3b76040 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/TopicHandle.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/TopicHandle.java @@ -1,18 +1,33 @@ package io.temporal.workflowstreams; +import io.temporal.api.common.v1.Payload; import io.temporal.common.Experimental; +import java.lang.reflect.Type; +import javax.annotation.Nullable; /** * Publishes to and subscribes from a single topic. Obtained via {@link WorkflowStreamClient#topic}. */ @Experimental -public final class TopicHandle { +public final class TopicHandle { private final String name; private final WorkflowStreamClient client; + @Nullable private final Class valueClass; + @Nullable private final Type valueType; TopicHandle(String name, WorkflowStreamClient client) { + this(name, client, null, null); + } + + TopicHandle( + String name, + WorkflowStreamClient client, + @Nullable Class valueClass, + @Nullable Type valueType) { this.name = name; this.client = client; + this.valueClass = valueClass; + this.valueType = valueType; } /** Returns the topic name. */ @@ -21,27 +36,42 @@ public String getName() { } /** Buffers {@code value} for publishing on this topic. See {@link #publish(Object, boolean)}. */ - public void publish(Object value) { + public void publish(T value) { publish(value, false); } /** * Buffers {@code value} for publishing on this topic. {@code value} goes through the client's * payload converters immediately, so an unconvertible value fails this call rather than a later - * background flush; a pre-built {@link io.temporal.api.common.v1.Payload} bypasses conversion. - * Pass {@code forceFlush} to wake the publisher and send immediately. + * background flush. Pass {@code forceFlush} to wake the publisher and send immediately. */ - public void publish(Object value, boolean forceFlush) { + public void publish(T value, boolean forceFlush) { client.publishToTopic(name, value, forceFlush); } + /** Buffers a pre-built payload, bypassing item conversion. */ + public void publishPayload(Payload payload) { + publishPayload(payload, false); + } + + /** Buffers a pre-built payload, bypassing item conversion. */ + public void publishPayload(Payload payload, boolean forceFlush) { + client.publishToTopic(name, payload, forceFlush); + } + /** * Returns a subscription over items on this topic, starting at {@code fromOffset}. See {@link * WorkflowStreamClient#subscribe(SubscribeOptions)}. */ - public WorkflowStreamSubscription subscribe(long fromOffset) { - return client.subscribe( - SubscribeOptions.newBuilder().setTopics(name).setFromOffset(fromOffset).build()); + @SuppressWarnings("unchecked") + public WorkflowStreamSubscription subscribe(long fromOffset) { + SubscribeOptions options = + SubscribeOptions.newBuilder().setTopics(name).setFromOffset(fromOffset).build(); + if (valueClass == null) { + return (WorkflowStreamSubscription) + (WorkflowStreamSubscription) client.subscribe(options); + } + return client.subscribe(options, valueClass, valueType); } /** @@ -50,8 +80,15 @@ public WorkflowStreamSubscription subscribe(long fromOffset) { * WorkflowStreamListener)}. */ public WorkflowStreamSubscriptionHandle subscribe( - long fromOffset, WorkflowStreamListener listener) { - return client.subscribe( - SubscribeOptions.newBuilder().setTopics(name).setFromOffset(fromOffset).build(), listener); + long fromOffset, WorkflowStreamListener listener) { + SubscribeOptions options = + SubscribeOptions.newBuilder().setTopics(name).setFromOffset(fromOffset).build(); + if (valueClass == null) { + @SuppressWarnings("unchecked") + WorkflowStreamListener rawListener = + (WorkflowStreamListener) (WorkflowStreamListener) listener; + return client.subscribe(options, rawListener); + } + return client.subscribe(options, valueClass, valueType, listener); } } diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java index 406f133b7f..e38434be39 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java @@ -50,7 +50,7 @@ private static final class InternalEntry { private final Map publisherLastSeen = new HashMap<>(); private boolean draining; - private final Map topicHandles = new HashMap<>(); + private final Map> topicHandles = new HashMap<>(); /** Constructs a stream with no prior state and default options. */ public static WorkflowStream newInstance() { @@ -100,8 +100,10 @@ private WorkflowStream(@Nullable WorkflowStreamState priorState, WorkflowStreamO * Returns a handle for publishing to {@code name}. Repeated calls with the same name return the * same handle. */ - public WorkflowTopicHandle topic(String name) { - return topicHandles.computeIfAbsent(name, n -> new WorkflowTopicHandle(n, this)); + @SuppressWarnings("unchecked") + public WorkflowTopicHandle topic(String name) { + return (WorkflowTopicHandle) + topicHandles.computeIfAbsent(name, n -> new WorkflowTopicHandle<>(n, this)); } /** Unblocks all waiting poll handlers and rejects new polls. Used before continue-as-new. */ diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java index c3c3fa7ef1..bd9d82f486 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java @@ -2,6 +2,7 @@ import io.temporal.activity.Activity; import io.temporal.activity.ActivityExecutionContext; +import io.temporal.api.common.v1.Payload; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowStub; import io.temporal.common.Experimental; @@ -17,6 +18,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; import javax.annotation.Nullable; /** @@ -35,8 +37,8 @@ public final class WorkflowStreamClient implements AutoCloseable { private final DataConverter itemDataConverter; @Nullable private final ScheduledExecutorService userPollExecutor; - private final Map topicHandles = new HashMap<>(); - private final Set liveSubscriptions = ConcurrentHashMap.newKeySet(); + private final Map> topicHandles = new HashMap<>(); + private final Set> liveSubscriptions = ConcurrentHashMap.newKeySet(); // Lazily created when the first subscription needs it and no user executor was supplied; // owned by this client and shut down in close(). Guarded by `this`. @@ -103,17 +105,21 @@ private WorkflowStreamClient( * same name return the same handle. */ public synchronized TopicHandle topic(String name) { - return topicHandles.computeIfAbsent(name, n -> new TopicHandle(n, this)); + return topicHandles.computeIfAbsent(name, n -> new TopicHandle<>(n, this)); } - /** Decodes an item using this client's configured, codec-free item converter. */ - public T decodeItem(WorkflowStreamItem item, Class valueClass) { - return decodeItem(item, valueClass, valueClass); + /** Returns a typed handle for publishing to and subscribing from {@code name}. */ + public TopicHandle topic(String name, Class valueClass) { + return topic(name, valueClass, valueClass); } - /** Decodes an item using this client's configured, codec-free item converter. */ - public T decodeItem(WorkflowStreamItem item, Class valueClass, Type valueType) { - return itemDataConverter.fromPayload(item.getPayload(), valueClass, valueType); + /** + * Returns a typed handle for publishing to and subscribing from {@code name}. Each call creates + * an independent typed view, so the same topic may be consumed using more than one compatible + * type. + */ + public TopicHandle topic(String name, Class valueClass, Type valueType) { + return new TopicHandle<>(name, this, valueClass, valueType); } /** @@ -137,20 +143,37 @@ public long getOffset() { * Returns a subscription that long-polls for new items. Iterate with: * * 
{@code
-   * try (WorkflowStreamSubscription subscription = streamClient.subscribe(options)) {
-   *   for (WorkflowStreamItem item : subscription) {
+   * try (WorkflowStreamSubscription subscription = streamClient.subscribe(options)) {
+   *   for (WorkflowStreamItem item : subscription) {
    *     // use item
    *   }
    * }
    * }
* *

The consuming thread blocks waiting for items; polling itself runs on the client's poll - * executor. Decode each item with {@link #decodeItem(WorkflowStreamItem, Class)}. The - * subscription ends cleanly when the workflow reaches a terminal state, automatically follows - * continue-as-new chains, and also ends when this client is closed. + * executor. The subscription ends cleanly when the workflow reaches a terminal state, + * automatically follows continue-as-new chains, and also ends when this client is closed. */ - public WorkflowStreamSubscription subscribe(SubscribeOptions options) { - return new WorkflowStreamSubscription(listener -> newSubscriptionDriver(options, listener)); + public WorkflowStreamSubscription subscribe(SubscribeOptions options) { + return new WorkflowStreamSubscription<>( + listener -> newSubscriptionDriver(options, listener, payload -> payload)); + } + + /** Returns a typed subscription that decodes each item with this client's item converter. */ + public WorkflowStreamSubscription subscribe( + SubscribeOptions options, Class valueClass) { + return subscribe(options, valueClass, valueClass); + } + + /** Returns a typed subscription that decodes each item with this client's item converter. */ + public WorkflowStreamSubscription subscribe( + SubscribeOptions options, Class valueClass, Type valueType) { + return new WorkflowStreamSubscription<>( + listener -> + newSubscriptionDriver( + options, + listener, + payload -> itemDataConverter.fromPayload(payload, valueClass, valueType))); } /** @@ -165,17 +188,47 @@ public WorkflowStreamSubscription subscribe(SubscribeOptions options) { * WorkflowStreamSubscriptionHandle#close}; closing this client also stops it. */ public WorkflowStreamSubscriptionHandle subscribe( - SubscribeOptions options, WorkflowStreamListener listener) { - SubscriptionDriver driver = newSubscriptionDriver(options, listener); + SubscribeOptions options, WorkflowStreamListener listener) { + SubscriptionDriver driver = + newSubscriptionDriver(options, listener, payload -> payload); + driver.start(); + return driver; + } + + /** Subscribes a listener that receives items decoded with this client's item converter. */ + public WorkflowStreamSubscriptionHandle subscribe( + SubscribeOptions options, Class valueClass, WorkflowStreamListener listener) { + return subscribe(options, valueClass, valueClass, listener); + } + + /** Subscribes a listener that receives items decoded with this client's item converter. */ + public WorkflowStreamSubscriptionHandle subscribe( + SubscribeOptions options, + Class valueClass, + Type valueType, + WorkflowStreamListener listener) { + SubscriptionDriver driver = + newSubscriptionDriver( + options, + listener, + payload -> itemDataConverter.fromPayload(payload, valueClass, valueType)); driver.start(); return driver; } - SubscriptionDriver newSubscriptionDriver( - SubscribeOptions options, WorkflowStreamListener listener) { - SubscriptionDriver driver = - new SubscriptionDriver( - client, workflowId, options, pollExecutor(), listener, liveSubscriptions::remove); + SubscriptionDriver newSubscriptionDriver( + SubscribeOptions options, + WorkflowStreamListener listener, + Function itemDecoder) { + SubscriptionDriver driver = + new SubscriptionDriver<>( + client, + workflowId, + options, + pollExecutor(), + listener, + itemDecoder, + liveSubscriptions::remove); liveSubscriptions.add(driver); return driver; } @@ -215,7 +268,7 @@ private ScheduledExecutorService pollExecutor() { @Override public void close() { publisher.close(); - for (SubscriptionDriver driver : liveSubscriptions.toArray(new SubscriptionDriver[0])) { + for (SubscriptionDriver driver : liveSubscriptions.toArray(new SubscriptionDriver[0])) { driver.close(); } ScheduledExecutorService owned; diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java index c4ba4e8504..204c5c6d4c 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java @@ -102,8 +102,7 @@ public Builder setMaxRetryDuration(Duration maxRetryDuration) { * The codec chain configured on the Temporal client runs once on the signal/update envelope * that carries each batch, so encoding items here too would double-encode them; the {@code * PayloadConverter[]} type makes that mistake impossible. Transfer conversion and payload - * conversion apply to each item. Decode subscribed items with {@link - * WorkflowStreamClient#decodeItem(WorkflowStreamItem, Class)}. + * conversion apply to each item in a typed subscription. * *

Default: the standard converter set. */ diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java index 9b1c7bddf9..1c75a28121 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java @@ -3,20 +3,25 @@ import io.temporal.api.common.v1.Payload; import io.temporal.common.Experimental; -/** - * A single decoded item yielded by a subscription. {@code payload} is the raw {@link Payload}; use - * {@link WorkflowStreamClient#decodeItem(WorkflowStreamItem, Class)} to decode it with the stream - * client's configured item converter. - */ +/** A single item yielded by a subscription, including its typed value and raw {@link Payload}. */ @Experimental -public final class WorkflowStreamItem { +public final class WorkflowStreamItem { private final String topic; private final Payload payload; + private final T value; private final long offset; + /** Creates a raw item whose value is its payload. */ + @SuppressWarnings("unchecked") public WorkflowStreamItem(String topic, Payload payload, long offset) { + this(topic, payload, (T) payload, offset); + } + + /** Creates an item with both its raw payload and decoded value. */ + public WorkflowStreamItem(String topic, Payload payload, T value, long offset) { this.topic = topic; this.payload = payload; + this.value = value; this.offset = offset; } @@ -28,6 +33,10 @@ public Payload getPayload() { return payload; } + public T getValue() { + return value; + } + public long getOffset() { return offset; } diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamListener.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamListener.java index 9326c6f456..ce337de247 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamListener.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamListener.java @@ -14,14 +14,14 @@ * delivery, return a pending stage from {@link #onNext}. */ @Experimental -public interface WorkflowStreamListener { +public interface WorkflowStreamListener { /** * Called with the next item on the stream. Return {@code null} or an already-completed stage to * receive the next item immediately; return a pending stage to defer both further delivery and * the next poll until it completes (backpressure). A stage that completes exceptionally — or an * exception thrown directly — stops the subscription and is reported to {@link #onError}. */ - CompletionStage onNext(WorkflowStreamItem item); + CompletionStage onNext(WorkflowStreamItem item); /** * Called once when the subscription stops because of an unrecoverable failure (including a diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamSubscription.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamSubscription.java index b3a51ac70f..e7eadfcb0e 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamSubscription.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamSubscription.java @@ -22,9 +22,9 @@ * server is not interrupted. */ @Experimental -public final class WorkflowStreamSubscription - implements Iterator, Iterable, AutoCloseable { - private final SubscriptionDriver driver; +public final class WorkflowStreamSubscription + implements Iterator>, Iterable>, AutoCloseable { + private final SubscriptionDriver driver; private final Object lock = new Object(); @@ -32,7 +32,7 @@ public final class WorkflowStreamSubscription // at no more than one item: each onNext parks the driver on a gate that next() releases when // the consumer takes the item, so the next long poll only fires once the consumer drains what // the driver already fetched — the same pacing as driving the poll loop on the consumer thread. - private final Deque buffer = new ArrayDeque<>(); + private final Deque> buffer = new ArrayDeque<>(); private CompletableFuture pendingGate; private Throwable error; private boolean streamDone; @@ -41,7 +41,8 @@ public final class WorkflowStreamSubscription private boolean started; private boolean errorThrown; - WorkflowStreamSubscription(Function driverFactory) { + WorkflowStreamSubscription( + Function, SubscriptionDriver> driverFactory) { this.driver = driverFactory.apply(new AdapterListener()); // One hook covers every way the stream ends: terminal state, failure, close(), and the // owning client closing. It wakes a consumer blocked in hasNext(). @@ -64,7 +65,7 @@ public final class WorkflowStreamSubscription * for-each loop). */ @Override - public Iterator iterator() { + public Iterator> iterator() { return this; } @@ -104,11 +105,11 @@ public boolean hasNext() { } @Override - public WorkflowStreamItem next() { + public WorkflowStreamItem next() { if (!hasNext()) { throw new NoSuchElementException(); } - WorkflowStreamItem item; + WorkflowStreamItem item; CompletableFuture gate = null; synchronized (lock) { item = buffer.poll(); @@ -139,9 +140,9 @@ public void close() { } } - private class AdapterListener implements WorkflowStreamListener { + private class AdapterListener implements WorkflowStreamListener { @Override - public CompletionStage onNext(WorkflowStreamItem item) { + public CompletionStage onNext(WorkflowStreamItem item) { CompletableFuture gate = new CompletableFuture<>(); synchronized (lock) { buffer.add(item); diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowTopicHandle.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowTopicHandle.java index b447b6ce33..5a4f06bdae 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowTopicHandle.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowTopicHandle.java @@ -1,10 +1,11 @@ package io.temporal.workflowstreams; +import io.temporal.api.common.v1.Payload; import io.temporal.common.Experimental; /** Publishes to a single topic from workflow code. Obtained via {@link WorkflowStream#topic}. */ @Experimental -public final class WorkflowTopicHandle { +public final class WorkflowTopicHandle { private final String name; private final WorkflowStream stream; @@ -21,9 +22,14 @@ public String getName() { /** * Appends {@code value} to the stream on this topic. {@code value} is serialized by the stream's * payload converters (see {@link WorkflowStreamOptions.Builder#setPayloadConverters}), defaulting - * to the standard set; a pre-built {@link io.temporal.api.common.v1.Payload} bypasses conversion. + * to the standard set. */ - public void publish(Object value) { + public void publish(T value) { stream.publishToTopic(name, value); } + + /** Appends a pre-built payload to the stream, bypassing item conversion. */ + public void publishPayload(Payload payload) { + stream.publishToTopic(name, payload); + } } diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java index fba824574f..cb5e89f3f3 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java @@ -1,5 +1,6 @@ package io.temporal.workflowstreams.internal; +import io.temporal.api.common.v1.Payload; import io.temporal.api.enums.v1.WorkflowExecutionStatus; import io.temporal.client.UpdateOptions; import io.temporal.client.WorkflowClient; @@ -27,6 +28,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; +import java.util.function.Function; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,7 +46,7 @@ *

This class is public only for internal wiring; construct subscriptions through {@link * io.temporal.workflowstreams.WorkflowStreamClient} instead. */ -public final class SubscriptionDriver implements WorkflowStreamSubscriptionHandle { +public final class SubscriptionDriver implements WorkflowStreamSubscriptionHandle { private static final Logger log = LoggerFactory.getLogger(SubscriptionDriver.class); private final WorkflowClient client; @@ -53,8 +55,9 @@ public final class SubscriptionDriver implements WorkflowStreamSubscriptionHandl private final List topics; private final long pollCooldownMs; private final ScheduledExecutorService executor; - private final WorkflowStreamListener listener; - private final Consumer onFinish; + private final WorkflowStreamListener listener; + private final Function itemDecoder; + private final Consumer> onFinish; private final CompletableFuture doneFuture = new CompletableFuture<>(); // Mutable poll state, owned by the single outstanding step (see class javadoc). @@ -73,8 +76,9 @@ public SubscriptionDriver( String workflowId, SubscribeOptions options, ScheduledExecutorService executor, - WorkflowStreamListener listener, - Consumer onFinish) { + WorkflowStreamListener listener, + Function itemDecoder, + Consumer> onFinish) { this.client = client; this.workflowId = workflowId; this.latestRunStub = client.newUntypedWorkflowStub(workflowId); @@ -83,6 +87,7 @@ public SubscriptionDriver( this.pollCooldownMs = options.getPollCooldown().toMillis(); this.executor = executor; this.listener = listener; + this.itemDecoder = itemDecoder; this.onFinish = onFinish; } @@ -151,15 +156,17 @@ private void deliver(PollResult result) { finishSilent(); return; } - List items = new ArrayList<>(result.items.size()); + List> items = new ArrayList<>(result.items.size()); for (WireItem item : result.items) { - items.add(new WorkflowStreamItem(item.topic, PayloadWire.decode(item.data), item.offset)); + Payload payload = PayloadWire.decode(item.data); + items.add( + new WorkflowStreamItem<>(item.topic, payload, itemDecoder.apply(payload), item.offset)); } offset = result.nextOffset; deliverFrom(items, 0, result.moreReady); } - private void deliverFrom(List items, int start, boolean moreReady) { + private void deliverFrom(List> items, int start, boolean moreReady) { // Iterate (rather than recurse) over items whose stages complete immediately, so a large // batch of synchronous onNext calls cannot grow the stack. for (int i = start; ; i++) { diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java index 97ca6e7b23..e1b1efca2d 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java @@ -19,6 +19,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.Assert; import org.junit.Rule; @@ -154,6 +155,75 @@ public void testTopicHandleListenerSubscribeFilters() throws Exception { stub.getResult(Void.class); } + @Test + public void testTypedListenerReceivesDecodedValue() throws Exception { + WorkflowStub stub = startHostWorkflow(); + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + streamClient.topic("evt").publish("value", true); + streamClient.flush(); + CountDownLatch delivered = new CountDownLatch(1); + AtomicReference> received = new AtomicReference<>(); + + try (WorkflowStreamSubscriptionHandle handle = + streamClient + .topic("evt", String.class) + .subscribe( + 0, + item -> { + received.set(item); + delivered.countDown(); + return null; + })) { + await(delivered, "typed item"); + Assert.assertEquals("value", received.get().getValue()); + Assert.assertEquals("value", decode(received.get())); + } + } + stub.signal("finish"); + stub.getResult(Void.class); + } + + @Test + public void testTypedListenerConversionFailureCallsOnError() throws Exception { + WorkflowStub stub = startHostWorkflow(); + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + streamClient.topic("evt").publish("not-an-integer", true); + streamClient.flush(); + CountDownLatch failed = new CountDownLatch(1); + AtomicReference error = new AtomicReference<>(); + AtomicInteger deliveries = new AtomicInteger(); + + WorkflowStreamSubscriptionHandle handle = + streamClient.subscribe( + FAST_POLL, + Integer.class, + new WorkflowStreamListener() { + @Override + public CompletionStage onNext(WorkflowStreamItem item) { + deliveries.incrementAndGet(); + return null; + } + + @Override + public void onError(Throwable failure) { + error.set(failure); + failed.countDown(); + } + }); + await(failed, "conversion failure"); + Assert.assertNotNull(error.get()); + Assert.assertEquals(0, deliveries.get()); + try { + handle.getDoneFuture().get(TIMEOUT_MS, TimeUnit.MILLISECONDS); + Assert.fail("done future must complete exceptionally"); + } catch (ExecutionException e) { + Assert.assertSame(error.get(), e.getCause()); + } + } + stub.signal("finish"); + stub.getResult(Void.class); + } + @Test public void testTerminalCallsOnCompleted() throws Exception { WorkflowStub stub = startHostWorkflow(); diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java index 66403721a0..9aee8da2f2 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java @@ -1,5 +1,6 @@ package io.temporal.workflowstreams; +import com.fasterxml.jackson.core.type.TypeReference; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowStub; @@ -10,6 +11,8 @@ import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflow; import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflowImpl; import java.time.Duration; +import java.util.Arrays; +import java.util.List; import org.junit.Assert; import org.junit.Rule; import org.junit.Test; @@ -55,6 +58,7 @@ public void testSubscribeDeliversItemsAndAdvancesOffset() { Assert.assertTrue(subscription.hasNext()); WorkflowStreamItem first = subscription.next(); Assert.assertEquals("evt", first.getTopic()); + Assert.assertSame(first.getPayload(), first.getValue()); Assert.assertEquals("a", decode(first)); Assert.assertEquals(0, first.getOffset()); @@ -77,16 +81,16 @@ public void testClientPublishedItemUsesTransferConverter() { streamClient.topic("evt").publish(new TransferStreamTestModel("client"), true); streamClient.flush(); - try (WorkflowStreamSubscription subscription = streamClient.subscribe(FAST_POLL)) { - WorkflowStreamItem item = subscription.next(); + try (WorkflowStreamSubscription subscription = + streamClient.subscribe(FAST_POLL, TransferStreamTestModel.class)) { + WorkflowStreamItem item = subscription.next(); Assert.assertEquals( "The configured payload converter must receive the protobuf transfer representation", "json/protobuf", item.getPayload() .getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY) .toStringUtf8()); - TransferStreamTestModel result = - streamClient.decodeItem(item, TransferStreamTestModel.class); + TransferStreamTestModel result = item.getValue(); Assert.assertEquals(new TransferStreamTestModel("client"), result); Assert.assertTrue(result.wasTransferred()); } @@ -101,9 +105,9 @@ public void testWorkflowPublishedItemUsesTransferConverter() { try (WorkflowStreamClient streamClient = newStreamClient(stub)) { stub.signal("publishTransfer", "evt", new TransferStreamTestModel("workflow")); - try (WorkflowStreamSubscription subscription = streamClient.subscribe(FAST_POLL)) { - TransferStreamTestModel result = - streamClient.decodeItem(subscription.next(), TransferStreamTestModel.class); + try (WorkflowStreamSubscription subscription = + streamClient.topic("evt", TransferStreamTestModel.class).subscribe(0)) { + TransferStreamTestModel result = subscription.next().getValue(); Assert.assertEquals(new TransferStreamTestModel("workflow"), result); Assert.assertTrue(result.wasTransferred()); } @@ -132,6 +136,66 @@ public void testTopicHandleSubscribeFilters() { stub.getResult(Void.class); } + @Test + public void testTypedTopicAllowsMultipleViews() { + WorkflowStub stub = startHostWorkflow(); + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + TopicHandle strings = streamClient.topic("evt", String.class); + TopicHandle objects = streamClient.topic("evt", Object.class); + strings.publish("value", true); + streamClient.flush(); + + try (WorkflowStreamSubscription stringSubscription = strings.subscribe(0); + WorkflowStreamSubscription objectSubscription = objects.subscribe(0)) { + Assert.assertEquals("value", stringSubscription.next().getValue()); + Assert.assertEquals("value", objectSubscription.next().getValue()); + } + } + stub.signal("finish"); + stub.getResult(Void.class); + } + + @Test + public void testTypedSubscriptionUsesParameterizedType() { + WorkflowStub stub = startHostWorkflow(); + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + streamClient.topic("evt").publish(Arrays.asList("a", "b"), true); + streamClient.flush(); + + try (WorkflowStreamSubscription subscription = + streamClient.subscribe( + FAST_POLL, List.class, new TypeReference>() {}.getType())) { + WorkflowStreamItem item = subscription.next(); + Assert.assertEquals(Arrays.asList("a", "b"), item.getValue()); + Assert.assertNotNull(item.getPayload()); + } + } + stub.signal("finish"); + stub.getResult(Void.class); + } + + @Test + public void testTypedConversionFailureSurfacesFromHasNext() { + WorkflowStub stub = startHostWorkflow(); + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + streamClient.topic("evt").publish("not-an-integer", true); + streamClient.flush(); + + try (WorkflowStreamSubscription subscription = + streamClient.subscribe(FAST_POLL, Integer.class)) { + try { + subscription.hasNext(); + Assert.fail("expected item conversion to fail"); + } catch (RuntimeException e) { + // Expected. + } + Assert.assertFalse(subscription.hasNext()); + } + } + stub.signal("finish"); + stub.getResult(Void.class); + } + @Test public void testSubscribeEndsCleanlyOnTerminal() { WorkflowStub stub = startHostWorkflow(); diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java index d00cd93fbb..ed623097d4 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java @@ -66,12 +66,12 @@ public void rollover() { @Override public void publishLocal(String topic, String value) { - stream.topic(topic).publish(value); + stream.topic(topic).publish(value); } @Override public void publishTransfer(String topic, TransferStreamTestModel value) { - stream.topic(topic).publish(value); + stream.topic(topic).publish(value); } @Override diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamCodecTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamCodecTest.java index 99a4952a60..ec66355023 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamCodecTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamCodecTest.java @@ -56,11 +56,11 @@ public void codecsApplyToTheEnvelopeButNotIndividualItems() { .build())) { client.topic("events").publish("value", true); client.flush(); - try (WorkflowStreamSubscription subscription = - client.subscribe(SubscribeOptions.getDefaultInstance())) { - WorkflowStreamItem item = subscription.next(); + try (WorkflowStreamSubscription subscription = + client.subscribe(SubscribeOptions.getDefaultInstance(), String.class)) { + WorkflowStreamItem item = subscription.next(); Assert.assertFalse(item.getPayload().containsMetadata(CODEC_METADATA_KEY)); - Assert.assertEquals("value", client.decodeItem(item, String.class)); + Assert.assertEquals("value", item.getValue()); } } Assert.assertTrue(CODEC.encodeCalls.get() > 0);