Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
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
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@ to docs, or any other relevant information.
to the given name.
- Added experimental `WorkflowUnsafe.isReadOnly()`, which reports whether the calling code is read-only.
replay.
- Added experimental `ActivityClientPlugin` and `NexusClientPlugin`, which plugins on the service stubs use to configure
`ActivityClient` and `NexusClient`. `SimplePlugin` implements both, and its builder gained
`addScheduleClientInterceptors`, `addActivityClientInterceptors`, and `addNexusClientInterceptors`.

### Changed
- Moved the Spring AI integration to [temporalio/ai-integrations](https://github.com/temporalio/ai-integrations/tree/main/java/spring-ai).
Expand All @@ -71,6 +74,8 @@ to docs, or any other relevant information.
- `TemporalDevServer`, `TemporalDevServerOptions`, `TestWorkflowEnvironment.startLocal`, and the `useDevServer` methods
of `TestWorkflowRule` and `TestWorkflowExtension` are no longer marked experimental. `TemporalDevServerOptions` remain
specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes.
- `SimplePlugin` now applies its data converter customizer to schedule and Nexus clients, and its context
propagators to schedule and activity clients. Previously `SimplePlugin.configureScheduleClient` did nothing.

### Fixed
- `ActivityClient` now sends the `temporal-namespace` gRPC header like `WorkflowClient` and `ScheduleClient` do,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,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 @@ -26,20 +27,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 the ActivityClientPlugins propagated from the service stubs plugins.
ActivityClientPlugin[] propagatedPlugins =
PluginUtils.extractPlugins(
stubs.getOptions().getPlugins(),
ActivityClientPlugin.class,
ActivityClientPlugin[]::new);
Comment thread
patbeqo marked this conversation as resolved.

// 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();

DataConverter internalDataConverter =
ClientDataConverterFactory.forClient(options.getDataConverter());
this.stubs = stubs;
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 @@ -18,6 +18,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.internal.payload.storage.ExternalStorageRunner;
import io.temporal.serviceclient.MetricsTag;
import io.temporal.serviceclient.WorkflowServiceStubs;
Expand All @@ -41,6 +42,32 @@ public class NexusClientImpl implements NexusClient {

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

// Extract the NexusClientPlugins propagated from the service stubs plugins.
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), NexusClient.class);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,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 @@ -40,18 +41,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 @@ -86,6 +90,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 @@ -123,11 +132,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 @@ -140,6 +152,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 @@ -188,6 +201,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 @@ -196,7 +219,8 @@ public NexusClientOptions build() {
interceptors,
dataConverter,
resolvedIdentity,
externalStorage);
externalStorage,
plugins == null ? EMPTY_PLUGINS : plugins);
}
}
}
Loading
Loading