diff --git a/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/README.md b/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/README.md index b369a95f..56949142 100644 --- a/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/README.md +++ b/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/README.md @@ -15,12 +15,18 @@ The caller Workflow: ### Running -Start a Temporal server: +Start a Temporal server with Workflow Update callbacks enabled: ```bash -temporal server start-dev +temporal server start-dev \ + --dynamic-config-value history.enableUpdateCallbacks=true \ + --dynamic-config-value history.enableCHASMSignalBacklinks=true ``` +`history.enableUpdateCallbacks` defaults to `false`. The `setLanguage` operation starts its Update +through the Nexus client, so its result is delivered over a Nexus completion callback; without this +flag the callback is never registered and the operation never completes. + Create the namespaces and Nexus endpoint: ```bash diff --git a/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/handler/NexusGreetingServiceImpl.java b/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/handler/NexusGreetingServiceImpl.java index 2a6d04c6..74bc763f 100644 --- a/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/handler/NexusGreetingServiceImpl.java +++ b/core/src/main/java/io/temporal/samples/nexusmessaging/callerpattern/handler/NexusGreetingServiceImpl.java @@ -3,7 +3,11 @@ import io.nexusrpc.handler.OperationHandler; import io.nexusrpc.handler.OperationImpl; import io.nexusrpc.handler.ServiceImpl; -import io.temporal.nexus.Nexus; +import io.temporal.client.UpdateOptions; +import io.temporal.client.WorkflowUpdateStage; +import io.temporal.nexus.TemporalNexusClient; +import io.temporal.nexus.TemporalOperationHandler; +import io.temporal.nexus.TemporalOperationResult; import io.temporal.samples.nexusmessaging.callerpattern.service.Language; import io.temporal.samples.nexusmessaging.callerpattern.service.NexusGreetingService; import org.slf4j.Logger; @@ -11,8 +15,12 @@ /** * Nexus operation handler implementation. Each operation receives a userId, which is mapped to a - * workflow ID using {@link #WORKFLOW_ID_PREFIX}. The operations are synchronous because queries and - * updates against a running workflow complete quickly. + * workflow ID using {@link #WORKFLOW_ID_PREFIX}. + * + *

