[Fix #484] Removing SortedArrayList

Maybe for larger list that has to be kept sorte that one is useful, but
in this case is faster to add everything unsorted (all insertions are
really fast) and then sort (there is only one sort of a small amount of
items)

Signed-off-by: Francisco Javier Tirado Sarti <[email protected]>

Author: fjtirado

impl/core/pom.xml

@@ -50,5 +50,10 @@
       <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

impl/core/src/main/java/io/serverlessworkflow/impl/LongFilter.java

@@ -0,0 +1,21 @@
+/*
+ * 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;
+
+import java.util.function.BiFunction;
+
+@FunctionalInterface
+public interface LongFilter extends BiFunction<WorkflowContext, TaskContext<?>, Long> {}

impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java

@@ -41,19 +41,6 @@ public TaskContext(JsonNode input, WorkflowPosition position) {
     this(input, null, position, Instant.now(), input, input, input, null, new HashMap<>());
   }
 
-  public TaskContext<T> copy() {
-    return new TaskContext<T>(
-        rawInput,
-        task,
-        position.copy(),
-        startedAt,
-        input,
-        output,
-        rawOutput,
-        flowDirective,
-        new HashMap<>(contextVariables));
-  }
-
   public TaskContext(JsonNode input, TaskContext<?> taskContext, T task) {
     this(
         input,
@@ -88,6 +75,19 @@ private TaskContext(
     this.contextVariables = contextVariables;
   }
 
+  public TaskContext<T> copy() {
+    return new TaskContext<T>(
+        rawInput,
+        task,
+        position.copy(),
+        startedAt,
+        input,
+        output,
+        rawOutput,
+        flowDirective,
+        new HashMap<>(contextVariables));
+  }
+
   public void input(JsonNode input) {
     this.input = input;
     this.rawOutput = input;

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java

@@ -94,18 +94,22 @@ public static Optional<WorkflowFilter> buildWorkflowFilter(
 
   public static StringFilter buildStringFilter(
       ExpressionFactory exprFactory, String expression, String literal) {
-    return expression != null ? from(buildWorkflowFilter(exprFactory, expression)) : from(literal);
+    return expression != null
+        ? toString(buildWorkflowFilter(exprFactory, expression))
+        : toString(literal);
   }
 
   public static StringFilter buildStringFilter(ExpressionFactory exprFactory, String str) {
-    return ExpressionUtils.isExpr(str) ? from(buildWorkflowFilter(exprFactory, str)) : from(str);
+    return ExpressionUtils.isExpr(str)
+        ? toString(buildWorkflowFilter(exprFactory, str))
+        : toString(str);
   }
 
-  public static StringFilter from(WorkflowFilter filter) {
+  public static StringFilter toString(WorkflowFilter filter) {
     return (w, t) -> filter.apply(w, t, t.input()).asText();
   }
 
-  private static StringFilter from(String literal) {
+  private static StringFilter toString(String literal) {
     return (w, t) -> literal;
   }
 
@@ -124,6 +128,21 @@ private static WorkflowFilter buildWorkflowFilter(
     throw new IllegalStateException("Both object and str are null");
   }
 
+  public static LongFilter buildLongFilter(
+      ExpressionFactory exprFactory, String expression, Long literal) {
+    return expression != null
+        ? toLong(buildWorkflowFilter(exprFactory, expression))
+        : toLong(literal);
+  }
+
+  public static LongFilter toLong(WorkflowFilter filter) {
+    return (w, t) -> filter.apply(w, t, t.input()).asLong();
+  }
+
+  private static LongFilter toLong(Long literal) {
+    return (w, t) -> literal;
+  }
+
   private static TaskItem findTaskByName(ListIterator<TaskItem> iter, String taskName) {
     int currentIndex = iter.nextIndex();
     while (iter.hasPrevious()) {

impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java

@@ -73,6 +73,8 @@ public TaskExecutor<? extends TaskBase> getTaskExecutor(
       return new TryExecutor(task.getTryTask(), definition);
     } else if (task.getForkTask() != null) {
       return new ForkExecutor(task.getForkTask(), definition);
+    } else if (task.getWaitTask() != null) {
+      return new WaitExecutor(task.getWaitTask(), definition);
     }
     throw new UnsupportedOperationException(task.get().getClass().getName() + " not supported yet");
   }

impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForkExecutor.java

@@ -15,6 +15,7 @@
  */
 package io.serverlessworkflow.impl.executors;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.api.types.ForkTask;
 import io.serverlessworkflow.api.types.ForkTaskConfiguration;
@@ -24,16 +25,16 @@
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowState;
 import io.serverlessworkflow.impl.WorkflowUtils;
-import io.serverlessworkflow.impl.generic.SortedArrayList;
 import io.serverlessworkflow.impl.json.JsonUtils;
 import java.lang.reflect.UndeclaredThrowableException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -47,8 +48,6 @@ protected ForkExecutor(ForkTask task, WorkflowDefinition definition) {
     service = definition.executorService();
   }
 
-  private record BranchContext(String taskName, TaskContext<?> taskContext) {}
-
   @Override
   protected void internalExecute(WorkflowContext workflow, TaskContext<ForkTask> taskContext) {
     ForkTaskConfiguration forkConfig = task.getFork();
@@ -62,13 +61,10 @@ protected void internalExecute(WorkflowContext workflow, TaskContext<ForkTask> t
             item.getName(),
             service.submit(() -> executeBranch(workflow, taskContext.copy(), item, i)));
       }
-      List<BranchContext> results =
-          new SortedArrayList<>(
-              (arg1, arg2) ->
-                  arg1.taskContext.completedAt().compareTo(arg2.taskContext.completedAt()));
+      List<Map.Entry<String, TaskContext<?>>> results = new ArrayList<>();
       for (Map.Entry<String, Future<TaskContext<?>>> entry : futures.entrySet()) {
         try {
-          results.add(new BranchContext(entry.getKey(), entry.getValue().get()));
+          results.add(Map.entry(entry.getKey(), entry.getValue().get()));
         } catch (ExecutionException ex) {
           Throwable cause = ex.getCause();
           if (cause instanceof RuntimeException) {
@@ -77,24 +73,25 @@ protected void internalExecute(WorkflowContext workflow, TaskContext<ForkTask> t
             throw new UndeclaredThrowableException(ex);
           }
         } catch (InterruptedException ex) {
-          logger.warn(
-              "Thred executing branch {} was interrupted, this branch will be ignored",
-              entry.getKey(),
-              ex);
+          logger.warn("Branch {} was interrupted, no result will be recorded", entry.getKey(), ex);
         }
       }
       if (!results.isEmpty()) {
+        Stream<Map.Entry<String, TaskContext<?>>> sortedStream =
+            results.stream()
+                .sorted(
+                    (arg1, arg2) ->
+                        arg1.getValue().completedAt().compareTo(arg2.getValue().completedAt()));
         taskContext.rawOutput(
             forkConfig.isCompete()
-                ? results.get(0).taskContext().output()
-                : JsonUtils.fromValue(
-                    results.stream()
-                        .map(
-                            e ->
-                                JsonUtils.mapper()
-                                    .createObjectNode()
-                                    .set(e.taskName(), e.taskContext().output()))
-                        .collect(Collectors.toList())));
+                ? sortedStream.map(e -> e.getValue().output()).findFirst().orElseThrow()
+                : sortedStream
+                    .<JsonNode>map(
+                        e ->
+                            JsonUtils.mapper()
+                                .createObjectNode()
+                                .set(e.getKey(), e.getValue().output()))
+                    .collect(JsonUtils.arrayNodeCollector()));
       }
     }
   }

