refactoring + tests

Signed-off-by: Dmitrii Tikhomirov <chani.liet@gmail.com>

Author: treblereel

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

@@ -17,6 +17,7 @@
 
 import com.github.dockerjava.api.command.CreateContainerCmd;
 import io.serverlessworkflow.api.types.Container;
+import java.util.function.Function;
 
 class CommandPropertySetter extends ContainerPropertySetter {
 
@@ -25,7 +26,7 @@ class CommandPropertySetter extends ContainerPropertySetter {
   }
 
   @Override
-  public void accept(StringExpressionResolver resolver) {
+  public void accept(Function<String, String> resolver) {
     if (configuration.getCommand() != null && !configuration.getCommand().isEmpty()) {
       createContainerCmd.withCmd("sh", "-c", configuration.getCommand());
     }

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

@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 
 class ContainerEnvironmentPropertySetter extends ContainerPropertySetter {
 
@@ -29,15 +30,15 @@ class ContainerEnvironmentPropertySetter extends ContainerPropertySetter {
   }
 
   @Override
-  public void accept(StringExpressionResolver resolver) {
+  public void accept(Function<String, String> resolver) {
     if (!(configuration.getEnvironment() == null
         || configuration.getEnvironment().getAdditionalProperties() == null)) {
       List<String> envs = new ArrayList<>();
       for (Map.Entry<String, Object> entry :
           configuration.getEnvironment().getAdditionalProperties().entrySet()) {
         String key = entry.getKey();
         if (entry.getValue() instanceof String value) {
-          String resolvedValue = resolver.resolve(value);
+          String resolvedValue = resolver.apply(value);
           envs.add(key + "=" + resolvedValue);
         } else {
           throw new IllegalArgumentException("Environment variable values must be strings");

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

@@ -18,8 +18,9 @@
 import com.github.dockerjava.api.command.CreateContainerCmd;
 import io.serverlessworkflow.api.types.Container;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
-abstract class ContainerPropertySetter implements Consumer<StringExpressionResolver> {
+abstract class ContainerPropertySetter implements Consumer<Function<String, String>> {
 
   protected final CreateContainerCmd createContainerCmd;
   protected final Container configuration;

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

@@ -20,8 +20,9 @@
 import com.github.dockerjava.api.DockerClient;
 import com.github.dockerjava.api.command.CreateContainerCmd;
 import com.github.dockerjava.api.command.CreateContainerResponse;
-import com.github.dockerjava.api.command.PullImageResultCallback;
 import com.github.dockerjava.api.command.WaitContainerResultCallback;
+import com.github.dockerjava.api.exception.DockerClientException;
+import com.github.dockerjava.api.exception.NotFoundException;
 import com.github.dockerjava.core.DefaultDockerClientConfig;
 import com.github.dockerjava.core.DockerClientImpl;
 import com.github.dockerjava.core.NameParser;
@@ -38,6 +39,7 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 
 class ContainerRunner {
 
@@ -62,47 +64,45 @@ private ContainerRunner(
     this.propertySetters = new ArrayList<>();
   }
 
-  /**
-   * Blocking container execution according to the lifetime policy. Returns an already completed
-   * CompletableFuture: - completedFuture(input) if exitCode == 0 - exceptionally completed if the
-   * exit code is non-zero or an error occurs. The method blocks the calling thread until the
-   * container finishes or the timeout expires.
-   */
-  CompletableFuture<WorkflowModel> startSync(
+  CompletableFuture<WorkflowModel> start(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
+    return CompletableFuture.supplyAsync(
+        () -> startSync(workflowContext, taskContext, input),
+        definition.application().executorService());
+  }
 
+  private WorkflowModel startSync(
+      WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     try {
       var resolver = new StringExpressionResolver(workflowContext, taskContext, input);
       applyPropertySetters(resolver);
       pullImageIfNeeded(container.getImage());
 
       String id = createAndStartContainer();
       int exit = waitAccordingToLifetime(id, workflowContext, taskContext, input);
-
-      return mapExitCode(exit, input);
+      if (exit == 0) {
+        return input;
+      }
+      throw mapExitCode(exit);
     } catch (InterruptedException ie) {
       Thread.currentThread().interrupt();
-      return failed("Interrupted while waiting for container");
+      throw failed("Container execution failed with exit code " + ie.getMessage());
     } catch (Exception e) {
-      return failed("Container run failed: " + e.getMessage());
+      throw failed("Container execution failed with exit code " + e.getMessage());
     }
   }
 
-  private void applyPropertySetters(StringExpressionResolver resolver) {
-    for (var setter : propertySetters) setter.accept(resolver);
+  private void applyPropertySetters(Function<String, String> resolver) {
+    propertySetters.forEach(setter -> setter.accept(resolver));
   }
 
   private void pullImageIfNeeded(String imageRef) throws InterruptedException {
     NameParser.ReposTag rt = NameParser.parseRepositoryTag(imageRef);
     NameParser.HostnameReposName hr = NameParser.resolveRepositoryName(imageRef);
 
     String repository = hr.reposName;
-    String tag = rt.tag != null && rt.tag.isEmpty() ? rt.tag : "latest";
-    dockerClient
-        .pullImageCmd(repository)
-        .withTag(tag)
-        .exec(new PullImageResultCallback())
-        .awaitCompletion();
+    String tag = (rt.tag == null || rt.tag.isBlank()) ? "latest" : rt.tag;
+    dockerClient.pullImageCmd(repository).withTag(tag).start().awaitCompletion();
   }
 
   private String createAndStartContainer() {
@@ -123,25 +123,22 @@ private int waitAccordingToLifetime(
     var policy = lifetime != null ? lifetime.getCleanup() : null;
 
     try (var cb = dockerClient.waitContainerCmd(id).exec(new WaitContainerResultCallback())) {
-
       if (policy == ContainerCleanupPolicy.EVENTUALLY) {
         Duration timeout = resolveAfter(lifetime, workflowContext, taskContext, input);
-        int exit = cb.awaitStatusCode(timeout.toMillis(), TimeUnit.MILLISECONDS);
-
-        if (isRunning(id)) {
-          safeStop(id, Duration.ofSeconds(10));
+        try {
+          Integer exit = cb.awaitStatusCode(timeout.toMillis(), TimeUnit.MILLISECONDS);
+          safeStop(id);
+          return exit != null ? exit : 0;
+        } catch (DockerClientException timeoutOrOther) {
+          safeStop(id);
         }
-        safeRemove(id);
-        return exit;
-
       } else {
-        int exit = cb.awaitStatusCode();
-        if (policy == ContainerCleanupPolicy.ALWAYS) {
-          safeRemove(id);
-        }
-        return exit;
+        return cb.awaitStatusCode();
       }
+    } catch (NotFoundException e) {
+      // container already removed
     }
+    return 0;
   }
 
   private Duration resolveAfter(
@@ -167,6 +164,20 @@ private boolean isRunning(String id) {
     }
   }
 
+  private void safeStop(String id) {
+    if (isRunning(id)) {
+      safeStop(id, Duration.ofSeconds(10));
+      try (var cb2 = dockerClient.waitContainerCmd(id).exec(new WaitContainerResultCallback())) {
+        cb2.awaitStatusCode();
+        safeRemove(id);
+      } catch (Exception ignore) {
+        // we can ignore this
+      }
+    } else {
+      safeRemove(id);
+    }
+  }
+
   private void safeStop(String id, Duration timeout) {
     try {
       dockerClient.stopContainerCmd(id).withTimeout((int) Math.max(1, timeout.toSeconds())).exec();
@@ -184,9 +195,8 @@ private void safeRemove(String id) {
     }
   }
 
-  private static <T> CompletableFuture<T> mapExitCode(int exit, T ok) {
+  private static Exception mapExitCode(int exit) {
     return switch (exit) {
-      case 0 -> CompletableFuture.completedFuture(ok);
       case 1 -> failed("General error (exit code 1)");
       case 2 -> failed("Shell syntax error (exit code 2)");
       case 126 -> failed("Command found but not executable (exit code 126)");
@@ -199,10 +209,8 @@ private static <T> CompletableFuture<T> mapExitCode(int exit, T ok) {
     };
   }
 
-  private static <T> CompletableFuture<T> failed(String message) {
-    CompletableFuture<T> f = new CompletableFuture<>();
-    f.completeExceptionally(new RuntimeException(message));
-    return f;
+  private static RuntimeException failed(String message) {
+    return new RuntimeException(message);
   }
 
   static ContainerRunnerBuilder builder() {

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

@@ -20,6 +20,7 @@
 import com.github.dockerjava.api.command.CreateContainerCmd;
 import io.serverlessworkflow.api.types.Container;
 import io.serverlessworkflow.api.types.ContainerLifetime;
+import java.util.function.Function;
 
 class LifetimePropertySetter extends ContainerPropertySetter {
 
@@ -28,7 +29,7 @@ class LifetimePropertySetter extends ContainerPropertySetter {
   }
 
   @Override
-  public void accept(StringExpressionResolver resolver) {
+  public void accept(Function<String, String> resolver) {
     // case of cleanup=eventually processed at ContainerRunner
     if (configuration.getLifetime() != null) {
       ContainerLifetime lifetime = configuration.getLifetime();

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

@@ -17,6 +17,7 @@
 
 import com.github.dockerjava.api.command.CreateContainerCmd;
 import io.serverlessworkflow.api.types.Container;
+import java.util.function.Function;
 
 class NamePropertySetter extends ContainerPropertySetter {
 
@@ -25,9 +26,9 @@ class NamePropertySetter extends ContainerPropertySetter {
   }
 
   @Override
-  public void accept(StringExpressionResolver resolver) {
+  public void accept(Function<String, String> resolver) {
     if (configuration.getName() != null && !configuration.getName().isEmpty()) {
-      String resolvedName = resolver.resolve(configuration.getName());
+      String resolvedName = resolver.apply(configuration.getName());
       createContainerCmd.withName(resolvedName);
     }
   }

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

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 
 class PortsPropertySetter extends ContainerPropertySetter {
 
@@ -30,7 +31,7 @@ class PortsPropertySetter extends ContainerPropertySetter {
   }
 
   @Override
-  public void accept(StringExpressionResolver resolver) {
+  public void accept(Function<String, String> resolver) {
     if (configuration.getPorts() != null
         && configuration.getPorts().getAdditionalProperties() != null) {
       Ports portBindings = new Ports();

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

@@ -41,7 +41,7 @@ public void init(RunContainer taskConfiguration, WorkflowDefinition definition)
   @Override
   public CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
-    return containerRunner.startSync(workflowContext, taskContext, input);
+    return containerRunner.start(workflowContext, taskContext, input);
   }
 
   @Override

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

@@ -20,8 +20,9 @@
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowUtils;
 import io.serverlessworkflow.impl.expressions.ExpressionUtils;
+import java.util.function.Function;
 
-class StringExpressionResolver {
+class StringExpressionResolver implements Function<String, String> {
 
   private final WorkflowContext workflowContext;
   private final TaskContext taskContext;
@@ -34,9 +35,9 @@ class StringExpressionResolver {
     this.input = input;
   }
 
-  String resolve(String expression) {
+  public String apply(String expression) {
     if (ExpressionUtils.isExpr(expression)) {
-      WorkflowUtils.buildStringResolver(
+      return WorkflowUtils.buildStringResolver(
               workflowContext.definition().application(),
               expression,
               taskContext.input().asJavaObject())

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

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 
 class VolumesPropertySetter extends ContainerPropertySetter {
 
@@ -30,16 +31,16 @@ class VolumesPropertySetter extends ContainerPropertySetter {
   }
 
   @Override
-  public void accept(StringExpressionResolver resolver) {
+  public void accept(Function<String, String> resolver) {
     if (configuration.getVolumes() != null
         && configuration.getVolumes().getAdditionalProperties() != null) {
       List<Bind> binds = new ArrayList<>();
       for (Map.Entry<String, Object> entry :
           configuration.getVolumes().getAdditionalProperties().entrySet()) {
         String hostPath = entry.getKey();
         if (entry.getValue() instanceof String containerPath) {
-          String resolvedHostPath = resolver.resolve(hostPath);
-          String resolvedContainerPath = resolver.resolve(containerPath);
+          String resolvedHostPath = resolver.apply(hostPath);
+          String resolvedContainerPath = resolver.apply(containerPath);
           binds.add(new Bind(resolvedHostPath, new Volume(resolvedContainerPath)));
         } else {
           throw new IllegalArgumentException("Volume container paths must be strings");