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 @@ -11,6 +11,7 @@
import io.temporal.internal.client.RootActivityClientInvoker;
import io.temporal.internal.client.external.GenericWorkflowClientImpl;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.internal.common.PluginUtils;
import io.temporal.internal.util.MethodExtractor;
import io.temporal.serviceclient.MetricsTag;
import io.temporal.serviceclient.WorkflowServiceStubs;
Expand All @@ -23,20 +24,51 @@
import java.util.Map;
import java.util.stream.Stream;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* Implementation of {@link ActivityClient} that delegates calls through the activity interceptor
* chain and ultimately to the Temporal service.
*/
class ActivityClientImpl implements ActivityClient, ActivityClientInternal {

private static final Logger log = LoggerFactory.getLogger(ActivityClientImpl.class);

private final WorkflowServiceStubs stubs;
private final ActivityClientOptions options;
private final ActivityClientCallsInterceptor invoker;
private final ManualActivityCompletionClientFactory manualActivityCompletionClientFactory;
private final Scope metricsScope;

ActivityClientImpl(WorkflowServiceStubs stubs, ActivityClientOptions options) {
// Extract ActivityClientPlugins from service stubs plugins (propagation)
ActivityClientPlugin[] propagatedPlugins =
PluginUtils.extractPlugins(
stubs.getOptions().getPlugins(),
ActivityClientPlugin.class,
ActivityClientPlugin[]::new);

// Merge propagated plugins with activity client-specified plugins
ActivityClientPlugin[] mergedPlugins =
PluginUtils.mergePlugins(
propagatedPlugins,
options.getPlugins(),
ActivityClientPlugin::getName,
log,
"service stubs",
ActivityClientPlugin.class);

// Apply plugin configuration phase (forward order) on user-provided options,
// so plugins see unmodified state before defaults and plugin merging
ActivityClientOptions.Builder builder = ActivityClientOptions.newBuilder(options);
for (ActivityClientPlugin plugin : mergedPlugins) {
plugin.configureActivityClient(builder);
}
// Set merged plugins after configuration, then build
builder.setPlugins(mergedPlugins);
options = builder.build();

this.stubs = stubs;
this.options = options;
this.metricsScope =
Expand Down
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
package io.temporal.client;

import io.temporal.common.Experimental;
import io.temporal.common.context.ContextPropagator;
import io.temporal.common.converter.DataConverter;
import io.temporal.common.converter.GlobalDataConverter;
import io.temporal.common.interceptors.ActivityClientInterceptor;
import java.lang.management.ManagementFactory;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
Expand Down Expand Up @@ -40,12 +42,14 @@ public static final class Builder {
Collections.emptyList();
private static final List<ActivityClientInterceptor> EMPTY_INTERCEPTORS =
Collections.emptyList();
private static final ActivityClientPlugin[] EMPTY_PLUGINS = new ActivityClientPlugin[0];

private String namespace;
private DataConverter dataConverter;
private String identity;
private List<ContextPropagator> contextPropagators;
private List<ActivityClientInterceptor> interceptors;
private ActivityClientPlugin[] plugins;

private Builder() {}

Expand All @@ -58,6 +62,7 @@ private Builder(ActivityClientOptions options) {
identity = options.identity;
contextPropagators = options.contextPropagators;
interceptors = options.interceptors;
plugins = options.plugins;
}

/** Set the namespace this client will operate on. */
Expand Down Expand Up @@ -102,14 +107,26 @@ public Builder setInterceptors(List<ActivityClientInterceptor> interceptors) {
return this;
}

/**
* Set the plugins for this client.
*
* @param plugins specifies the plugins to use with the client.
*/
@Experimental
public Builder setPlugins(ActivityClientPlugin... plugins) {
this.plugins = Objects.requireNonNull(plugins);
return this;
}

public ActivityClientOptions build() {
String name = identity == null ? ManagementFactory.getRuntimeMXBean().getName() : identity;
return new ActivityClientOptions(
namespace == null ? DEFAULT_NAMESPACE : namespace,
dataConverter == null ? GlobalDataConverter.get() : dataConverter,
name,
contextPropagators == null ? EMPTY_CONTEXT_PROPAGATORS : contextPropagators,
interceptors == null ? EMPTY_INTERCEPTORS : interceptors);
interceptors == null ? EMPTY_INTERCEPTORS : interceptors,
plugins == null ? EMPTY_PLUGINS : plugins);
}
}

Expand All @@ -118,18 +135,21 @@ public ActivityClientOptions build() {
private final String identity;
private final List<ContextPropagator> contextPropagators;
private final List<ActivityClientInterceptor> interceptors;
private final ActivityClientPlugin[] plugins;

private ActivityClientOptions(
String namespace,
DataConverter dataConverter,
String identity,
List<ContextPropagator> contextPropagators,
List<ActivityClientInterceptor> interceptors) {
List<ActivityClientInterceptor> interceptors,
ActivityClientPlugin[] plugins) {
this.namespace = namespace;
this.dataConverter = dataConverter;
this.identity = identity;
this.contextPropagators = contextPropagators;
this.interceptors = interceptors;
this.plugins = plugins;
}

/**
Expand Down Expand Up @@ -177,6 +197,16 @@ public List<ActivityClientInterceptor> getInterceptors() {
return interceptors;
}

/**
* Get the plugins of this client.
*
* @return The plugins to use with the client.
*/
@Experimental
public ActivityClientPlugin[] getPlugins() {
return plugins;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
Expand All @@ -186,12 +216,19 @@ public boolean equals(Object o) {
&& Objects.equals(dataConverter, that.dataConverter)
&& Objects.equals(identity, that.identity)
&& Objects.equals(contextPropagators, that.contextPropagators)
&& Objects.equals(interceptors, that.interceptors);
&& Objects.equals(interceptors, that.interceptors)
&& Arrays.equals(plugins, that.plugins);
}

@Override
public int hashCode() {
return Objects.hash(namespace, dataConverter, identity, contextPropagators, interceptors);
return Objects.hash(
namespace,
dataConverter,
identity,
contextPropagators,
interceptors,
Arrays.hashCode(plugins));
}

@Override
Expand All @@ -209,6 +246,8 @@ public String toString() {
+ contextPropagators
+ ", interceptors="
+ interceptors
+ ", plugins="
+ Arrays.toString(plugins)
+ '}';
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
package io.temporal.client;

import io.temporal.common.Experimental;
import javax.annotation.Nonnull;

/**
* Plugin interface for customizing Temporal activity client configuration.
*
* <p>Plugins that implement both {@link io.temporal.serviceclient.WorkflowServiceStubsPlugin} and
* {@code ActivityClientPlugin} are automatically propagated from the service stubs to the activity
* client.
*
* @see io.temporal.serviceclient.WorkflowServiceStubsPlugin
*/
@Experimental
public interface ActivityClientPlugin {

/**
* Returns a unique name for this plugin. Used for logging and duplicate detection. Recommended
* format: "organization.plugin-name" (e.g., "io.temporal.tracing")
*
* @return fully qualified plugin name
*/
@Nonnull
String getName();

/**
* Allows the plugin to modify activity client options before the client is created. Called during
* configuration phase in forward (registration) order.
*
* @param builder the options builder to modify
*/
void configureActivityClient(@Nonnull ActivityClientOptions.Builder builder);
}
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import io.temporal.internal.client.RootNexusClientInvoker;
import io.temporal.internal.client.external.GenericWorkflowClient;
import io.temporal.internal.client.external.GenericWorkflowClientImpl;
import io.temporal.internal.common.PluginUtils;
import io.temporal.serviceclient.MetricsTag;
import io.temporal.serviceclient.WorkflowServiceStubs;
import java.util.List;
Expand All @@ -39,6 +40,32 @@ public class NexusClientImpl implements NexusClient {

public static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) {
enforceNonWorkflowThread();

// Extract NexusClientPlugins from service stubs plugins (propagation)
NexusClientPlugin[] propagatedPlugins =
PluginUtils.extractPlugins(
service.getOptions().getPlugins(), NexusClientPlugin.class, NexusClientPlugin[]::new);

// Merge propagated plugins with Nexus client-specified plugins
NexusClientPlugin[] mergedPlugins =
PluginUtils.mergePlugins(
propagatedPlugins,
options.getPlugins(),
NexusClientPlugin::getName,
log,
"service stubs",
NexusClientPlugin.class);

// Apply plugin configuration phase (forward order) on user-provided options,
// so plugins see unmodified state before defaults and plugin merging
NexusClientOptions.Builder builder = NexusClientOptions.newBuilder(options);
for (NexusClientPlugin plugin : mergedPlugins) {
plugin.configureNexusClient(builder);
}
// Set merged plugins after configuration, then build
builder.setPlugins(mergedPlugins);
options = builder.build();

return WorkflowThreadMarker.protectFromWorkflowThread(
new NexusClientImpl(service, options.toResolvedOptions()), NexusClient.class);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import java.lang.management.ManagementFactory;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import javax.annotation.Nullable;

/**
Expand Down Expand Up @@ -42,18 +43,21 @@ public class NexusClientOptions {
private final DataConverter dataConverter;
private final String identity;
private final @Nullable ExternalStorage externalStorage;
private final NexusClientPlugin[] plugins;

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

/** Get the namespace this client will operate on. */
Expand Down Expand Up @@ -88,6 +92,11 @@ public String getIdentity() {
return identity;
}

/** Get the plugins of this client. */
public NexusClientPlugin[] getPlugins() {
return plugins;
}

/**
* Converts this {@link NexusClientOptions} instance into a {@link NexusClientResolvedOptions}
* instance, which contains the fully resolved runtime settings used by the internal Nexus client.
Expand Down Expand Up @@ -130,11 +139,14 @@ public static NexusClientOptions getDefaultInstance() {

/** Builder for {@link NexusClientOptions}. */
public static class Builder {
private static final NexusClientPlugin[] EMPTY_PLUGINS = new NexusClientPlugin[0];

private String namespace;
private List<NexusClientInterceptor> interceptors = Collections.emptyList();
private DataConverter dataConverter = GlobalDataConverter.get();
private String identity;
private ExternalStorage externalStorage;
private NexusClientPlugin[] plugins;

private Builder() {}

Expand All @@ -147,6 +159,7 @@ private Builder(NexusClientOptions options) {
dataConverter = options.dataConverter;
identity = options.identity;
externalStorage = options.externalStorage;
plugins = options.plugins;
}

/** Set the namespace this client will operate on. */
Expand Down Expand Up @@ -195,6 +208,16 @@ public NexusClientOptions.Builder setExternalStorage(
return this;
}

/**
* Set the plugins for this client.
*
* @param plugins specifies the plugins to use with the client.
*/
public NexusClientOptions.Builder setPlugins(NexusClientPlugin... plugins) {
this.plugins = Objects.requireNonNull(plugins);
return this;
}

public NexusClientOptions build() {
String resolvedIdentity =
identity == null ? ManagementFactory.getRuntimeMXBean().getName() : identity;
Expand All @@ -203,7 +226,8 @@ public NexusClientOptions build() {
interceptors,
dataConverter,
resolvedIdentity,
externalStorage);
externalStorage,
plugins == null ? EMPTY_PLUGINS : plugins);
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
package io.temporal.client;

import io.temporal.common.Experimental;
import javax.annotation.Nonnull;

/**
* Plugin interface for customizing Temporal Nexus client configuration.
*
* <p>Plugins that implement both {@link io.temporal.serviceclient.WorkflowServiceStubsPlugin} and
* {@code NexusClientPlugin} are automatically propagated from the service stubs to the Nexus
* client.
*
* @see io.temporal.serviceclient.WorkflowServiceStubsPlugin
*/
@Experimental
public interface NexusClientPlugin {

/**
* Returns a unique name for this plugin. Used for logging and duplicate detection. Recommended
* format: "organization.plugin-name" (e.g., "io.temporal.tracing")
*
* @return fully qualified plugin name
*/
@Nonnull
String getName();

/**
* Allows the plugin to modify Nexus client options before the client is created. Called during
* configuration phase in forward (registration) order.
*
* @param builder the options builder to modify
*/
void configureNexusClient(@Nonnull NexusClientOptions.Builder builder);
}
Loading
Loading