Progress checkin

Author: Evanthx

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

@@ -2,28 +2,19 @@
 
 import io.temporal.client.NexusClientInterceptor.CountNexusOperationExecutionsInput;
 import io.temporal.client.NexusClientInterceptor.CountNexusOperationExecutionsOutput;
-import io.temporal.client.NexusClientInterceptor.DeleteNexusOperationExecutionInput;
-import io.temporal.client.NexusClientInterceptor.DescribeNexusOperationExecutionInput;
-import io.temporal.client.NexusClientInterceptor.DescribeNexusOperationExecutionOutput;
 import io.temporal.client.NexusClientInterceptor.ListNexusOperationExecutionsInput;
 import io.temporal.client.NexusClientInterceptor.ListNexusOperationExecutionsOutput;
-import io.temporal.client.NexusClientInterceptor.PollNexusOperationExecutionInput;
-import io.temporal.client.NexusClientInterceptor.PollNexusOperationExecutionOutput;
-import io.temporal.client.NexusClientInterceptor.RequestCancelNexusOperationExecutionInput;
 import io.temporal.client.NexusClientInterceptor.StartNexusOperationExecutionInput;
 import io.temporal.client.NexusClientInterceptor.StartNexusOperationExecutionOutput;
-import io.temporal.client.NexusClientInterceptor.TerminateNexusOperationExecutionInput;
 import io.temporal.common.Experimental;
 import io.temporal.serviceclient.WorkflowServiceStubs;
-import java.lang.reflect.Type;
-import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nullable;
 
 /**
- * Handle for interacting with a standalone Nexus operation execution.
+ * Client for managing standalone Nexus operation executions.
  *
- * <p>Returned by {@link WorkflowClient} when starting a Nexus operation, and also constructable
- * from an existing operation ID for operating on an operation that was started elsewhere.
+ * <p>Per-operation actions (describe, cancel, terminate, delete, get result) live on {@link
+ * NexusClientHandle}; obtain a handle via {@link #getHandle}.
  */
 @Experimental
 public interface NexusClient {
@@ -36,49 +27,25 @@ static NexusClient newInstance(
     return NexusClientImpl.newInstance(service, options);
   }
 
-  NexusClientHandle getHandle(String scheduleID);
+  /** Obtain a handle to an existing operation; targets the latest run. */
+  NexusClientHandle getHandle(String operationId);
 
-  UntypedNexusClientHandle getHandle(String operationId, @Nullable String operationRunId);
-
-  /** Obtains typed handle to existing operations. */
-  <R> NexusClientHandle<R> getHandle(
-      String operationId, @Nullable String operationRunId, Class<R> resultClass);
-
-  /** Obtains typed handle to existing operations. For use with generic return types. */
-  <R> NexusClientHandle<R> getHandle(
-      String operationId,
-      @Nullable String operationRunId,
-      Class<R> resultClass,
-      @Nullable Type resultType);
+  /** Obtain a handle to an existing operation, optionally pinned to a specific run. */
+  NexusClientHandle getHandle(String operationId, @Nullable String runId);
 
   UntypedNexusServiceClient newUntypeNexusServiceClient();
 
   <R> NexusServiceClient<R> newNexusServiceClient();
 
+  /** Start a new standalone Nexus operation execution. */
   StartNexusOperationExecutionOutput startNexusOperationExecution(
       StartNexusOperationExecutionInput input);
 
-  DescribeNexusOperationExecutionOutput describeNexusOperationExecution(
-      DescribeNexusOperationExecutionInput input);
-
-  CompletableFuture<DescribeNexusOperationExecutionOutput> describeNexusOperationExecutionAsync(
-      DescribeNexusOperationExecutionInput input);
-
-  PollNexusOperationExecutionOutput pollNexusOperationExecution(
-      PollNexusOperationExecutionInput input);
-
-  CompletableFuture<PollNexusOperationExecutionOutput> pollNexusOperationExecutionAsync(
-      PollNexusOperationExecutionInput input);
-
+  /** List standalone Nexus operation executions matching a query. */
   ListNexusOperationExecutionsOutput listNexusOperationExecutions(
       ListNexusOperationExecutionsInput input);
 
+  /** Count standalone Nexus operation executions matching a query. */
   CountNexusOperationExecutionsOutput countNexusOperationExecutions(
       CountNexusOperationExecutionsInput input);
-
-  void requestCancelNexusOperationExecution(RequestCancelNexusOperationExecutionInput input);
-
-  void terminateNexusOperationExecution(TerminateNexusOperationExecutionInput input);
-
-  void deleteNexusOperationExecution(DeleteNexusOperationExecutionInput input);
 }

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

