Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import io.temporal.common.interceptors.NexusClientInterceptor;
import io.temporal.internal.WorkflowThreadMarker;
import io.temporal.internal.client.NamespaceInjectWorkflowServiceStubs;
import io.temporal.internal.client.NexusClientResolvedOptions;
import io.temporal.internal.client.NexusOperationHandleImpl;
import io.temporal.internal.client.RootNexusClientInvoker;
import io.temporal.internal.client.external.GenericWorkflowClient;
Expand All @@ -30,7 +31,7 @@ public class NexusClientImpl implements NexusClient {
private static final Logger log = LoggerFactory.getLogger(NexusClientImpl.class);

private final WorkflowServiceStubs workflowServiceStubs;
private final NexusClientOptions options;
private final NexusClientResolvedOptions options;
private final GenericWorkflowClient genericClient;
private final Scope metricsScope;
private final NexusClientCallsInterceptor nexusClientCallsInvoker;
Expand All @@ -39,10 +40,10 @@ public class NexusClientImpl implements NexusClient {
public static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) {
enforceNonWorkflowThread();
return WorkflowThreadMarker.protectFromWorkflowThread(
new NexusClientImpl(service, options), NexusClient.class);
new NexusClientImpl(service, options.toResolvedOptions()), NexusClient.class);
}

