temporalio · brianstrauch · Sep 16, 2026
@@ -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.<String>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<String> 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<T>` 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<Payload>`.
+
+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<String>() {
           @Override
-          public CompletionStage<Void> onNext(WorkflowStreamItem item) {
-            String value = client.decodeItem(item, String.class);
+          public CompletionStage<Void> onNext(WorkflowStreamItem<String> 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<String> subscription =
+    client.subscribe(options, String.class)) {
+  for (WorkflowStreamItem<String> item : subscription) {
+    System.out.printf(
+        "offset=%d topic=%s value=%s%n",
+        item.getOffset(), item.getTopic(), item.getValue());
   }
 }
 ```

@@ -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<T> {
   private final String name;
   private final WorkflowStreamClient client;
+  @Nullable private final Class<T> valueClass;
+  @Nullable private final Type valueType;
 
   TopicHandle(String name, WorkflowStreamClient client) {
+    this(name, client, null, null);
+  }
+
+  TopicHandle(
+      String name,
+      WorkflowStreamClient client,
+      @Nullable Class<T> 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<T> subscribe(long fromOffset) {
+    SubscribeOptions options =
+        SubscribeOptions.newBuilder().setTopics(name).setFromOffset(fromOffset).build();
+    if (valueClass == null) {
+      return (WorkflowStreamSubscription<T>)
+          (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<T> listener) {
+    SubscribeOptions options =
+        SubscribeOptions.newBuilder().setTopics(name).setFromOffset(fromOffset).build();
+    if (valueClass == null) {
+      @SuppressWarnings("unchecked")
+      WorkflowStreamListener<Payload> rawListener =
+          (WorkflowStreamListener<Payload>) (WorkflowStreamListener<?>) listener;
+      return client.subscribe(options, rawListener);
+    }
+    return client.subscribe(options, valueClass, valueType, listener);
   }
 }
@@ -50,7 +50,7 @@ private static final class InternalEntry {
   private final Map<String, Double> publisherLastSeen = new HashMap<>();
   private boolean draining;
 
-  private final Map<String, WorkflowTopicHandle> topicHandles = new HashMap<>();
+  private final Map<String, WorkflowTopicHandle<?>> 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 <T> WorkflowTopicHandle<T> topic(String name) {
+    return (WorkflowTopicHandle<T>)
+        topicHandles.computeIfAbsent(name, n -> new WorkflowTopicHandle<>(n, this));
   }
 
   /** Unblocks all waiting poll handlers and rejects new polls. Used before continue-as-new. */

@@ -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<String, TopicHandle> topicHandles = new HashMap<>();
-  private final Set<SubscriptionDriver> liveSubscriptions = ConcurrentHashMap.newKeySet();
+  private final Map<String, TopicHandle<?>> topicHandles = new HashMap<>();
+  private final Set<SubscriptionDriver<?>> 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> T decodeItem(WorkflowStreamItem item, Class<T> valueClass) {
-    return decodeItem(item, valueClass, valueClass);
+  /** Returns a typed handle for publishing to and subscribing from {@code name}. */
+  public <T> TopicHandle<T> topic(String name, Class<T> valueClass) {
+    return topic(name, valueClass, valueClass);
   }
 
-  /** Decodes an item using this client's configured, codec-free item converter. */
-  public <T> T decodeItem(WorkflowStreamItem item, Class<T> 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 <T> TopicHandle<T> topic(String name, Class<T> 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:
    *
    * <pre>{@code
-   * try (WorkflowStreamSubscription subscription = streamClient.subscribe(options)) {
-   *   for (WorkflowStreamItem item : subscription) {
+   * try (WorkflowStreamSubscription<Payload> subscription = streamClient.subscribe(options)) {
+   *   for (WorkflowStreamItem<Payload> item : subscription) {
    *     // use item
    *   }
    * }
    * }</pre>
    *
    * <p>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<Payload> 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 <T> WorkflowStreamSubscription<T> subscribe(
+      SubscribeOptions options, Class<T> valueClass) {
+    return subscribe(options, valueClass, valueClass);
+  }
+
+  /** Returns a typed subscription that decodes each item with this client's item converter. */
+  public <T> WorkflowStreamSubscription<T> subscribe(
+      SubscribeOptions options, Class<T> 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<Payload> listener) {
+    SubscriptionDriver<Payload> driver =
+        newSubscriptionDriver(options, listener, payload -> payload);
+    driver.start();
+    return driver;
+  }
+
+  /** Subscribes a listener that receives items decoded with this client's item converter. */
+  public <T> WorkflowStreamSubscriptionHandle subscribe(
+      SubscribeOptions options, Class<T> valueClass, WorkflowStreamListener<T> listener) {
+    return subscribe(options, valueClass, valueClass, listener);
+  }
+
+  /** Subscribes a listener that receives items decoded with this client's item converter. */
+  public <T> WorkflowStreamSubscriptionHandle subscribe(
+      SubscribeOptions options,
+      Class<T> valueClass,
+      Type valueType,
+      WorkflowStreamListener<T> listener) {
+    SubscriptionDriver<T> 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);
+  <T> SubscriptionDriver<T> newSubscriptionDriver(
+      SubscribeOptions options,
+      WorkflowStreamListener<T> listener,
+      Function<Payload, T> itemDecoder) {
+    SubscriptionDriver<T> 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;

@@ -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.
      *
      * <p>Default: the standard converter set.
      */