Add data converters, activities, workflows, Nexus services

Author: donald-pinckney

temporal-sdk/src/main/java/io/temporal/common/SimplePlugin.java

@@ -25,6 +25,7 @@
 import io.temporal.client.schedules.ScheduleClientOptions;
 import io.temporal.client.schedules.ScheduleClientPlugin;
 import io.temporal.common.context.ContextPropagator;
+import io.temporal.common.converter.DataConverter;
 import io.temporal.common.interceptors.WorkerInterceptor;
 import io.temporal.common.interceptors.WorkflowClientInterceptor;
 import io.temporal.serviceclient.WorkflowServiceStubs;
@@ -125,6 +126,10 @@ public abstract class SimplePlugin
   private final List<WorkerInterceptor> workerInterceptors;
   private final List<WorkflowClientInterceptor> clientInterceptors;
   private final List<ContextPropagator> contextPropagators;
+  private final DataConverter dataConverter;
+  private final List<Class<?>> workflowImplementationTypes;
+  private final List<Object> activitiesImplementations;
+  private final List<Object> nexusServiceImplementations;
 
   /**
    * Creates a new plugin with the specified name. Use this constructor when subclassing to override
@@ -149,6 +154,10 @@ protected SimplePlugin(@Nonnull String name) {
     this.workerInterceptors = Collections.emptyList();
     this.clientInterceptors = Collections.emptyList();
     this.contextPropagators = Collections.emptyList();
+    this.dataConverter = null;
+    this.workflowImplementationTypes = Collections.emptyList();
+    this.activitiesImplementations = Collections.emptyList();
+    this.nexusServiceImplementations = Collections.emptyList();
   }
 
   /**
@@ -174,6 +183,10 @@ protected SimplePlugin(@Nonnull Builder builder) {
     this.workerInterceptors = new ArrayList<>(builder.workerInterceptors);
     this.clientInterceptors = new ArrayList<>(builder.clientInterceptors);
     this.contextPropagators = new ArrayList<>(builder.contextPropagators);
+    this.dataConverter = builder.dataConverter;
+    this.workflowImplementationTypes = new ArrayList<>(builder.workflowImplementationTypes);
+    this.activitiesImplementations = new ArrayList<>(builder.activitiesImplementations);
+    this.nexusServiceImplementations = new ArrayList<>(builder.nexusServiceImplementations);
   }
 
   /**
@@ -207,6 +220,11 @@ public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder build
       customizer.accept(builder);
     }
 
+    // Set data converter
+    if (dataConverter != null) {
+      builder.setDataConverter(dataConverter);
+    }
+
     // Add client interceptors
     if (!clientInterceptors.isEmpty()) {
       WorkflowClientInterceptor[] existing = builder.build().getInterceptors();
@@ -260,6 +278,23 @@ public void configureWorker(@Nonnull String taskQueue, @Nonnull WorkerOptions.Bu
 
   @Override
   public void initializeWorker(@Nonnull String taskQueue, @Nonnull Worker worker) {
+    // Register workflow implementation types
+    if (!workflowImplementationTypes.isEmpty()) {
+      worker.registerWorkflowImplementationTypes(
+          workflowImplementationTypes.toArray(new Class<?>[0]));
+    }
+
+    // Register activities implementations
+    if (!activitiesImplementations.isEmpty()) {
+      worker.registerActivitiesImplementations(activitiesImplementations.toArray());
+    }
+
+    // Register nexus service implementations
+    for (Object nexusService : nexusServiceImplementations) {
+      worker.registerNexusServiceImplementation(nexusService);
+    }
+
+    // Apply custom initializers
     for (BiConsumer<String, Worker> initializer : workerInitializers) {
       initializer.accept(taskQueue, worker);
     }
@@ -331,6 +366,10 @@ public static final class Builder {
     private final List<WorkerInterceptor> workerInterceptors = new ArrayList<>();
     private final List<WorkflowClientInterceptor> clientInterceptors = new ArrayList<>();
     private final List<ContextPropagator> contextPropagators = new ArrayList<>();
+    private DataConverter dataConverter;
+    private final List<Class<?>> workflowImplementationTypes = new ArrayList<>();
+    private final List<Object> activitiesImplementations = new ArrayList<>();
+    private final List<Object> nexusServiceImplementations = new ArrayList<>();
 
     private Builder(@Nonnull String name) {
       this.name = Objects.requireNonNull(name, "Plugin name cannot be null");
@@ -558,6 +597,78 @@ public Builder addContextPropagators(ContextPropagator... propagators) {
       return this;
     }
 
+    /**
+     * Sets the data converter to use for serializing workflow and activity arguments and results.
+     * This overrides any data converter previously set on the client options.
+     *
+     * @param dataConverter the data converter to use
+     * @return this builder for chaining
+     */
+    public Builder setDataConverter(@Nonnull DataConverter dataConverter) {
+      this.dataConverter = Objects.requireNonNull(dataConverter);
+      return this;
+    }
+
+    /**
+     * Registers workflow implementation types. These workflows will be registered on all workers
+     * created by the factory.
+     *
+     * <p>Example:
+     *
+     * <pre>{@code
+     * SimplePlugin.newBuilder("my-plugin")
+     *     .registerWorkflowImplementationTypes(MyWorkflowImpl.class, OtherWorkflowImpl.class)
+     *     .build();
+     * }</pre>
+     *
+     * @param workflowImplementationTypes workflow implementation classes to register
+     * @return this builder for chaining
+     */
+    public Builder registerWorkflowImplementationTypes(Class<?>... workflowImplementationTypes) {
+      this.workflowImplementationTypes.addAll(Arrays.asList(workflowImplementationTypes));
+      return this;
+    }
+
+    /**
+     * Registers activity implementations. These activities will be registered on all workers
+     * created by the factory.
+     *
+     * <p>Example:
+     *
+     * <pre>{@code
+     * SimplePlugin.newBuilder("my-plugin")
+     *     .registerActivitiesImplementations(new MyActivityImpl(), new OtherActivityImpl())
+     *     .build();
+     * }</pre>
+     *
+     * @param activitiesImplementations activity implementation instances to register
+     * @return this builder for chaining
+     */
+    public Builder registerActivitiesImplementations(Object... activitiesImplementations) {
+      this.activitiesImplementations.addAll(Arrays.asList(activitiesImplementations));
+      return this;
+    }
+
+    /**
+     * Registers a Nexus service implementation. The service will be registered on all workers
+     * created by the factory.
+     *
+     * <p>Example:
+     *
+     * <pre>{@code
+     * SimplePlugin.newBuilder("my-plugin")
+     *     .registerNexusServiceImplementation(new MyNexusServiceImpl())
+     *     .build();
+     * }</pre>
+     *
+     * @param nexusServiceImplementation the Nexus service implementation to register
+     * @return this builder for chaining
+     */
+    public Builder registerNexusServiceImplementation(@Nonnull Object nexusServiceImplementation) {
+      this.nexusServiceImplementations.add(Objects.requireNonNull(nexusServiceImplementation));
+      return this;
+    }
+
     /**
      * Builds the plugin with the configured settings.
      *

temporal-sdk/src/test/java/io/temporal/common/SimplePluginBuilderTest.java

@@ -21,13 +21,16 @@
 package io.temporal.common;
 
 import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
 
 import io.temporal.client.WorkflowClientOptions;
+import io.temporal.common.converter.DataConverter;
 import io.temporal.common.interceptors.WorkerInterceptor;
 import io.temporal.common.interceptors.WorkerInterceptorBase;
 import io.temporal.common.interceptors.WorkflowClientInterceptor;
 import io.temporal.common.interceptors.WorkflowClientInterceptorBase;
 import io.temporal.serviceclient.WorkflowServiceStubsOptions;
+import io.temporal.worker.Worker;
 import io.temporal.worker.WorkerFactoryOptions;
 import io.temporal.worker.WorkerOptions;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -409,4 +412,107 @@ public void testNullName() {
   public void testNullCustomizer() {
     SimplePlugin.newBuilder("test").customizeClient(null);
   }
+
+  @Test
+  public void testSetDataConverter() {
+    DataConverter customConverter = mock(DataConverter.class);
+
+    SimplePlugin plugin = SimplePlugin.newBuilder("test").setDataConverter(customConverter).build();
+
+    WorkflowClientOptions.Builder builder = WorkflowClientOptions.newBuilder();
+    ((io.temporal.client.WorkflowClientPlugin) plugin).configureWorkflowClient(builder);
+
+    assertSame(customConverter, builder.build().getDataConverter());
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void testNullDataConverter() {
+    SimplePlugin.newBuilder("test").setDataConverter(null);
+  }
+
+  @Test
+  public void testRegisterWorkflowImplementationTypes() {
+    SimplePlugin plugin =
+        SimplePlugin.newBuilder("test")
+            .registerWorkflowImplementationTypes(String.class, Integer.class)
+            .build();
+
+    Worker mockWorker = mock(Worker.class);
+    ((io.temporal.worker.WorkerPlugin) plugin).initializeWorker("test-queue", mockWorker);
+
+    verify(mockWorker).registerWorkflowImplementationTypes(String.class, Integer.class);
+  }
+
+  @Test
+  public void testRegisterActivitiesImplementations() {
+    Object activity1 = new Object();
+    Object activity2 = new Object();
+
+    SimplePlugin plugin =
+        SimplePlugin.newBuilder("test")
+            .registerActivitiesImplementations(activity1, activity2)
+            .build();
+
+    Worker mockWorker = mock(Worker.class);
+    ((io.temporal.worker.WorkerPlugin) plugin).initializeWorker("test-queue", mockWorker);
+
+    verify(mockWorker).registerActivitiesImplementations(activity1, activity2);
+  }
+
+  @Test
+  public void testRegisterNexusServiceImplementation() {
+    Object nexusService = new Object();
+
+    SimplePlugin plugin =
+        SimplePlugin.newBuilder("test").registerNexusServiceImplementation(nexusService).build();
+
+    Worker mockWorker = mock(Worker.class);
+    ((io.temporal.worker.WorkerPlugin) plugin).initializeWorker("test-queue", mockWorker);
+
+    verify(mockWorker).registerNexusServiceImplementation(nexusService);
+  }
+
+  @Test
+  public void testRegisterMultipleNexusServiceImplementations() {
+    Object nexusService1 = new Object();
+    Object nexusService2 = new Object();
+
+    SimplePlugin plugin =
+        SimplePlugin.newBuilder("test")
+            .registerNexusServiceImplementation(nexusService1)
+            .registerNexusServiceImplementation(nexusService2)
+            .build();
+
+    Worker mockWorker = mock(Worker.class);
+    ((io.temporal.worker.WorkerPlugin) plugin).initializeWorker("test-queue", mockWorker);
+
+    verify(mockWorker).registerNexusServiceImplementation(nexusService1);
+    verify(mockWorker).registerNexusServiceImplementation(nexusService2);
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void testNullNexusServiceImplementation() {
+    SimplePlugin.newBuilder("test").registerNexusServiceImplementation(null);
+  }
+
+  @Test
+  public void testRegistrationsWithCustomInitializer() {
+    AtomicBoolean customInitializerCalled = new AtomicBoolean(false);
+
+    SimplePlugin plugin =
+        SimplePlugin.newBuilder("test")
+            .registerWorkflowImplementationTypes(String.class)
+            .registerActivitiesImplementations(new Object())
+            .initializeWorker((taskQueue, worker) -> customInitializerCalled.set(true))
+            .build();
+
+    Worker mockWorker = mock(Worker.class);
+    ((io.temporal.worker.WorkerPlugin) plugin).initializeWorker("test-queue", mockWorker);
+
+    // Verify registrations happen before custom initializer
+    verify(mockWorker).registerWorkflowImplementationTypes(String.class);
+    verify(mockWorker).registerActivitiesImplementations(any());
+    assertTrue(
+        "Custom initializer should be called after registrations", customInitializerCalled.get());
+  }
 }