[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/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/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/generic/SortedArrayList.java

@@ -36,9 +36,16 @@
 import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.BinaryOperator;
 import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collector;
 
 public class JsonUtils {
 
@@ -48,6 +55,38 @@ public static ObjectMapper mapper() {
     return mapper;
   }
 
+  public static Collector<JsonNode, ArrayNode, ArrayNode> arrayNodeCollector() {
+    return new Collector<JsonNode, ArrayNode, ArrayNode>() {
+      @Override
+      public BiConsumer<ArrayNode, JsonNode> accumulator() {
+        return (arrayNode, item) -> arrayNode.add(item);
+      }
+
+      @Override
+      public Set<Characteristics> characteristics() {
+        return Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));
+      }
+
+      @Override
+      public BinaryOperator<ArrayNode> combiner() {
+        return (r1, r2) -> {
+          r1.addAll(r2);
+          return r1;
+        };
+      }
+
+      @Override
+      public Function<ArrayNode, ArrayNode> finisher() {
+        return arrayNode -> arrayNode;
+      }
+
+      @Override
+      public Supplier<ArrayNode> supplier() {
+        return () -> mapper.createArrayNode();
+      }
+    };
+  }
+
   /*
    * Implementation note:
    * Although we can use directly ObjectMapper.convertValue for implementing fromValue and toJavaValue methods,