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

@@ -49,22 +49,19 @@ public class WorkflowDefinition implements AutoCloseable {
 
   private WorkflowDefinition(
       WorkflowApplication application, Workflow workflow, ResourceLoader resourceLoader) {
-
     this.workflow = workflow;
     this.application = application;
     this.resourceLoader = resourceLoader;
     if (workflow.getInput() != null) {
       Input input = workflow.getInput();
       this.inputSchemaValidator =
-          getSchemaValidator(
-              application.validatorFactory(), schemaToNode(resourceLoader, input.getSchema()));
+          getSchemaValidator(application.validatorFactory(), resourceLoader, input.getSchema());
       this.inputFilter = buildWorkflowFilter(application.expressionFactory(), input.getFrom());
     }
     if (workflow.getOutput() != null) {
       Output output = workflow.getOutput();
       this.outputSchemaValidator =
-          getSchemaValidator(
-              application.validatorFactory(), schemaToNode(resourceLoader, output.getSchema()));
+          getSchemaValidator(application.validatorFactory(), resourceLoader, output.getSchema());
       this.outputFilter = buildWorkflowFilter(application.expressionFactory(), output.getAs());
     }
   }

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

@@ -19,6 +19,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.NullNode;
+import io.serverlessworkflow.impl.executors.TaskExecutorHelper;
 import java.time.Instant;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -42,7 +43,7 @@ public class WorkflowInstance {
         .ifPresent(f -> taskContext.input(f.apply(workflowContext, taskContext, input)));
     state = new AtomicReference<>(WorkflowState.STARTED);
     context = new AtomicReference<>(NullNode.getInstance());
-    WorkflowUtils.processTaskList(definition.workflow().getDo(), workflowContext, taskContext);
+    TaskExecutorHelper.processTaskList(definition.workflow().getDo(), workflowContext, taskContext);
     definition
         .outputFilter()
         .ifPresent(

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

@@ -19,14 +19,11 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.serverlessworkflow.api.WorkflowFormat;
 import io.serverlessworkflow.api.types.ExportAs;
-import io.serverlessworkflow.api.types.FlowDirective;
 import io.serverlessworkflow.api.types.InputFrom;
 import io.serverlessworkflow.api.types.OutputAs;
 import io.serverlessworkflow.api.types.SchemaExternal;
 import io.serverlessworkflow.api.types.SchemaInline;
 import io.serverlessworkflow.api.types.SchemaUnion;
-import io.serverlessworkflow.api.types.TaskBase;
-import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.impl.expressions.Expression;
 import io.serverlessworkflow.impl.expressions.ExpressionFactory;
 import io.serverlessworkflow.impl.expressions.ExpressionUtils;
@@ -38,8 +35,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
-import java.util.List;
-import java.util.ListIterator;
 import java.util.Map;
 import java.util.Optional;
 
@@ -48,8 +43,8 @@ public class WorkflowUtils {
   private WorkflowUtils() {}
 
   public static Optional<SchemaValidator> getSchemaValidator(
-      SchemaValidatorFactory validatorFactory, Optional<JsonNode> node) {
-    return node.map(n -> validatorFactory.getValidator(n));
+      SchemaValidatorFactory validatorFactory, ResourceLoader resourceLoader, SchemaUnion schema) {
+    return schemaToNode(resourceLoader, schema).map(n -> validatorFactory.getValidator(n));
   }
 
   public static Optional<JsonNode> schemaToNode(ResourceLoader resourceLoader, SchemaUnion schema) {
@@ -94,18 +89,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) {
+  private 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,76 +123,19 @@ private static WorkflowFilter buildWorkflowFilter(
     throw new IllegalStateException("Both object and str are null");
   }
 
-  private static TaskItem findTaskByName(ListIterator<TaskItem> iter, String taskName) {
-    int currentIndex = iter.nextIndex();
-    while (iter.hasPrevious()) {
-      TaskItem item = iter.previous();
-      if (item.getName().equals(taskName)) {
-        return item;
-      }
-    }
-    while (iter.nextIndex() < currentIndex) {
-      iter.next();
-    }
-    while (iter.hasNext()) {
-      TaskItem item = iter.next();
-      if (item.getName().equals(taskName)) {
-        return item;
-      }
-    }
-    throw new IllegalArgumentException("Cannot find task with name " + taskName);
+  public static LongFilter buildLongFilter(
+      ExpressionFactory exprFactory, String expression, Long literal) {
+    return expression != null
+        ? toLong(buildWorkflowFilter(exprFactory, expression))
+        : toLong(literal);
   }
 
-  public static void processTaskList(
-      List<TaskItem> tasks, WorkflowContext context, TaskContext<?> parentTask) {
-    parentTask.position().addProperty("do");
-    TaskContext<? extends TaskBase> currentContext = parentTask;
-    if (!tasks.isEmpty()) {
-      ListIterator<TaskItem> iter = tasks.listIterator();
-      TaskItem nextTask = iter.next();
-      while (nextTask != null) {
-        TaskItem task = nextTask;
-        parentTask.position().addIndex(iter.previousIndex());
-        currentContext = executeTask(context, parentTask, task, currentContext.output());
-        FlowDirective flowDirective = currentContext.flowDirective();
-        if (flowDirective.getFlowDirectiveEnum() != null) {
-          switch (flowDirective.getFlowDirectiveEnum()) {
-            case CONTINUE:
-              nextTask = iter.hasNext() ? iter.next() : null;
-              break;
-            case END:
-              context.instance().state(WorkflowState.COMPLETED);
-            case EXIT:
-              nextTask = null;
-              break;
-          }
-        } else {
-          nextTask = WorkflowUtils.findTaskByName(iter, flowDirective.getString());
-        }
-        parentTask.position().back();
-      }
-    }
-    parentTask.position().back();
-    parentTask.rawOutput(currentContext.output());
+  private static LongFilter toLong(WorkflowFilter filter) {
+    return (w, t) -> filter.apply(w, t, t.input()).asLong();
   }
 
-  public static TaskContext<?> executeTask(
-      WorkflowContext context, TaskContext<?> parentTask, TaskItem task, JsonNode input) {
-    parentTask.position().addProperty(task.getName());
-    TaskContext<?> result =
-        context
-            .definition()
-            .taskExecutors()
-            .computeIfAbsent(
-                parentTask.position().jsonPointer(),
-                k ->
-                    context
-                        .definition()
-                        .taskFactory()
-                        .getTaskExecutor(task.getTask(), context.definition()))
-            .apply(context, parentTask, input);
-    parentTask.position().back();
-    return result;
+  private static LongFilter toLong(Long literal) {
+    return (w, t) -> literal;
   }
 
   public static WorkflowFilter buildWorkflowFilter(ExpressionFactory exprFactory, String str) {

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

@@ -55,8 +55,7 @@ private void buildInputProcessors(WorkflowDefinition definition) {
       this.inputProcessor = buildWorkflowFilter(definition.expressionFactory(), input.getFrom());
       this.inputSchemaValidator =
           getSchemaValidator(
-              definition.validatorFactory(),
-              schemaToNode(definition.resourceLoader(), input.getSchema()));
+              definition.validatorFactory(), definition.resourceLoader(), input.getSchema());
     }
   }
 
@@ -66,8 +65,7 @@ private void buildOutputProcessors(WorkflowDefinition definition) {
       this.outputProcessor = buildWorkflowFilter(definition.expressionFactory(), output.getAs());
       this.outputSchemaValidator =
           getSchemaValidator(
-              definition.validatorFactory(),
-              schemaToNode(definition.resourceLoader(), output.getSchema()));
+              definition.validatorFactory(), definition.resourceLoader(), output.getSchema());
     }
   }
 
@@ -79,8 +77,7 @@ private void buildContextProcessors(WorkflowDefinition definition) {
       }
       this.contextSchemaValidator =
           getSchemaValidator(
-              definition.validatorFactory(),
-              schemaToNode(definition.resourceLoader(), export.getSchema()));
+              definition.validatorFactory(), definition.resourceLoader(), export.getSchema());
     }
   }
 

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/DoExecutor.java

@@ -19,7 +19,6 @@
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
-import io.serverlessworkflow.impl.WorkflowUtils;
 
 public class DoExecutor extends AbstractTaskExecutor<DoTask> {
 
@@ -29,6 +28,6 @@ protected DoExecutor(DoTask task, WorkflowDefinition definition) {
 
   @Override
   protected void internalExecute(WorkflowContext workflow, TaskContext<DoTask> taskContext) {
-    WorkflowUtils.processTaskList(task.getDo(), workflow, taskContext);
+    TaskExecutorHelper.processTaskList(task.getDo(), workflow, taskContext);
   }
 }

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

@@ -52,7 +52,7 @@ protected void internalExecute(WorkflowContext workflow, TaskContext<ForTask> ta
       JsonNode item = iter.next();
       taskContext.variables().put(task.getFor().getEach(), item);
       taskContext.variables().put(task.getFor().getAt(), i++);
-      WorkflowUtils.processTaskList(task.getDo(), workflow, taskContext);
+      TaskExecutorHelper.processTaskList(task.getDo(), workflow, taskContext);
     }
   }
 }