diff --git a/README.md b/README.md index 898cc5b1..284e4c32 100644 --- a/README.md +++ b/README.md @@ -174,6 +174,8 @@ Load client configuration from TOML files with programmatic overrides. - [**Standalone Nexus Operations**](/core/src/main/java/io/temporal/samples/nexusstandalone): Demonstrates how to start Standalone Nexus Operations — Nexus Operations that run independently without a Workflow. +- [**Nexus Standalone Activity**](/core/src/main/java/io/temporal/samples/nexusstandaloneactivity): Demonstrates how to back a Nexus Operation with a Standalone Activity. + - [**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. diff --git a/build.gradle b/build.gradle index b0ef3d59..7d6a9d0d 100644 --- a/build.gradle +++ b/build.gradle @@ -21,7 +21,7 @@ subprojects { ext { otelVersion = '1.30.1' otelVersionAlpha = "${otelVersion}-alpha" - javaSDKVersion = '1.37.0' + javaSDKVersion = '1.38.0' camelVersion = '3.22.1' jarVersion = '1.0.0' } diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/ClientStarter.java b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/ClientStarter.java new file mode 100644 index 00000000..62e0a8d2 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/ClientStarter.java @@ -0,0 +1,51 @@ +package io.temporal.samples.nexusstandaloneactivity; + +import io.temporal.client.NexusClient; +import io.temporal.client.NexusClientOptions; +import io.temporal.client.NexusServiceClient; +import io.temporal.client.StartNexusOperationOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.samples.nexusstandaloneactivity.service.ClientOptions; +import io.temporal.samples.nexusstandaloneactivity.service.GreetingNexusService; +import io.temporal.samples.nexusstandaloneactivity.service.GreetingNexusService.GreetingInput; +import io.temporal.samples.nexusstandaloneactivity.service.GreetingNexusService.GreetingOutput; +import io.temporal.serviceclient.WorkflowServiceStubs; +import java.time.Duration; +import java.util.UUID; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +// Executes the Activity-backed Nexus operation from client code. The operation is standalone: it is +// started directly by this client rather than from within a caller Workflow. +public class ClientStarter { + private static final Logger logger = LoggerFactory.getLogger(ClientStarter.class); + + // Must match the Nexus endpoint configured on the server (see README). + public static final String ENDPOINT_NAME = "my-nexus-endpoint"; + + public static void main(String[] args) { + WorkflowClient client = ClientOptions.getWorkflowClient(); + WorkflowServiceStubs stubs = client.getWorkflowServiceStubs(); + String namespace = client.getOptions().getNamespace(); + + NexusClient nexusClient = + NexusClient.newInstance( + stubs, NexusClientOptions.newBuilder().setNamespace(namespace).build()); + // Typed service client: dispatches operations by method reference on the service interface. + NexusServiceClient greetingClient = + nexusClient.newNexusServiceClient(GreetingNexusService.class, ENDPOINT_NAME); + + // execute() starts the operation and blocks until it completes. The handler backs the operation + // with a standalone Activity, so this returns once that Activity has produced its result. + GreetingOutput result = + greetingClient.execute( + GreetingNexusService::greet, + StartNexusOperationOptions.newBuilder() + .setId("greeting-" + UUID.randomUUID()) + .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + .build(), + new GreetingInput("World")); + + logger.info(result.getMessage()); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/README.md b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/README.md new file mode 100644 index 00000000..d2e789d6 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/README.md @@ -0,0 +1,69 @@ +## Nexus Operation Backed by a Standalone Activity + +> [!WARNING] +> Standalone Nexus operations and standalone Activities are experimental and may be subject to +> backwards-incompatible changes. They require a Temporal server that implements and enables them +> via the dynamic configs shown below. Use the dev server build at +> https://github.com/temporalio/cli/releases/tag/v1.7.4-standalone-nexus-operations. + +This sample shows how to implement a Nexus operation whose backing execution is a **standalone +Activity**. `TemporalOperationHandler` maps the Temporal execution onto the Nexus operation: +starting the operation starts the Activity, and when the Activity finishes Temporal delivers its +result to the Nexus caller. + +### Sample structure + +| File | Purpose | +|------------------------------------------------------------------------------------|---| +| [`service/GreetingNexusService.java`](./service/GreetingNexusService.java) | Nexus service definition shared by caller and handler | +| [`handler/GreetingActivityImpl.java`](./handler/GreetingActivityImpl.java) | The standalone Activity backing the operation | +| [`handler/GreetingNexusServiceImpl.java`](./handler/GreetingNexusServiceImpl.java) | Operation implementation, via `TemporalOperationHandler.create` and `startActivity` | +| [`handler/HandlerWorker.java`](./handler/HandlerWorker.java) | Worker hosting the Nexus handler and the Activity | +| [`ClientStarter.java`](./ClientStarter.java) | Executes the Nexus operation from client code | + +The starter and worker connect to two different namespaces (a "caller" namespace and a "handler" +namespace) — this mirrors how Nexus is typically used to cross namespace boundaries. The client is +configured via the SDK's [environment configuration](https://docs.temporal.io/develop/environment-configuration) +support (`ClientConfigProfile.load()`), which reads `TEMPORAL_NAMESPACE`, `TEMPORAL_ADDRESS`, etc. +from the environment (and optionally a profile from `temporal.toml`). + +### Run locally against a dev server + +1. Start the [Temporal dev server build that supports standalone Nexus operations](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support) + with the required namespaces pre-created and Activity callbacks enabled: + + ```bash + ./temporal server start-dev \ + --dynamic-config-value activity.enableCallbacks=true \ + --namespace my-caller-namespace \ + --namespace my-handler-namespace + ``` + +2. Create a Nexus endpoint that routes to the handler namespace and the worker's task queue: + + ```bash + ./temporal operator nexus endpoint create \ + --name my-nexus-endpoint \ + --target-namespace my-handler-namespace \ + --target-task-queue nexus-handler-queue + ``` + +3. In a second terminal, start the handler worker in the handler namespace: + + ```bash + TEMPORAL_NAMESPACE=my-handler-namespace \ + ./gradlew -q :core:execute -PmainClass=io.temporal.samples.nexusstandaloneactivity.handler.HandlerWorker + ``` + +4. In a third terminal, run the starter in the caller namespace: + + ```bash + TEMPORAL_NAMESPACE=my-caller-namespace \ + ./gradlew -q :core:execute -PmainClass=io.temporal.samples.nexusstandaloneactivity.ClientStarter + ``` + +Expected output: + +```text +Hello, World! +``` diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingActivity.java b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingActivity.java new file mode 100644 index 00000000..2db8ec41 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingActivity.java @@ -0,0 +1,13 @@ +package io.temporal.samples.nexusstandaloneactivity.handler; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.samples.nexusstandaloneactivity.service.GreetingNexusService; + +/** Activity used as the backing execution for the Nexus operation. */ +@ActivityInterface +public interface GreetingActivity { + + @ActivityMethod + GreetingNexusService.GreetingOutput createGreeting(GreetingNexusService.GreetingInput input); +} diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingActivityImpl.java b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingActivityImpl.java new file mode 100644 index 00000000..8205bb89 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingActivityImpl.java @@ -0,0 +1,12 @@ +package io.temporal.samples.nexusstandaloneactivity.handler; + +import io.temporal.samples.nexusstandaloneactivity.service.GreetingNexusService; + +public class GreetingActivityImpl implements GreetingActivity { + + @Override + public GreetingNexusService.GreetingOutput createGreeting( + GreetingNexusService.GreetingInput input) { + return new GreetingNexusService.GreetingOutput("Hello, " + input.getName() + "!"); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingNexusServiceImpl.java b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingNexusServiceImpl.java new file mode 100644 index 00000000..631184c9 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/GreetingNexusServiceImpl.java @@ -0,0 +1,43 @@ +package io.temporal.samples.nexusstandaloneactivity.handler; + +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.client.StartActivityOptions; +import io.temporal.nexus.Nexus; +import io.temporal.nexus.TemporalOperationHandler; +import io.temporal.samples.nexusstandaloneactivity.service.GreetingNexusService; +import java.time.Duration; + +// Implements the GreetingNexusService operation on top of a standalone Activity. +@ServiceImpl(service = GreetingNexusService.class) +public class GreetingNexusServiceImpl { + + // TemporalOperationHandler.create maps a Temporal execution onto a Nexus operation. Here the + // execution is a standalone Activity: startActivity returns an asynchronous operation result, so + // the Nexus operation stays running until the Activity completes, at which point Temporal + // delivers the Activity's result to the Nexus caller. + @OperationImpl + public OperationHandler + greet() { + return TemporalOperationHandler.create( + (ctx, client, input) -> + client.startActivity( + GreetingActivity.class, + GreetingActivity::createGreeting, + input, + StartActivityOptions.newBuilder() + // Use a business identifier from the operation input so callers can identify + // the same Activity independently of any individual Nexus request. + .setId(getActivityId(input)) + // The task queue is required. This sample runs the Activity on the same queue + // as the Nexus Worker that is handling this operation. + .setTaskQueue(Nexus.getOperationContext().getInfo().getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build())); + } + + static String getActivityId(GreetingNexusService.GreetingInput input) { + return "greeting-" + input.getName(); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/HandlerWorker.java b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/HandlerWorker.java new file mode 100644 index 00000000..5defc283 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/handler/HandlerWorker.java @@ -0,0 +1,24 @@ +package io.temporal.samples.nexusstandaloneactivity.handler; + +import io.temporal.client.WorkflowClient; +import io.temporal.samples.nexusstandaloneactivity.service.ClientOptions; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; + +// Worker that hosts the Nexus service implementation and the Activity backing its operation. The +// task queue must match the Nexus endpoint's target task queue (see README). +public class HandlerWorker { + public static final String TASK_QUEUE_NAME = "nexus-handler-queue"; + + public static void main(String[] args) { + WorkflowClient client = ClientOptions.getWorkflowClient(); + + WorkerFactory factory = WorkerFactory.newInstance(client); + + Worker worker = factory.newWorker(TASK_QUEUE_NAME); + worker.registerActivitiesImplementations(new GreetingActivityImpl()); + worker.registerNexusServiceImplementation(new GreetingNexusServiceImpl()); + + factory.start(); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/service/ClientOptions.java b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/service/ClientOptions.java new file mode 100644 index 00000000..8edb8133 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/service/ClientOptions.java @@ -0,0 +1,28 @@ +package io.temporal.samples.nexusstandaloneactivity.service; + +import io.temporal.client.WorkflowClient; +import io.temporal.envconfig.ClientConfigProfile; +import io.temporal.serviceclient.WorkflowServiceStubs; + +/** + * Builds a {@link WorkflowClient} from the {@code default} profile loaded by {@link + * ClientConfigProfile#load()}. By default, this reads the TOML file at {@code + * TEMPORAL_CONFIG_FILE}, or, if that is unset, {@code [user config dir]/temporalio/temporal.toml}. + * Point that profile at a different server or namespace — or override via {@code TEMPORAL_*} + * environment variables — to run against, for example, a Temporal Cloud namespace with an API key. + */ +public class ClientOptions { + + public static WorkflowClient getWorkflowClient() { + ClientConfigProfile profile; + try { + profile = ClientConfigProfile.load(); + } catch (Exception e) { + throw new RuntimeException("Failed to load client configuration", e); + } + + WorkflowServiceStubs service = + WorkflowServiceStubs.newServiceStubs(profile.toWorkflowServiceStubsOptions()); + return WorkflowClient.newInstance(service, profile.toWorkflowClientOptions()); + } +} diff --git a/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/service/GreetingNexusService.java b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/service/GreetingNexusService.java new file mode 100644 index 00000000..80dded7c --- /dev/null +++ b/core/src/main/java/io/temporal/samples/nexusstandaloneactivity/service/GreetingNexusService.java @@ -0,0 +1,45 @@ +package io.temporal.samples.nexusstandaloneactivity.service; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import io.nexusrpc.Operation; +import io.nexusrpc.Service; + +// Nexus service definition shared by the caller and the handler. It declares a single operation +// whose backing execution is a standalone Activity. +@Service +public interface GreetingNexusService { + + class GreetingInput { + private final String name; + + @JsonCreator(mode = JsonCreator.Mode.PROPERTIES) + public GreetingInput(@JsonProperty("name") String name) { + this.name = name; + } + + @JsonProperty("name") + public String getName() { + return name; + } + } + + class GreetingOutput { + private final String message; + + @JsonCreator(mode = JsonCreator.Mode.PROPERTIES) + public GreetingOutput(@JsonProperty("message") String message) { + this.message = message; + } + + @JsonProperty("message") + public String getMessage() { + return message; + } + } + + // Asynchronous operation: starting it starts a standalone Activity, and the operation completes + // when that Activity returns its result. + @Operation + GreetingOutput greet(GreetingInput input); +} diff --git a/core/src/test/java/io/temporal/samples/nexusstandaloneactivity/NexusStandaloneActivityTest.java b/core/src/test/java/io/temporal/samples/nexusstandaloneactivity/NexusStandaloneActivityTest.java new file mode 100644 index 00000000..1857812d --- /dev/null +++ b/core/src/test/java/io/temporal/samples/nexusstandaloneactivity/NexusStandaloneActivityTest.java @@ -0,0 +1,78 @@ +package io.temporal.samples.nexusstandaloneactivity; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.temporal.api.nexus.v1.Endpoint; +import io.temporal.client.NexusClient; +import io.temporal.client.NexusClientOptions; +import io.temporal.client.NexusServiceClient; +import io.temporal.client.StartNexusOperationOptions; +import io.temporal.samples.nexusstandaloneactivity.handler.GreetingActivityImpl; +import io.temporal.samples.nexusstandaloneactivity.handler.GreetingNexusServiceImpl; +import io.temporal.samples.nexusstandaloneactivity.handler.HandlerWorker; +import io.temporal.samples.nexusstandaloneactivity.service.GreetingNexusService; +import io.temporal.testing.TemporalDevServerOptions; +import io.temporal.testing.TestWorkflowEnvironment; +import io.temporal.worker.Worker; +import java.time.Duration; +import java.util.UUID; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +public class NexusStandaloneActivityTest { + private static final String DEV_SERVER_VERSION = "v1.7.4-standalone-nexus-operations"; + + private static TestWorkflowEnvironment testEnv; + private static Endpoint endpoint; + + @BeforeAll + public static void setUp() { + testEnv = + TestWorkflowEnvironment.startLocal( + TemporalDevServerOptions.newBuilder() + .setDownloadVersion(DEV_SERVER_VERSION) + .setExtraArgs("--dynamic-config-value", "activity.enableCallbacks=true") + .build()); + + endpoint = + testEnv.createNexusEndpoint( + "test-nexus-endpoint-" + UUID.randomUUID(), HandlerWorker.TASK_QUEUE_NAME); + + Worker worker = testEnv.newWorker(HandlerWorker.TASK_QUEUE_NAME); + worker.registerActivitiesImplementations(new GreetingActivityImpl()); + worker.registerNexusServiceImplementation(new GreetingNexusServiceImpl()); + testEnv.start(); + } + + @AfterAll + public static void tearDown() { + if (testEnv != null) { + if (endpoint != null) { + testEnv.deleteNexusEndpoint(endpoint); + } + testEnv.close(); + } + } + + @Test + public void testNexusOperationBackedByStandaloneActivity() { + NexusClient nexusClient = + NexusClient.newInstance( + testEnv.getWorkflowServiceStubs(), + NexusClientOptions.newBuilder().setNamespace(testEnv.getNamespace()).build()); + NexusServiceClient greetingClient = + nexusClient.newNexusServiceClient(GreetingNexusService.class, endpoint.getSpec().getName()); + + GreetingNexusService.GreetingOutput output = + greetingClient.execute( + GreetingNexusService::greet, + StartNexusOperationOptions.newBuilder() + .setId("greeting-" + UUID.randomUUID()) + .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + .build(), + new GreetingNexusService.GreetingInput("Test")); + + assertEquals("Hello, Test!", output.getMessage()); + } +}