Every operation is a {@link TemporalOperationHandler}: the start handler receives a {@link + * TemporalNexusClient} scoped to the invocation and returns a {@link TemporalOperationResult}. The + * queries and the signal complete inline and return a sync result; {@code setLanguage} starts the + * update through the Nexus client, which makes it an asynchronous operation. */ @ServiceImpl(service = NexusGreetingService.class) public class NexusGreetingServiceImpl { @@ -29,8 +37,8 @@ public static String getWorkflowId(String userId) { return WORKFLOW_ID_PREFIX + userId; } - private GreetingWorkflow getWorkflowStub(String userId) { - return Nexus.getOperationContext() + private GreetingWorkflow getWorkflowStub(TemporalNexusClient client, String userId) { + return client .getWorkflowClient() .newWorkflowStub(GreetingWorkflow.class, getWorkflowId(userId)); } @@ -39,41 +47,56 @@ private GreetingWorkflow getWorkflowStub(String userId) { public OperationHandler< NexusGreetingService.GetLanguagesInput, NexusGreetingService.GetLanguagesOutput> getLanguages() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Query for GetLanguages was received for user {}", input.getUserId()); - return getWorkflowStub(input.getUserId()).getLanguages(input); + return TemporalOperationResult.sync( + getWorkflowStub(client, input.getUserId()).getLanguages(input)); }); } @OperationImpl public OperationHandler getLanguage() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Query for GetLanguage was received for user {}", input.getUserId()); - return getWorkflowStub(input.getUserId()).getLanguage(); + return TemporalOperationResult.sync( + getWorkflowStub(client, input.getUserId()).getLanguage()); }); } // Routes to setLanguageUsingActivity (not setLanguage) so that new languages not already in the // greetings map can be fetched via an activity. + // + // Starting the update through the Nexus client makes this an asynchronous Nexus operation: the + // caller receives an operation token and the update result is delivered later over the Nexus + // completion callback. If the update is already complete when the start call returns, the result + // comes back synchronously instead. @OperationImpl public OperationHandler setLanguage() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Update for SetLanguage was received for user {}", input.getUserId()); - return getWorkflowStub(input.getUserId()).setLanguageUsingActivity(input); + return client.startWorkflowUpdate( + GreetingWorkflow.class, + getWorkflowId(input.getUserId()), + GreetingWorkflow::setLanguageUsingActivity, + input, + UpdateOptions.newBuilder(Language.class) + .setUpdateName("setLanguageUsingActivity") + .setWaitForStage(WorkflowUpdateStage.ACCEPTED) + .build()); }); } @OperationImpl public OperationHandler approve() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Signal for Approve was received for user {}", input.getUserId()); - getWorkflowStub(input.getUserId()).approve(input); - return new NexusGreetingService.ApproveOutput(); + getWorkflowStub(client, input.getUserId()).approve(input); + return TemporalOperationResult.sync(new NexusGreetingService.ApproveOutput()); }); } } diff --git a/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/README.md b/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/README.md index c987e85f..cd734102 100644 --- a/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/README.md +++ b/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/README.md @@ -15,12 +15,18 @@ The caller Workflow: ### Running -Start a Temporal server: +Start a Temporal server with Workflow Update callbacks enabled: ```bash -temporal server start-dev +temporal server start-dev \ + --dynamic-config-value history.enableUpdateCallbacks=true \ + --dynamic-config-value history.enableCHASMSignalBacklinks=true ``` +`history.enableUpdateCallbacks` defaults to `false`. The `setLanguage` operation starts its Update +through the Nexus client, so its result is delivered over a Nexus completion callback; without this +flag the callback is never registered and the operation never completes. + Create the namespaces and Nexus endpoint: ```bash diff --git a/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/handler/NexusRemoteGreetingServiceImpl.java b/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/handler/NexusRemoteGreetingServiceImpl.java index b7397987..b373959a 100644 --- a/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/handler/NexusRemoteGreetingServiceImpl.java +++ b/core/src/main/java/io/temporal/samples/nexusmessaging/ondemandpattern/handler/NexusRemoteGreetingServiceImpl.java @@ -3,8 +3,13 @@ import io.nexusrpc.handler.OperationHandler; import io.nexusrpc.handler.OperationImpl; import io.nexusrpc.handler.ServiceImpl; +import io.temporal.client.UpdateOptions; import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowUpdateStage; import io.temporal.nexus.Nexus; +import io.temporal.nexus.TemporalNexusClient; +import io.temporal.nexus.TemporalOperationHandler; +import io.temporal.nexus.TemporalOperationResult; import io.temporal.nexus.WorkflowHandle; import io.temporal.nexus.WorkflowRunOperation; import io.temporal.samples.nexusmessaging.ondemandpattern.service.Language; @@ -15,6 +20,12 @@ /** * Nexus operation handler for the on-demand pattern. Each operation receives the target workflow ID * in its input, and {@code runFromRemote} starts a brand-new GreetingWorkflow. + * + *