NexusClientImpl(WorkflowServiceStubs workflowServiceStubs, NexusClientOptions options) {
NexusClientImpl(WorkflowServiceStubs workflowServiceStubs, NexusClientResolvedOptions options) {
workflowServiceStubs =
new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace());
this.workflowServiceStubs = workflowServiceStubs;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,14 @@
import io.temporal.common.converter.DataConverter;
import io.temporal.common.converter.GlobalDataConverter;
import io.temporal.common.interceptors.NexusClientInterceptor;
import io.temporal.internal.client.NexusClientResolvedOptions;
import io.temporal.internal.payload.storage.ExternalStorageDataConverter;
import io.temporal.internal.payload.storage.ExternalStorageRunner;
import io.temporal.payload.storage.ExternalStorage;
import java.lang.management.ManagementFactory;
import java.util.Collections;
import java.util.List;
import javax.annotation.Nullable;

/**
* Options that configure a {@link NexusClient} (and the service-bound clients it produces).
Expand Down Expand Up @@ -36,16 +41,19 @@ public class NexusClientOptions {
private final List<NexusClientInterceptor> interceptors;
private final DataConverter dataConverter;
private final String identity;
private final @Nullable ExternalStorage externalStorage;

private NexusClientOptions(
String namespace,
List<NexusClientInterceptor> interceptors,
DataConverter dataConverter,
String identity) {
String identity,
@Nullable ExternalStorage externalStorage) {
this.namespace = namespace;
this.interceptors = interceptors;
this.dataConverter = dataConverter;
this.identity = identity;
this.externalStorage = externalStorage;
}

/** Get the namespace this client will operate on. */
Expand All @@ -63,6 +71,15 @@ public DataConverter getDataConverter() {
return dataConverter;
}

/**
* Get the external storage used to offload large operation payloads, or null if payloads are sent
* inline.
*/
@Nullable
public ExternalStorage getExternalStorage() {
Comment thread
cconstable marked this conversation as resolved.
return externalStorage;
}

/**
* Human-readable identity of this client. Stamped onto outgoing write requests (start, cancel,
* terminate) so server-side history and audit trails can attribute the action to a caller.
Expand All @@ -71,6 +88,22 @@ public String getIdentity() {
return identity;
}

/**
* Converts this {@link NexusClientOptions} instance into a {@link NexusClientResolvedOptions}
* instance, which contains the fully resolved runtime settings used by the internal Nexus client.
*
* @return a {@link NexusClientResolvedOptions} instance with the resolved options
*/
NexusClientResolvedOptions toResolvedOptions() {
DataConverter resolvedDataConverter = dataConverter;
if (externalStorage != null) {
resolvedDataConverter =
new ExternalStorageDataConverter(
resolvedDataConverter, ExternalStorageRunner.create(externalStorage));
}
return new NexusClientResolvedOptions(namespace, interceptors, resolvedDataConverter, identity);
}

/** Returns a fresh builder. */
public static NexusClientOptions.Builder newBuilder() {
return new NexusClientOptions.Builder();
Expand Down Expand Up @@ -101,6 +134,7 @@ public static class Builder {
private List<NexusClientInterceptor> interceptors = Collections.emptyList();
private DataConverter dataConverter = GlobalDataConverter.get();
private String identity;
private ExternalStorage externalStorage;

private Builder() {}

Expand All @@ -112,6 +146,7 @@ private Builder(NexusClientOptions options) {
interceptors = options.interceptors;
dataConverter = options.dataConverter;
identity = options.identity;
externalStorage = options.externalStorage;
}

/** Set the namespace this client will operate on. */
Expand Down Expand Up @@ -148,14 +183,27 @@ public NexusClientOptions.Builder setIdentity(String identity) {
return this;
}

/**
* Offload operation payloads that exceed the storage threshold to {@code externalStorage},
* sending a reference to the server in their place. The client wraps its {@link DataConverter}
* to store outbound payloads and resolve inbound references. Defaults to null, meaning all
* payloads are sent inline.
*/
public NexusClientOptions.Builder setExternalStorage(
Comment thread
cconstable marked this conversation as resolved.
@Nullable ExternalStorage externalStorage) {
this.externalStorage = externalStorage;
return this;
}

public NexusClientOptions build() {
String resolvedIdentity =
identity == null ? ManagementFactory.getRuntimeMXBean().getName() : identity;
return new NexusClientOptions(
namespace == null ? DEFAULT_NAMESPACE : namespace,
interceptors,
dataConverter,
resolvedIdentity);
resolvedIdentity,
externalStorage);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import io.nexusrpc.ServiceDefinition;
import io.temporal.common.Experimental;
import io.temporal.common.interceptors.NexusClientCallsInterceptor;
import io.temporal.internal.client.NexusClientResolvedOptions;
import io.temporal.internal.util.MethodExtractor;
import io.temporal.workflow.Functions;
import java.lang.reflect.Method;
Expand All @@ -26,7 +27,7 @@ class NexusServiceClientImpl<T> extends UntypedNexusServiceClientImpl
NexusClientCallsInterceptor invoker,
Class<T> serviceInterface,
String endpoint,
NexusClientOptions options) {
NexusClientResolvedOptions options) {
this(
invoker,
serviceInterface,
Expand All @@ -40,7 +41,7 @@ private NexusServiceClientImpl(
Class<T> serviceInterface,
ServiceDefinition serviceDef,
String endpoint,
NexusClientOptions options) {
NexusClientResolvedOptions options) {
super(invoker, endpoint, serviceDef.getName(), options);
this.serviceInterface = serviceInterface;
this.serviceDef = serviceDef;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import io.temporal.common.interceptors.NexusClientCallsInterceptor;
import io.temporal.common.interceptors.NexusClientCallsInterceptor.StartNexusOperationExecutionInput;
import io.temporal.common.interceptors.NexusClientCallsInterceptor.StartNexusOperationExecutionOutput;
import io.temporal.internal.client.NexusClientResolvedOptions;
import io.temporal.internal.client.NexusOperationHandleImpl;
import java.lang.reflect.Type;
import java.util.Collections;
Expand All @@ -28,7 +29,7 @@ class UntypedNexusServiceClientImpl implements UntypedNexusServiceClient {
NexusClientCallsInterceptor invoker,
String endpoint,
String serviceName,
NexusClientOptions clientOptions) {
NexusClientResolvedOptions clientOptions) {
if (invoker == null || endpoint == null || serviceName == null || clientOptions == null) {
throw new IllegalArgumentException(
"invoker, endpoint, serviceName, and clientOptions are all required");
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
package io.temporal.internal.client;

import io.temporal.common.converter.DataConverter;
import io.temporal.common.interceptors.NexusClientInterceptor;
import java.util.List;

/** Resolved runtime settings used by the internal Nexus client implementation. */
public final class NexusClientResolvedOptions {

private final String namespace;
private final List<NexusClientInterceptor> interceptors;
private final DataConverter dataConverter;
private final String identity;

public NexusClientResolvedOptions(
String namespace,
List<NexusClientInterceptor> interceptors,
DataConverter dataConverter,
String identity) {
this.namespace = namespace;
this.interceptors = interceptors;
this.dataConverter = dataConverter;
this.identity = identity;
}

public String getNamespace() {
return namespace;
}

public List<NexusClientInterceptor> getInterceptors() {
return interceptors;
}

public DataConverter getDataConverter() {
return dataConverter;
}

public String getIdentity() {
return identity;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
import io.temporal.api.workflowservice.v1.StartNexusOperationExecutionRequest;
import io.temporal.api.workflowservice.v1.StartNexusOperationExecutionResponse;
import io.temporal.api.workflowservice.v1.TerminateNexusOperationExecutionRequest;
import io.temporal.client.NexusClientOptions;
import io.temporal.client.NexusOperationAlreadyStartedException;
import io.temporal.client.NexusOperationExecutionCount;
import io.temporal.client.NexusOperationExecutionDescription;
Expand Down Expand Up @@ -53,10 +52,10 @@
public class RootNexusClientInvoker implements NexusClientCallsInterceptor {

private final GenericWorkflowClient genericClient;
private final NexusClientOptions clientOptions;
private final NexusClientResolvedOptions clientOptions;

public RootNexusClientInvoker(
GenericWorkflowClient genericClient, NexusClientOptions clientOptions) {
GenericWorkflowClient genericClient, NexusClientResolvedOptions clientOptions) {
this.genericClient = genericClient;
this.clientOptions = clientOptions;
}
Expand Down
Loading
Loading