Stand alone operations for Nexus

temporal-sdk/src/main/java/io/temporal/client/ActivityHandle.java

@@ -30,7 +30,8 @@ public interface ActivityHandle<R> extends UntypedActivityHandle {
 
   /**
    * Blocks until the standalone activity completes and returns the typed result, or throws if the
-   * client-side timeout expires first.
+   *
+   * <p>client-side timeout expires first.
    *
    * @param timeout maximum time to wait
    * @param unit unit of {@code timeout}

temporal-sdk/src/main/java/io/temporal/client/NexusClient.java

@@ -0,0 +1,151 @@
+package io.temporal.client;
+
+import io.temporal.common.Experimental;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import java.lang.reflect.Type;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+
+/**
+ * Client for managing standalone Nexus operation executions. Obtain an instance via {@link
+ * #newInstance(WorkflowServiceStubs)} or {@link #newInstance(WorkflowServiceStubs,
+ * NexusClientOptions)}. Do not create this object per request; share it for the lifetime of the
+ * process.
+ *
+ * <p>Standalone Nexus operations run independently of any workflow — they are scheduled, monitored,
+ * and managed directly through this client (and the service-bound clients it produces) rather than
+ * from within a workflow execution.
+ *
+ * <p>To start operations, build a service-bound client and call {@code start}/{@code execute}:
+ *
+ * <pre>{@code
+ * NexusClient client = NexusClient.newInstance(stubs, options);
+ *
+ * // Typed: bind to an @ServiceInterface and invoke a method reference.
+ * NexusServiceClient<MyService> svc =
+ *     NexusServiceClient.newInstance(MyService.class, "my-endpoint", stubs, options);
+ * String result = svc.execute(MyService::greet, "world");
+ *
+ * // Untyped: dispatch by operation name string.
+ * UntypedNexusServiceClient untyped =
+ *     client.newUntypedNexusServiceClient("my-endpoint", "MyService");
+ * UntypedNexusOperationHandle handle = untyped.start("greet", null, "world");
+ * }</pre>
+ *
+ * <p>To act on an existing operation (describe, cancel, terminate, get result), obtain a handle via
+ * {@link #getHandle}:
+ *
+ * <pre>{@code
+ * NexusOperationHandle<String> handle = client.getHandle(operationId, runId, String.class);
+ * String result = handle.getResult();
+ * handle.cancel("user requested");
+ * }</pre>
+ *
+ * <p>For visibility queries across all operations in the namespace, see {@link
+ * #listNexusOperationExecutions} and {@link #countNexusOperationExecutions}.
+ *
+ * @see NexusServiceClient
+ * @see UntypedNexusServiceClient
+ * @see NexusOperationHandle
+ */
+@Experimental
+public interface NexusClient {
+
+  /**
+   * Creates a client with default {@link NexusClientOptions}.
+   *
+   * @param service gRPC stubs connected to a Temporal Service endpoint
+   */
+  static NexusClient newInstance(WorkflowServiceStubs service) {
+    return NexusClientImpl.newInstance(service, NexusClientOptions.getDefaultInstance());
+  }
+
+  /**
+   * Creates a client with the supplied options.
+   *
+   * @param service gRPC stubs connected to a Temporal Service endpoint
+   * @param options namespace, data converter, interceptors, and defaults applied to operations
+   *     started through this client
+   */
+  static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) {
+    return NexusClientImpl.newInstance(service, options);
+  }
+
+  /** Returns the underlying gRPC stubs this client routes RPCs through. */
+  WorkflowServiceStubs getWorkflowServiceStubs();
+
+  /**
+   * Returns an untyped handle to an existing operation execution, targeting the latest run. To bind
+   * a result type, wrap the handle with {@link NexusOperationHandle#fromUntyped}.
+   *
+   * @param operationId the user-assigned operation ID
+   * @return an untyped handle
+   */
+  UntypedNexusOperationHandle getHandle(String operationId);
+
+  /**
+   * Returns an untyped handle to an existing operation execution, optionally pinned to a specific
+   * run.
+   *
+   * @param operationId the user-assigned operation ID
+   * @param runId the server-assigned run ID, or {@code null} to target the latest run
+   * @return an untyped handle
+   */
+  UntypedNexusOperationHandle getHandle(String operationId, @Nullable String runId);
+
+  /**
+   * Returns a typed handle to an existing operation execution, bound to {@code resultClass}.
+   *
+   * @param operationId the user-assigned operation ID
+   * @param runId the server-assigned run ID, or {@code null} to target the latest run
+   * @param resultClass expected result type
+   * @param <R> result type
+   */
+  <R> NexusOperationHandle<R> getHandle(
+      String operationId, @Nullable String runId, Class<R> resultClass);
+
+  /**
+   * Returns a typed handle to an existing operation execution, bound to {@code resultClass}/{@code
+   * resultType}. Use the {@code resultType} variant when the result is a generic type whose
+   * parameters cannot be captured by {@link Class} alone (e.g. {@code List<String>}).
+   *
+   * @param operationId the user-assigned operation ID
+   * @param runId the server-assigned run ID, or {@code null} to target the latest run
+   * @param resultClass expected result class
+   * @param resultType generic type for deserialization; may be {@code null}
+   * @param <R> result type
+   */
+  <R> NexusOperationHandle<R> getHandle(
+      String operationId, @Nullable String runId, Class<R> resultClass, @Nullable Type resultType);
+
+  /**
+   * Builds an untyped service-bound client targeting the given endpoint and service. Use this to
+   * dispatch operations by name string when no service interface is available.
+   *
+   * @param endpoint Nexus endpoint name registered on the Temporal Service
+   * @param serviceName Nexus service name on that endpoint
+   */
+  UntypedNexusServiceClient newUntypedNexusServiceClient(String endpoint, String serviceName);
+
+  /**
+   * Returns a stream of standalone Nexus operation executions matching the given visibility query.
+   * The stream paginates lazily over server-side results — pages are fetched on demand as the
+   * stream is consumed.
+   *
+   * @param query Temporal visibility query string, or {@code null} to return all executions in the
+   *     client namespace
+   * @return a lazy stream of matching executions
+   */
+  Stream<NexusOperationExecutionMetadata> listNexusOperationExecutions(@Nullable String query);
+
+  /**
+   * Returns the count of standalone Nexus operation executions matching the given visibility query,
+   * optionally with aggregation groups.
+   *
+   * @param query Temporal visibility query string, or {@code null} to count all executions in the
+   *     client namespace
+   * @return execution count, optionally with aggregation groups when the query uses {@code GROUP
+   *     BY}
+   */
+  NexusOperationExecutionCount countNexusOperationExecutions(@Nullable String query);
+}

temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java

@@ -0,0 +1,154 @@
+package io.temporal.client;
+
+import static io.temporal.internal.WorkflowThreadMarker.enforceNonWorkflowThread;
+
+import com.uber.m3.tally.Scope;
+import io.temporal.common.Experimental;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.CountNexusOperationExecutionsInput;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.CountNexusOperationExecutionsOutput;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.ListNexusOperationExecutionsInput;
+import io.temporal.common.interceptors.NexusClientCallsInterceptor.ListNexusOperationExecutionsOutput;
+import io.temporal.common.interceptors.NexusClientInterceptor;
+import io.temporal.internal.WorkflowThreadMarker;
+import io.temporal.internal.client.NamespaceInjectWorkflowServiceStubs;
+import io.temporal.internal.client.NexusOperationHandleImpl;
+import io.temporal.internal.client.RootNexusClientInvoker;
+import io.temporal.internal.client.external.GenericWorkflowClient;
+import io.temporal.internal.client.external.GenericWorkflowClientImpl;
+import io.temporal.serviceclient.MetricsTag;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Experimental
+public class NexusClientImpl implements NexusClient {
+
+  private static final Logger log = LoggerFactory.getLogger(NexusClientImpl.class);
+
+  private final WorkflowServiceStubs workflowServiceStubs;
+  private final NexusClientOptions options;
+  private final GenericWorkflowClient genericClient;
+  private final Scope metricsScope;
+  private final NexusClientCallsInterceptor nexusClientCallsInvoker;
+  private final List<NexusClientInterceptor> interceptors;
+
+  public static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) {
+    enforceNonWorkflowThread();
+    return WorkflowThreadMarker.protectFromWorkflowThread(
+        new NexusClientImpl(service, options), NexusClient.class);
+  }
+
+  NexusClientImpl(WorkflowServiceStubs workflowServiceStubs, NexusClientOptions options) {
+    workflowServiceStubs =
+        new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace());
+    this.workflowServiceStubs = workflowServiceStubs;
+    this.options = options;
+    this.metricsScope =
+        workflowServiceStubs
+            .getOptions()
+            .getMetricsScope()
+            .tagged(MetricsTag.defaultTags(options.getNamespace()));
+    this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
+    this.interceptors = options.getInterceptors();
+    this.nexusClientCallsInvoker = initializeClientInvoker();
+    if (log.isDebugEnabled()) {
+      log.debug(
+          "NexusClient initialized: namespace={}, interceptors={}",
+          options.getNamespace(),
+          interceptors.size());
+    }
+  }
+
+  private NexusClientCallsInterceptor initializeClientInvoker() {
+    NexusClientCallsInterceptor invoker = new RootNexusClientInvoker(genericClient, options);
+    for (NexusClientInterceptor clientInterceptor : interceptors) {
+      NexusClientCallsInterceptor wrapped = clientInterceptor.nexusClientCallsInterceptor(invoker);
+      if (wrapped == null) {
+        throw new IllegalStateException(
+            "NexusClientInterceptor "
+                + clientInterceptor.getClass().getName()
+                + " returned null from nexusClientCallsInterceptor; expected a non-null"
+                + " NexusClientCallsInterceptor wrapping the supplied next link");
+      }
+      invoker = wrapped;
+    }
+    return invoker;
+  }
+
+  @Override
+  public WorkflowServiceStubs getWorkflowServiceStubs() {
+    return workflowServiceStubs;
+  }
+
+  @Override
+  public UntypedNexusOperationHandle getHandle(String operationId) {
+    return getHandle(operationId, null);
+  }
+
+  @Override
+  public UntypedNexusOperationHandle getHandle(String operationId, @Nullable String runId) {
+    return new NexusOperationHandleImpl(
+        operationId, runId, nexusClientCallsInvoker, options.getDataConverter());
+  }
+
+  @Override
+  public <R> NexusOperationHandle<R> getHandle(
+      String operationId, @Nullable String runId, Class<R> resultClass) {
+    return getHandle(operationId, runId, resultClass, null);
+  }
+
+  @Override
+  public <R> NexusOperationHandle<R> getHandle(
+      String operationId,
+      @Nullable String runId,
+      Class<R> resultClass,
+      @Nullable java.lang.reflect.Type resultType) {
+    return NexusOperationHandle.fromUntyped(getHandle(operationId, runId), resultClass, resultType);
+  }
+
+  @Override
+  public UntypedNexusServiceClient newUntypedNexusServiceClient(
+      String endpoint, String serviceName) {
+    return new UntypedNexusServiceClientImpl(
+        nexusClientCallsInvoker, endpoint, serviceName, options);
+  }
+
+  /**
+   * Returns the head of the interceptor chain. Package-private so service-client builders can route
+   * start RPCs through the chain without exposing it on the public {@link NexusClient} interface.
+   */
+  NexusClientCallsInterceptor getNexusClientCallsInvoker() {
+    return nexusClientCallsInvoker;
+  }
+
+  @Override
+  public Stream<NexusOperationExecutionMetadata> listNexusOperationExecutions(
+      @Nullable String query) {
+    // Pagination is handled inside the interceptor invoker; we receive a fully materialized list
+    // and expose a Stream view of it to honour the public API contract.
+    ListNexusOperationExecutionsOutput out =
+        nexusClientCallsInvoker.listNexusOperationExecutions(
+            new ListNexusOperationExecutionsInput(query));
+    return out.getOperations().stream().map(NexusOperationExecutionMetadata::fromListInfo);
+  }
+
+  @Override
+  public NexusOperationExecutionCount countNexusOperationExecutions(@Nullable String query) {
+    CountNexusOperationExecutionsOutput out =
+        nexusClientCallsInvoker.countNexusOperationExecutions(
+            new CountNexusOperationExecutionsInput(query));
+    List<NexusOperationExecutionCount.AggregationGroup> publicGroups =
+        out.getGroups().stream()
+            .map(
+                g ->
+                    new NexusOperationExecutionCount.AggregationGroup(
+                        g.getCount(), g.getGroupValues()))
+            .collect(Collectors.toList());
+    return new NexusOperationExecutionCount(out.getCount(), publicGroups);
+  }
+}