impl/core/src/main/java/io/serverlessworkflow/impl/executors/WaitExecutor.java

@@ -0,0 +1,58 @@
+/*
+ * 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.executors;
+
+import io.serverlessworkflow.api.types.DurationInline;
+import io.serverlessworkflow.api.types.WaitTask;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import java.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class WaitExecutor extends AbstractTaskExecutor<WaitTask> {
+
+  private static Logger logger = LoggerFactory.getLogger(WaitExecutor.class);
+  private final Duration millisToWait;
+
+  protected WaitExecutor(WaitTask task, WorkflowDefinition definition) {
+    super(task, definition);
+    this.millisToWait =
+        task.getWait().getDurationInline() != null
+            ? toLong(task.getWait().getDurationInline())
+            : Duration.parse(task.getWait().getDurationExpression());
+  }
+
+  private Duration toLong(DurationInline durationInline) {
+    Duration duration = Duration.ofMillis(durationInline.getMilliseconds());
+    duration.plus(Duration.ofSeconds(durationInline.getSeconds()));
+    duration.plus(Duration.ofMinutes(durationInline.getMinutes()));
+    duration.plus(Duration.ofHours(durationInline.getHours()));
+    duration.plus(Duration.ofDays(durationInline.getDays()));
+    return duration;
+  }
+
+  @Override
+  protected void internalExecute(WorkflowContext workflow, TaskContext<WaitTask> taskContext) {
+    try {
+      Thread.sleep(millisToWait.toMillis());
+    } catch (InterruptedException e) {
+      logger.warn("Waiting thread was interrupted", e);
+      Thread.currentThread().interrupt();
+    }
+  }
+}