The messaging operations are {@link TemporalOperationHandler}s: the start handler receives a + * {@link TemporalNexusClient} scoped to the invocation and returns a {@link + * TemporalOperationResult}. The queries and the signal complete inline and return a sync result; + * {@code setLanguage} starts the update through the Nexus client, which makes it an asynchronous + * operation. */ @ServiceImpl(service = NexusRemoteGreetingService.class) public class NexusRemoteGreetingServiceImpl { @@ -32,8 +43,8 @@ public static String getWorkflowId(String userId) { return WORKFLOW_ID_PREFIX + userId; } - private GreetingWorkflow getWorkflowStub(String userId) { - return Nexus.getOperationContext() + private GreetingWorkflow getWorkflowStub(TemporalNexusClient client, String userId) { + return client .getWorkflowClient() .newWorkflowStub(GreetingWorkflow.class, getWorkflowId(userId)); } @@ -63,31 +74,46 @@ public OperationHandler r NexusRemoteGreetingService.GetLanguagesInput, NexusRemoteGreetingService.GetLanguagesOutput> getLanguages() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Query for GetLanguages was received for userId {}", input.getUserId()); - return getWorkflowStub(input.getUserId()) - .getLanguages(new GreetingWorkflow.GetLanguagesInput(input.isIncludeUnsupported())); + return TemporalOperationResult.sync( + getWorkflowStub(client, input.getUserId()) + .getLanguages( + new GreetingWorkflow.GetLanguagesInput(input.isIncludeUnsupported()))); }); } @OperationImpl public OperationHandler getLanguage() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Query for GetLanguage was received for userId {}", input.getUserId()); - return getWorkflowStub(input.getUserId()).getLanguage(); + return TemporalOperationResult.sync( + getWorkflowStub(client, input.getUserId()).getLanguage()); }); } // Uses setLanguageUsingActivity so that new languages are fetched via an activity. + // + // Starting the update through the Nexus client makes this an asynchronous Nexus operation: the + // caller receives an operation token and the update result is delivered later over the Nexus + // completion callback. If the update is already complete when the start call returns, the result + // comes back synchronously instead. @OperationImpl public OperationHandler setLanguage() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Update for SetLanguage was received for userId {}", input.getUserId()); - return getWorkflowStub(input.getUserId()) - .setLanguageUsingActivity(new GreetingWorkflow.SetLanguageInput(input.getLanguage())); + return client.startWorkflowUpdate( + GreetingWorkflow.class, + getWorkflowId(input.getUserId()), + GreetingWorkflow::setLanguageUsingActivity, + new GreetingWorkflow.SetLanguageInput(input.getLanguage()), + UpdateOptions.newBuilder(Language.class) + .setUpdateName("setLanguageUsingActivity") + .setWaitForStage(WorkflowUpdateStage.ACCEPTED) + .build()); }); } @@ -95,12 +121,12 @@ public OperationHandler s public OperationHandler< NexusRemoteGreetingService.ApproveInput, NexusRemoteGreetingService.ApproveOutput> approve() { - return OperationHandler.sync( - (ctx, details, input) -> { + return TemporalOperationHandler.create( + (ctx, client, input) -> { logger.info("Signal for Approve was received for userId {}", input.getUserId()); - getWorkflowStub(input.getUserId()) + getWorkflowStub(client, input.getUserId()) .approve(new GreetingWorkflow.ApproveInput(input.getName())); - return new NexusRemoteGreetingService.ApproveOutput(); + return TemporalOperationResult.sync(new NexusRemoteGreetingService.ApproveOutput()); }); } } diff --git a/core/src/main/java/io/temporal/samples/nexusstandalone/README.MD b/core/src/main/java/io/temporal/samples/nexusstandalone/README.MD index eee6d684..a1de93a7 100644 --- a/core/src/main/java/io/temporal/samples/nexusstandalone/README.MD +++ b/core/src/main/java/io/temporal/samples/nexusstandalone/README.MD @@ -30,6 +30,7 @@ from the environment (and optionally a profile from `temporal.toml`). ```bash ./temporal server start-dev \ + --dynamic-config-value nexusoperation.enableStandalone=true \ --namespace my-caller-namespace \ --namespace my-handler-namespace ``` diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/README.md b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/README.md index fb84bd6a..2d9f23a2 100644 --- a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/README.md +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/README.md @@ -33,6 +33,7 @@ from the environment (and optionally a profile from `temporal.toml`). ```bash ./temporal server start-dev \ + --dynamic-config-value nexusoperation.enableStandalone=true \ --dynamic-config-value activity.enableCallbacks=true \ --namespace my-caller-namespace \ --namespace my-handler-namespace