@@ -4,14 +4,39 @@
 import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nullable;
 
-public interface NexusClientHandle<R> extends UntypedNexusClientHandle {
-  public <R> NexusClientHandle<R> fromUntyped(
-      UntypedNexusClientHandle handle, Class<R> resultClass);
+/**
+ * Handle for interacting with an existing standalone Nexus operation execution. Returned by {@link
+ * NexusClient#getHandle(String)} (and overloads).
+ */
+public interface NexusClientHandle {
+  /** Operation ID this handle was constructed for. Always non-null. */
+  String getNexusOperationId();
 
-  public <R> NexusClientHandle<R> fromUntyped(
-      UntypedNexusClientHandle handle, Class<R> resultClass, @Nullable Type resultType);
+  /**
+   * Present if the handle was returned by start or set when calling {@link
+   * NexusClient#getHandle(String, String)}. Null if no run ID was provided — call {@link
+   * #describe()} to learn the current run ID.
+   */
+  @Nullable
+  String getNexusOperationRunId();
 
-  public R getResult();
+  NexusClientOperationExecutionDescription describe();
 
-  public CompletableFuture<R> getResultAsync();
+  void cancel();
+
+  void cancel(@Nullable String reason);
+
+  void terminate();
+
+  void terminate(@Nullable String reason);
+
+  void delete();
+
+  <R> R getResult(Class<R> resultClass);
+
+  <R> R getResult(Class<R> resultClass, @Nullable Type resultType);
+
+  <R> CompletableFuture<R> getResultAsync(Class<R> resultClass);
+
+  <R> CompletableFuture<R> getResultAsync(Class<R> resultClass, @Nullable Type resultType);
 }

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

@@ -1,74 +1,117 @@
 package io.temporal.client;
 
+import io.grpc.Deadline;
+import io.temporal.client.NexusClientInterceptor.DeleteNexusOperationExecutionInput;
+import io.temporal.client.NexusClientInterceptor.DescribeNexusOperationExecutionInput;
+import io.temporal.client.NexusClientInterceptor.DescribeNexusOperationExecutionOutput;
+import io.temporal.client.NexusClientInterceptor.RequestCancelNexusOperationExecutionInput;
+import io.temporal.client.NexusClientInterceptor.TerminateNexusOperationExecutionInput;
 import java.lang.reflect.Type;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 
-public class NexusClientHandleImpl<R> implements NexusClientHandle<R> {
+public class NexusClientHandleImpl implements NexusClientHandle {
 
-  private final NexusClientInterceptor interceptor;
+  // Default deadline applied to per-handle RPCs that need one (describe). Long-poll callers
+  // should reach for a typed result API once it exists, which can take a caller-supplied
+  // deadline.
+  private static final long DEFAULT_DEADLINE_SECONDS = 30;
 
-  public NexusClientHandleImpl(NexusClientInterceptor interceptor) {
+  private final NexusClientInterceptor interceptor;
+  private final String operationId;
+  private final @Nullable String runId;
+
+  public NexusClientHandleImpl(
+      NexusClientInterceptor interceptor, String operationId, @Nullable String runId) {
+    if (interceptor == null) {
+      throw new IllegalArgumentException("interceptor is required");
+    }
+    if (operationId == null) {
+      throw new IllegalArgumentException("operationId is required");
+    }
     this.interceptor = interceptor;
+    this.operationId = operationId;
+    this.runId = runId;
   }
 
-  public <T> NexusClientHandle<T> fromUntyped(
-      UntypedNexusClientHandle handle, Class<T> resultClass) {
-    return null;
-  }
-
-  public <T> NexusClientHandle<T> fromUntyped(
-      UntypedNexusClientHandle handle, Class<T> resultClass, @Nullable Type resultType) {
-    return null;
-  }
-
-  public R getResult() {
-    return null;
+  @Override
+  public String getNexusOperationId() {
+    return operationId;
   }
 
-  public CompletableFuture<R> getResultAsync() {
-    return null;
+  @Override
+  public @Nullable String getNexusOperationRunId() {
+    return runId;
   }
 
   @Override
-  public @Nullable String getNexusOperationRunId() {
-    return "";
+  public NexusClientOperationExecutionDescription describe() {
+    DescribeNexusOperationExecutionInput input =
+        new DescribeNexusOperationExecutionInput(
+            operationId,
+            runId,
+            /* includeInput= */ false,
+            /* includeOutcome= */ true,
+            Deadline.after(DEFAULT_DEADLINE_SECONDS, TimeUnit.SECONDS));
+    DescribeNexusOperationExecutionOutput output =
+        interceptor.describeNexusOperationExecution(input);
+    return output.getDescription();
   }
 
   @Override
-  public <R> R getResult(Class<R> resultClass) {
-    return null;
+  public void cancel() {
+    cancel(null);
   }
 
   @Override
-  public <R> R getResult(Class<R> resultClass, @Nullable Type resultType) {
-    return null;
+  public void cancel(@Nullable String reason) {
+    interceptor.requestCancelNexusOperationExecution(
+        new RequestCancelNexusOperationExecutionInput(operationId, runId, reason));
   }
 
   @Override
-  public <R> CompletableFuture<R> getResultAsync(Class<R> resultClass) {
-    return null;
+  public void terminate() {
+    terminate(null);
   }
 
   @Override
-  public <R> CompletableFuture<R> getResultAsync(Class<R> resultClass, @Nullable Type resultType) {
-    return null;
+  public void terminate(@Nullable String reason) {
+    interceptor.terminateNexusOperationExecution(
+        new TerminateNexusOperationExecutionInput(operationId, runId, reason));
   }
 
   @Override
-  public NexusClientOperationExecutionDescription describe() {
-    return null;
+  public void delete() {
+    interceptor.deleteNexusOperationExecution(
+        new DeleteNexusOperationExecutionInput(operationId, runId));
   }
 
   @Override
-  public void cancel() {}
+  public <R> R getResult(Class<R> resultClass) {
+    throw new UnsupportedOperationException(
+        "getResult is not yet implemented — pending DataConverter support on NexusClientOperationOptions"
+            + " and a poll-until-completion strategy");
+  }
 
   @Override
-  public void cancel(@Nullable String reason) {}
+  public <R> R getResult(Class<R> resultClass, @Nullable Type resultType) {
+    throw new UnsupportedOperationException(
+        "getResult is not yet implemented — pending DataConverter support on NexusClientOperationOptions"
+            + " and a poll-until-completion strategy");
+  }
 
   @Override
-  public void terminate() {}
+  public <R> CompletableFuture<R> getResultAsync(Class<R> resultClass) {
+    throw new UnsupportedOperationException(
+        "getResultAsync is not yet implemented — pending DataConverter support on NexusClientOperationOptions"
+            + " and a poll-until-completion strategy");
+  }
 
   @Override
-  public void terminate(@Nullable String reason) {}
+  public <R> CompletableFuture<R> getResultAsync(Class<R> resultClass, @Nullable Type resultType) {
+    throw new UnsupportedOperationException(
+        "getResultAsync is not yet implemented — pending DataConverter support on NexusClientOperationOptions"
+            + " and a poll-until-completion strategy");
+  }
 }