diff --git a/README.md b/README.md index b04d1f31e..a8af530f4 100644 --- a/README.md +++ b/README.md @@ -155,6 +155,8 @@ See the README.md file in each main sample directory for cut/paste Gradle comman - [**Getting Started**](/core/src/main/java/io/temporal/samples/nexus): Demonstrates how to get started with Temporal and Nexus. +- [**Mapping Multiple Arguments**](/core/src/main/java/io/temporal/samples/nexus): Demonstrates how map a Nexus operation to a Workflow that takes multiple arguments. + - [**Cancellation**](/core/src/main/java/io/temporal/samples/nexuscancellation): Demonstrates how to cancel an async Nexus operation. - [**Context/Header Propagation**](/core/src/main/java/io/temporal/samples/nexuscontextpropagation): Demonstrates how to propagate context through Nexus operation headers. diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/README.MD b/core/src/main/java/io/temporal/samples/nexusmultipleargs/README.MD new file mode 100644 index 000000000..754545091 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/README.MD @@ -0,0 +1,36 @@ +# Nexus Multiple Arguments Sample + +This sample shows how to map a Nexus operation to a caller workflow that takes multiple input arguments using [WorkflowRunOperation.fromWorkflowHandle](https://javadoc.io/doc/io.temporal/temporal-sdk/latest/io/temporal/nexus/WorkflowRunOperation.html#fromWorkflowHandle(io.temporal.nexus.WorkflowHandleFactory)). + +To run this sample, set up your environment following the instructions in the main [Nexus Sample](../nexus/README.md). + +In separate terminal windows: + +### Nexus handler worker + +``` +./gradlew -q execute -PmainClass=io.temporal.samples.nexusmultipleargs.handler.HandlerWorker \ + --args="-target-host localhost:7233 -namespace my-target-namespace" +``` + +### Nexus caller worker + +``` +./gradlew -q execute -PmainClass=io.temporal.samples.nexusmultipleargs.caller.CallerWorker \ + --args="-target-host localhost:7233 -namespace my-caller-namespace" +``` + +### Start caller workflow + +``` +./gradlew -q execute -PmainClass=io.temporal.samples.nexusmultipleargs.caller.CallerStarter \ + --args="-target-host localhost:7233 -namespace my-caller-namespace" +``` + +### Output + +which should result in: +``` +[main] INFO i.t.s.nexus.caller.CallerStarter - Workflow result: Nexus Echo πŸ‘‹ +[main] INFO i.t.s.nexus.caller.CallerStarter - Workflow result: Β‘Hola! Nexus πŸ‘‹ +``` diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/CallerStarter.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/CallerStarter.java new file mode 100644 index 000000000..beebdf406 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/CallerStarter.java @@ -0,0 +1,39 @@ +package io.temporal.samples.nexusmultipleargs.caller; + +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.samples.nexus.caller.CallerWorker; +import io.temporal.samples.nexus.caller.EchoCallerWorkflow; +import io.temporal.samples.nexus.caller.HelloCallerWorkflow; +import io.temporal.samples.nexus.options.ClientOptions; +import io.temporal.samples.nexus.service.NexusService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class CallerStarter { + private static final Logger logger = LoggerFactory.getLogger(CallerStarter.class); + + public static void main(String[] args) { + WorkflowClient client = ClientOptions.getWorkflowClient(args); + + WorkflowOptions workflowOptions = + WorkflowOptions.newBuilder().setTaskQueue(CallerWorker.DEFAULT_TASK_QUEUE_NAME).build(); + EchoCallerWorkflow echoWorkflow = + client.newWorkflowStub(EchoCallerWorkflow.class, workflowOptions); + WorkflowExecution execution = WorkflowClient.start(echoWorkflow::echo, "Nexus Echo πŸ‘‹"); + logger.info( + "Started EchoCallerWorkflow workflowId: {} runId: {}", + execution.getWorkflowId(), + execution.getRunId()); + logger.info("Workflow result: {}", echoWorkflow.echo("Nexus Echo πŸ‘‹")); + HelloCallerWorkflow helloWorkflow = + client.newWorkflowStub(HelloCallerWorkflow.class, workflowOptions); + execution = WorkflowClient.start(helloWorkflow::hello, "Nexus", NexusService.Language.EN); + logger.info( + "Started HelloCallerWorkflow workflowId: {} runId: {}", + execution.getWorkflowId(), + execution.getRunId()); + logger.info("Workflow result: {}", helloWorkflow.hello("Nexus", NexusService.Language.ES)); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/CallerWorker.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/CallerWorker.java new file mode 100644 index 000000000..1db03a677 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/CallerWorker.java @@ -0,0 +1,32 @@ +package io.temporal.samples.nexusmultipleargs.caller; + +import io.temporal.client.WorkflowClient; +import io.temporal.samples.nexus.options.ClientOptions; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import io.temporal.worker.WorkflowImplementationOptions; +import io.temporal.workflow.NexusServiceOptions; +import java.util.Collections; + +public class CallerWorker { + public static final String DEFAULT_TASK_QUEUE_NAME = "my-caller-workflow-task-queue"; + + public static void main(String[] args) { + WorkflowClient client = ClientOptions.getWorkflowClient(args); + + WorkerFactory factory = WorkerFactory.newInstance(client); + + Worker worker = factory.newWorker(DEFAULT_TASK_QUEUE_NAME); + worker.registerWorkflowImplementationTypes( + WorkflowImplementationOptions.newBuilder() + .setNexusServiceOptions( + Collections.singletonMap( + "NexusService", + NexusServiceOptions.newBuilder().setEndpoint("my-nexus-endpoint-name").build())) + .build(), + EchoCallerWorkflowImpl.class, + HelloCallerWorkflowImpl.class); + + factory.start(); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/EchoCallerWorkflow.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/EchoCallerWorkflow.java new file mode 100644 index 000000000..47f1adce2 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/EchoCallerWorkflow.java @@ -0,0 +1,10 @@ +package io.temporal.samples.nexusmultipleargs.caller; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface EchoCallerWorkflow { + @WorkflowMethod + String echo(String message); +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/EchoCallerWorkflowImpl.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/EchoCallerWorkflowImpl.java new file mode 100644 index 000000000..ea9aaabb0 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/EchoCallerWorkflowImpl.java @@ -0,0 +1,25 @@ +package io.temporal.samples.nexusmultipleargs.caller; + +import io.temporal.samples.nexus.caller.EchoCallerWorkflow; +import io.temporal.samples.nexus.service.NexusService; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +public class EchoCallerWorkflowImpl implements EchoCallerWorkflow { + NexusService nexusService = + Workflow.newNexusServiceStub( + NexusService.class, + NexusServiceOptions.newBuilder() + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + .build()) + .build()); + + @Override + public String echo(String message) { + return nexusService.echo(new NexusService.EchoInput(message)).getMessage(); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/HelloCallerWorkflow.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/HelloCallerWorkflow.java new file mode 100644 index 000000000..aeb7a13db --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/HelloCallerWorkflow.java @@ -0,0 +1,11 @@ +package io.temporal.samples.nexusmultipleargs.caller; + +import io.temporal.samples.nexus.service.NexusService; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface HelloCallerWorkflow { + @WorkflowMethod + String hello(String message, NexusService.Language language); +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/HelloCallerWorkflowImpl.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/HelloCallerWorkflowImpl.java new file mode 100644 index 000000000..94f4c3fc5 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/caller/HelloCallerWorkflowImpl.java @@ -0,0 +1,32 @@ +package io.temporal.samples.nexusmultipleargs.caller; + +import io.temporal.samples.nexus.caller.HelloCallerWorkflow; +import io.temporal.samples.nexus.service.NexusService; +import io.temporal.workflow.NexusOperationHandle; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +public class HelloCallerWorkflowImpl implements HelloCallerWorkflow { + NexusService nexusService = + Workflow.newNexusServiceStub( + NexusService.class, + NexusServiceOptions.newBuilder() + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + .build()) + .build()); + + @Override + public String hello(String message, NexusService.Language language) { + NexusOperationHandle handle = + Workflow.startNexusOperation( + nexusService::hello, new NexusService.HelloInput(message, language)); + // Optionally wait for the operation to be started. NexusOperationExecution will contain the + // operation token in case this operation is asynchronous. + handle.getExecution().get(); + return handle.getResult().get().getMessage(); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HandlerWorker.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HandlerWorker.java new file mode 100644 index 000000000..0ec77889c --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HandlerWorker.java @@ -0,0 +1,22 @@ +package io.temporal.samples.nexusmultipleargs.handler; + +import io.temporal.client.WorkflowClient; +import io.temporal.samples.nexus.options.ClientOptions; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; + +public class HandlerWorker { + public static final String DEFAULT_TASK_QUEUE_NAME = "my-handler-task-queue"; + + public static void main(String[] args) { + WorkflowClient client = ClientOptions.getWorkflowClient(args); + + WorkerFactory factory = WorkerFactory.newInstance(client); + + Worker worker = factory.newWorker(DEFAULT_TASK_QUEUE_NAME); + worker.registerWorkflowImplementationTypes(HelloHandlerWorkflowImpl.class); + worker.registerNexusServiceImplementation(new NexusServiceImpl()); + + factory.start(); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HelloHandlerWorkflow.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HelloHandlerWorkflow.java new file mode 100644 index 000000000..0a443d468 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HelloHandlerWorkflow.java @@ -0,0 +1,11 @@ +package io.temporal.samples.nexusmultipleargs.handler; + +import io.temporal.samples.nexus.service.NexusService; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface HelloHandlerWorkflow { + @WorkflowMethod + NexusService.HelloOutput hello(String name, NexusService.Language language); +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HelloHandlerWorkflowImpl.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HelloHandlerWorkflowImpl.java new file mode 100644 index 000000000..b802cee34 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/HelloHandlerWorkflowImpl.java @@ -0,0 +1,24 @@ +package io.temporal.samples.nexusmultipleargs.handler; + +import io.temporal.failure.ApplicationFailure; +import io.temporal.samples.nexus.service.NexusService; + +public class HelloHandlerWorkflowImpl implements HelloHandlerWorkflow { + @Override + public NexusService.HelloOutput hello(String name, NexusService.Language language) { + switch (language) { + case EN: + return new NexusService.HelloOutput("Hello " + name + " πŸ‘‹"); + case FR: + return new NexusService.HelloOutput("Bonjour " + name + " πŸ‘‹"); + case DE: + return new NexusService.HelloOutput("Hallo " + name + " πŸ‘‹"); + case ES: + return new NexusService.HelloOutput("Β‘Hola! " + name + " πŸ‘‹"); + case TR: + return new NexusService.HelloOutput("Merhaba " + name + " πŸ‘‹"); + } + throw ApplicationFailure.newFailure( + "Unsupported language: " + language, "UNSUPPORTED_LANGUAGE"); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/NexusServiceImpl.java b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/NexusServiceImpl.java new file mode 100644 index 000000000..20d55fed8 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusmultipleargs/handler/NexusServiceImpl.java @@ -0,0 +1,60 @@ +package io.temporal.samples.nexusmultipleargs.handler; + +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.client.WorkflowOptions; +import io.temporal.nexus.Nexus; +import io.temporal.nexus.WorkflowHandle; +import io.temporal.nexus.WorkflowRunOperation; +import io.temporal.samples.nexus.service.NexusService; + +// To create a service implementation, annotate the class with @ServiceImpl and provide the +// interface that the service implements. The service implementation class should have methods that +// return OperationHandler that correspond to the operations defined in the service interface. +@ServiceImpl(service = NexusService.class) +public class NexusServiceImpl { + @OperationImpl + public OperationHandler echo() { + // OperationHandler.sync is a meant for exposing simple RPC handlers. + return OperationHandler.sync( + // The method is for making arbitrary short calls to other services or databases, or + // perform simple computations such as this one. Users can also access a workflow client by + // calling + // Nexus.getOperationContext().getWorkflowClient(ctx) to make arbitrary calls such as + // signaling, querying, or listing workflows. + (ctx, details, input) -> new NexusService.EchoOutput(input.getMessage())); + } + + @OperationImpl + public OperationHandler hello() { + // If the operation input parameters are different from the workflow input parameters, + // use the WorkflowRunOperation.fromWorkflowHandler constructor and the appropriate constructor + // method on WorkflowHandle to map the Nexus input to the workflow parameters. + return WorkflowRunOperation.fromWorkflowHandle( + (ctx, details, input) -> + WorkflowHandle.fromWorkflowMethod( + Nexus.getOperationContext() + .getWorkflowClient() + .newWorkflowStub( + HelloHandlerWorkflow.class, + // Workflow IDs should typically be business meaningful IDs and are used + // to + // dedupe workflow starts. + // For this example, we're using the request ID allocated by Temporal + // when + // the + // caller workflow schedules + // the operation, this ID is guaranteed to be stable across retries of + // this + // operation. + // + // Task queue defaults to the task queue this operation is handled on. + WorkflowOptions.newBuilder() + .setWorkflowId(details.getRequestId()) + .build()) + ::hello, + input.getName(), + input.getLanguage())); + } +}