[Fix #936] Alternative approach

Signed-off-by: fjtirado <[email protected]>

Author: fjtirado

impl/container/src/main/java/io/serverlessworkflow/impl/container/executors/ContainerRunner.java

@@ -28,24 +28,21 @@
 import com.github.dockerjava.core.DockerClientImpl;
 import com.github.dockerjava.core.NameParser;
 import com.github.dockerjava.httpclient5.ApacheDockerHttpClient;
-import io.serverlessworkflow.api.types.Container;
-import io.serverlessworkflow.api.types.ContainerLifetime;
-import io.serverlessworkflow.api.types.TimeoutAfter;
+import io.serverlessworkflow.api.types.ContainerLifetime.ContainerCleanupPolicy;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
-import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowUtils;
 import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.executors.CallableTask;
 import java.io.IOException;
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
-class ContainerRunner {
+class ContainerRunner implements CallableTask {
 
   private static final DefaultDockerClientConfig DEFAULT_CONFIG =
       DefaultDockerClientConfig.createDefaultConfigBuilder().build();
@@ -64,14 +61,19 @@ private static class DockerClientHolder {
   private final ContainerCleanupPolicy policy;
   private final String containerImage;
 
-  private ContainerRunner(ContainerRunnerBuilder builder) {
-    this.propertySetters = builder.propertySetters;
-    this.timeout = Optional.ofNullable(builder.timeout);
-    this.policy = builder.policy;
-    this.containerImage = builder.containerImage;
+  public ContainerRunner(
+      Collection<ContainerPropertySetter> propertySetters,
+      Optional<WorkflowValueResolver<Duration>> timeout,
+      ContainerCleanupPolicy policy,
+      String containerImage) {
+    this.propertySetters = propertySetters;
+    this.timeout = timeout;
+    this.policy = policy;
+    this.containerImage = containerImage;
   }
 
-  CompletableFuture<WorkflowModel> start(
+  @Override
+  public CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     return CompletableFuture.supplyAsync(
         () -> startSync(workflowContext, taskContext, input),
@@ -215,52 +217,4 @@ private static RuntimeException mapExitCode(int exit) {
   private static RuntimeException failed(String message) {
     return new RuntimeException(message);
   }
-
-  static ContainerRunnerBuilder builder() {
-    return new ContainerRunnerBuilder();
-  }
-
-  public static class ContainerRunnerBuilder {
-    private Container container;
-    private WorkflowDefinition definition;
-    private WorkflowValueResolver<Duration> timeout;
-    private ContainerCleanupPolicy policy;
-    private String containerImage;
-    private Collection<ContainerPropertySetter> propertySetters = new ArrayList<>();
-
-    private ContainerRunnerBuilder() {}
-
-    ContainerRunnerBuilder withContainer(Container container) {
-      this.container = container;
-      return this;
-    }
-
-    public ContainerRunnerBuilder withWorkflowDefinition(WorkflowDefinition definition) {
-      this.definition = definition;
-      return this;
-    }
-
-    ContainerRunner build() {
-      propertySetters.add(new NamePropertySetter(definition, container));
-      propertySetters.add(new CommandPropertySetter(definition, container));
-      propertySetters.add(new ContainerEnvironmentPropertySetter(definition, container));
-      propertySetters.add(new LifetimePropertySetter(container));
-      propertySetters.add(new PortsPropertySetter(container));
-      propertySetters.add(new VolumesPropertySetter(definition, container));
-
-      containerImage = container.getImage();
-      if (containerImage == null || container.getImage().isBlank()) {
-        throw new IllegalArgumentException("Container image must be provided");
-      }
-      ContainerLifetime lifetime = container.getLifetime();
-      if (lifetime != null) {
-        policy = lifetime.getCleanup();
-        TimeoutAfter afterTimeout = lifetime.getAfter();
-        if (afterTimeout != null)
-          timeout = WorkflowUtils.fromTimeoutAfter(definition.application(), afterTimeout);
-      }
-
-      return new ContainerRunner(this);
-    }
-  }
 }

impl/container/src/main/java/io/serverlessworkflow/impl/container/executors/RunContainerExecutor.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.container.executors;
+
+import io.serverlessworkflow.api.types.Container;
+import io.serverlessworkflow.api.types.ContainerLifetime;
+import io.serverlessworkflow.api.types.ContainerLifetime.ContainerCleanupPolicy;
+import io.serverlessworkflow.api.types.RunContainer;
+import io.serverlessworkflow.api.types.RunTaskConfiguration;
+import io.serverlessworkflow.api.types.TimeoutAfter;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowUtils;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.RunnableTaskBuilder;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Objects;
+import java.util.Optional;
+
+public class RunContainerExecutorBuilder implements RunnableTaskBuilder<RunContainer> {
+
+  @Override
+  public CallableTask build(RunContainer taskConfiguration, WorkflowDefinition definition) {
+    Collection<ContainerPropertySetter> propertySetters = new ArrayList<>();
+    Container container = taskConfiguration.getContainer();
+    propertySetters.add(new NamePropertySetter(definition, container));
+    propertySetters.add(new CommandPropertySetter(definition, container));
+    propertySetters.add(new ContainerEnvironmentPropertySetter(definition, container));
+    propertySetters.add(new LifetimePropertySetter(container));
+    propertySetters.add(new PortsPropertySetter(container));
+    propertySetters.add(new VolumesPropertySetter(definition, container));
+
+    if (WorkflowUtils.isValid(container.getImage())) {
+      throw new IllegalArgumentException();
+    }
+    ContainerCleanupPolicy policy = null;
+    WorkflowValueResolver<Duration> timeout = null;
+    ContainerLifetime lifetime = container.getLifetime();
+    if (lifetime != null) {
+      policy = lifetime.getCleanup();
+      TimeoutAfter afterTimeout = lifetime.getAfter();
+      if (afterTimeout != null)
+        timeout = WorkflowUtils.fromTimeoutAfter(definition.application(), afterTimeout);
+    }
+    return new ContainerRunner(
+        propertySetters,
+        Optional.ofNullable(timeout),
+        policy,
+        Objects.requireNonNull(container.getImage(), "Container image must be provided"));
+  }
+
+  @Override
+  public boolean accept(Class<? extends RunTaskConfiguration> clazz) {
+    return RunContainer.class.equals(clazz);
+  }
+}

impl/container/src/main/java/io/serverlessworkflow/impl/container/executors/RunContainerExecutorBuilder.java

@@ -1 +1 @@
-io.serverlessworkflow.impl.container.executors.RunContainerExecutor
+io.serverlessworkflow.impl.container.executors.RunContainerExecutorBuilder

impl/container/src/main/resources/META-INF/services/io.serverlessworkflow.impl.executors.RunnableTask renamed to impl/container/src/main/resources/META-INF/services/io.serverlessworkflow.impl.executors.RunnableTaskBuilder

@@ -45,12 +45,10 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   protected final WorkflowContext workflowContext;
   protected Instant startedAt;
 
-  protected final Map<String, Object> additionalObjects = new ConcurrentHashMap<String, Object>();
-
   protected AtomicReference<CompletableFuture<WorkflowModel>> futureRef = new AtomicReference<>();
   protected Instant completedAt;
 
-  protected final Map<String, Object> additionalObjects = new ConcurrentHashMap<String, Object>();
+  protected final Map<String, Object> additionalObjects = new ConcurrentHashMap<>();
 
   private Lock statusLock = new ReentrantLock();
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
@@ -292,8 +290,4 @@ public <T> T additionalObject(String key, Supplier<T> supplier) {
   }
 
   public void restoreContext(WorkflowContext workflow, TaskContext context) {}
-
-  public <T> T additionalObject(String key, Supplier<T> supplier) {
-    return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());
-  }
 }