-
Notifications
You must be signed in to change notification settings - Fork 222
Add @TemporalOperation annotation for Nexus operations #2928
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
Quinn-With-Two-Ns
wants to merge
3
commits into
temporalio:main
Choose a base branch
from
Quinn-With-Two-Ns:NEXUS-415
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
+1,010
−257
Open
Changes from all commits
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
83 changes: 83 additions & 0 deletions
83
temporal-kotlin/src/test/kotlin/io/temporal/workflow/KotlinTemporalOperationTest.kt
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
|
|
||
| package io.temporal.workflow | ||
|
|
||
| import io.nexusrpc.Operation | ||
| import io.nexusrpc.Service | ||
| import io.nexusrpc.handler.ServiceImpl | ||
| import io.temporal.client.WorkflowClientOptions | ||
| import io.temporal.client.WorkflowOptions | ||
| import io.temporal.common.converter.DefaultDataConverter | ||
| import io.temporal.common.converter.JacksonJsonPayloadConverter | ||
| import io.temporal.common.converter.KotlinObjectMapperFactory | ||
| import io.temporal.nexus.TemporalNexusClient | ||
| import io.temporal.nexus.TemporalOperation | ||
| import io.temporal.nexus.TemporalOperationResult | ||
| import io.temporal.nexus.TemporalOperationStartContext | ||
| import io.temporal.testing.internal.SDKTestWorkflowRule | ||
| import org.junit.Assert.assertEquals | ||
| import org.junit.Rule | ||
| import org.junit.Test | ||
| import java.time.Duration | ||
|
|
||
| class KotlinTemporalOperationTest { | ||
|
|
||
| @Rule | ||
| @JvmField | ||
| var testWorkflowRule: SDKTestWorkflowRule = SDKTestWorkflowRule.newBuilder() | ||
| .setWorkflowTypes(CallerWorkflowImpl::class.java) | ||
| .setNexusServiceImplementation(KotlinSugarServiceImpl()) | ||
| .setWorkflowClientOptions( | ||
| WorkflowClientOptions.newBuilder() | ||
| .setDataConverter(DefaultDataConverter(JacksonJsonPayloadConverter(KotlinObjectMapperFactory.new()))) | ||
| .build() | ||
| ) | ||
| .build() | ||
|
|
||
| @Service | ||
| interface KotlinSugarService { | ||
| @Operation | ||
| fun greet(input: String): String | ||
| } | ||
|
|
||
| @ServiceImpl(service = KotlinSugarService::class) | ||
| class KotlinSugarServiceImpl { | ||
| @TemporalOperation | ||
| fun greet( | ||
| ctx: TemporalOperationStartContext, | ||
| client: TemporalNexusClient, | ||
| input: String | ||
| ): TemporalOperationResult<String> { | ||
| return TemporalOperationResult.sync("kotlin-$input") | ||
| } | ||
| } | ||
|
|
||
| @WorkflowInterface | ||
| interface CallerWorkflow { | ||
| @WorkflowMethod | ||
| fun execute(arg: String): String | ||
| } | ||
|
|
||
| class CallerWorkflowImpl : CallerWorkflow { | ||
| override fun execute(arg: String): String { | ||
| val stub = Workflow.newNexusServiceStub( | ||
| KotlinSugarService::class.java, | ||
| NexusServiceOptions { | ||
| setOperationOptions( | ||
| NexusOperationOptions { | ||
| setScheduleToCloseTimeout(Duration.ofSeconds(10)) | ||
| } | ||
| ) | ||
| } | ||
| ) | ||
| return stub.greet(arg) | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| fun temporalOperationSugar_endToEnd() { | ||
| val client = testWorkflowRule.workflowClient | ||
| val options = WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.taskQueue).build() | ||
| val workflowStub = client.newWorkflowStub(CallerWorkflow::class.java, options) | ||
| assertEquals("kotlin-hi", workflowStub.execute("hi")) | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
178 changes: 178 additions & 0 deletions
178
temporal-sdk/src/main/java/io/temporal/internal/nexus/TemporalOperationProcessor.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,178 @@ | ||
| package io.temporal.internal.nexus; | ||
|
|
||
| import com.google.common.collect.ImmutableList; | ||
| import com.google.common.primitives.Primitives; | ||
| import io.nexusrpc.OperationDefinition; | ||
| import io.nexusrpc.ServiceDefinition; | ||
| import io.nexusrpc.handler.MethodExtension; | ||
| import io.nexusrpc.handler.OperationImpl; | ||
| import io.nexusrpc.handler.ServiceImplInstance; | ||
| import io.temporal.nexus.TemporalNexusClient; | ||
| import io.temporal.nexus.TemporalOperation; | ||
| import io.temporal.nexus.TemporalOperationHandler; | ||
| import io.temporal.nexus.TemporalOperationResult; | ||
| import io.temporal.nexus.TemporalOperationStartContext; | ||
| import java.lang.invoke.MethodHandle; | ||
| import java.lang.invoke.MethodHandles; | ||
| import java.lang.reflect.Method; | ||
| import java.lang.reflect.Modifier; | ||
| import java.lang.reflect.ParameterizedType; | ||
| import java.lang.reflect.Type; | ||
| import java.util.Arrays; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * Entry point for registering a Nexus service instance whose class may contain {@link | ||
| * TemporalOperation}-annotated methods. Delegates to {@link | ||
| * ServiceImplInstance#fromInstance(Object, java.util.List)} with a single {@link MethodExtension} | ||
| * that recognizes {@link TemporalOperation} alongside the built-in {@link OperationImpl}. | ||
| */ | ||
| public final class TemporalOperationProcessor { | ||
|
|
||
| private static final ImmutableList<MethodExtension> EXTENSIONS = | ||
| ImmutableList.of(new TemporalOperationExtension()); | ||
|
|
||
| private TemporalOperationProcessor() {} | ||
|
|
||
| public static ServiceImplInstance process(Object instance) { | ||
| return ServiceImplInstance.fromInstance(instance, EXTENSIONS); | ||
| } | ||
|
|
||
| /** Recognizes {@link TemporalOperation}-annotated methods during nexusrpc service scanning. */ | ||
| private static final class TemporalOperationExtension implements MethodExtension { | ||
| @Override | ||
| public Result extract(Object instance, Method method, ServiceDefinition serviceDefinition) { | ||
| if (method.getDeclaredAnnotation(TemporalOperation.class) == null) { | ||
| return null; | ||
| } | ||
| if (method.isAnnotationPresent(OperationImpl.class)) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation and @OperationImpl cannot be combined on method " | ||
| + method.getName()); | ||
| } | ||
|
|
||
| validateSignature(method); | ||
|
|
||
| OperationDefinition operationDefinition = | ||
| serviceDefinition.getOperations().values().stream() | ||
| .filter(o -> method.getName().equals(o.getMethodName())) | ||
| .findFirst() | ||
| .orElseThrow( | ||
| () -> | ||
| new IllegalStateException( | ||
| "No matching @Operation on service " | ||
| + serviceDefinition.getName() | ||
| + " for @TemporalOperation method " | ||
| + method.getName())); | ||
|
|
||
| validateTypes(method, operationDefinition); | ||
|
|
||
| MethodHandle handle; | ||
| try { | ||
| handle = MethodHandles.lookup().unreflect(method).bindTo(instance); | ||
| } catch (IllegalAccessException e) { | ||
| throw new RuntimeException( | ||
| "Failed to obtain method handle for @TemporalOperation method " + method.getName(), e); | ||
| } | ||
|
|
||
| TemporalOperationHandler.StartHandler<Object, Object> startHandler = | ||
| (ctx, client, input) -> invokeStartHandler(handle, ctx, client, input); | ||
|
|
||
| return new Result( | ||
| operationDefinition.getName(), | ||
| new TemporalOperationHandler<Object, Object>(startHandler) {}); | ||
| } | ||
| } | ||
|
|
||
| private static void validateSignature(Method method) { | ||
| if (!Modifier.isPublic(method.getModifiers())) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation method " + method.getName() + " must be public"); | ||
| } | ||
| if (Modifier.isStatic(method.getModifiers())) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation method " + method.getName() + " must not be static"); | ||
| } | ||
| Class<?>[] paramTypes = method.getParameterTypes(); | ||
| if (paramTypes.length != 3 | ||
| || !TemporalOperationStartContext.class.equals(paramTypes[0]) | ||
| || !TemporalNexusClient.class.equals(paramTypes[1])) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation method " | ||
| + method.getName() | ||
| + " must accept (TemporalOperationStartContext, TemporalNexusClient, I); got " | ||
| + describeSignature(method)); | ||
| } | ||
| if (!TemporalOperationResult.class.equals(method.getReturnType())) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation method " | ||
| + method.getName() | ||
| + " must return TemporalOperationResult<?>; got " | ||
| + method.getGenericReturnType().getTypeName() | ||
| + ". Use @OperationImpl for custom handler shapes."); | ||
| } | ||
| } | ||
|
|
||
| private static void validateTypes(Method method, OperationDefinition operationDefinition) { | ||
| Type expectedInputType = operationDefinition.getInputType(); | ||
| Type declaredInputType = method.getGenericParameterTypes()[2]; | ||
| if (!typesMatch(declaredInputType, expectedInputType)) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation method " | ||
| + method.getName() | ||
| + " input type mismatch: expected " | ||
| + expectedInputType.getTypeName() | ||
| + " but got " | ||
| + declaredInputType.getTypeName()); | ||
| } | ||
| Type returnType = method.getGenericReturnType(); | ||
| if (!(returnType instanceof ParameterizedType)) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation method " | ||
| + method.getName() | ||
| + " must use parameterized TemporalOperationResult<R>, not the raw type."); | ||
| } | ||
| Type resultTypeArg = ((ParameterizedType) returnType).getActualTypeArguments()[0]; | ||
| if (!typesMatch(resultTypeArg, operationDefinition.getOutputType())) { | ||
| throw new IllegalArgumentException( | ||
| "@TemporalOperation method " | ||
| + method.getName() | ||
| + " output type mismatch: expected " | ||
| + operationDefinition.getOutputType().getTypeName() | ||
| + " but got " | ||
| + resultTypeArg.getTypeName()); | ||
| } | ||
| } | ||
|
|
||
| // Package-private for testing. | ||
| @SuppressWarnings("unchecked") | ||
| static TemporalOperationResult<Object> invokeStartHandler( | ||
| MethodHandle handle, | ||
| TemporalOperationStartContext ctx, | ||
| TemporalNexusClient client, | ||
| Object input) { | ||
| try { | ||
| return (TemporalOperationResult<Object>) handle.invoke(ctx, client, input); | ||
| } catch (RuntimeException | Error e) { | ||
| throw e; | ||
| } catch (Throwable t) { | ||
| throw new RuntimeException("@TemporalOperation method threw checked exception", t); | ||
| } | ||
| } | ||
|
|
||
| private static boolean typesMatch(Type declared, Type expected) { | ||
| if (declared.equals(expected)) { | ||
| return true; | ||
| } | ||
| if (declared instanceof Class && expected instanceof Class) { | ||
| return Primitives.wrap((Class<?>) declared).equals(Primitives.wrap((Class<?>) expected)); | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| private static String describeSignature(Method method) { | ||
| return Arrays.stream(method.getParameterTypes()) | ||
| .map(Class::getSimpleName) | ||
| .collect(Collectors.joining(", ", "(", ")")); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
53 changes: 53 additions & 0 deletions
53
temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperation.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,53 @@ | ||
| package io.temporal.nexus; | ||
|
|
||
| import io.nexusrpc.handler.OperationCancelDetails; | ||
| import io.nexusrpc.handler.OperationContext; | ||
| import io.nexusrpc.handler.OperationImpl; | ||
| import io.nexusrpc.handler.ServiceImpl; | ||
| import io.temporal.common.Experimental; | ||
| import java.lang.annotation.*; | ||
|
|
||
| /** | ||
| * Marks a method on a {@link ServiceImpl}-annotated class as a Temporal-backed Nexus operation. The | ||
| * method body <i>is</i> the start handler — the framework wraps it in a {@link | ||
| * TemporalOperationHandler} at registration time, with default cancel behavior matching {@link | ||
| * TemporalOperationHandler#cancel(OperationContext, OperationCancelDetails)}. | ||
| * | ||
| * <p>The method must: | ||
| * | ||
| * <ul> | ||
| * <li>be {@code public}, | ||
| * <li>accept exactly three parameters: {@link TemporalOperationStartContext}, {@link | ||
| * TemporalNexusClient}, and the operation input type, | ||
| * <li>return {@link TemporalOperationResult}. | ||
| * </ul> | ||
| * | ||
| * <p>Workflow-run example: | ||
| * | ||
| * <pre>{@code | ||
| * @ServiceImpl(service = TransferService.class) | ||
| * public class TransferServiceImpl { | ||
| * @TemporalOperation | ||
| * public TemporalOperationResult<TransferResult> transfer( | ||
| * TemporalOperationStartContext ctx, TemporalNexusClient client, TransferInput input) { | ||
| * return client.startWorkflow( | ||
| * TransferWorkflow.class, | ||
| * TransferWorkflow::transfer, | ||
| * input, | ||
| * WorkflowOptions.newBuilder() | ||
| * .setWorkflowId("transfer-" + input.getTransferId()) | ||
| * .build()); | ||
| * } | ||
| * } | ||
| * }</pre> | ||
| * | ||
| * <p>For custom cancel, or any other handler composition, use {@link OperationImpl} with a {@link | ||
| * TemporalOperationHandler} subclass that overrides {@link | ||
| * TemporalOperationHandler#cancelWorkflowRun}. Both annotations can coexist on the same {@link | ||
| * ServiceImpl} class, but never on the same method. | ||
| */ | ||
| @Experimental | ||
| @Documented | ||
| @Retention(RetentionPolicy.RUNTIME) | ||
| @Target(ElementType.METHOD) | ||
| public @interface TemporalOperation {} |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In a normal checkout that does not also have
../nexus/nexus-sdk-java, Gradle evaluates this settings file before any task and the build is blocked by the missing included build; Gradle documentsincludeBuildhere as adding another build to the composite, while the CLI--include-buildis the optional one-off form. Please don't commit a machine-local composite path; use the published dependency or make local Nexus development opt-in so./gradlew testworks from a clean sdk-java checkout.Useful? React with 👍 / 👎.