[Fix #484] Status related changes

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,11 +19,12 @@
 
 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;
 
 public class WorkflowInstance {
-  private final AtomicReference<WorkflowState> state;
+  private final AtomicReference<WorkflowStatus> status;
   private final TaskContext<?> taskContext;
   private final String id;
   private final JsonNode input;
@@ -40,15 +41,16 @@ public class WorkflowInstance {
     definition
         .inputFilter()
         .ifPresent(f -> taskContext.input(f.apply(workflowContext, taskContext, input)));
-    state = new AtomicReference<>(WorkflowState.STARTED);
+    status = new AtomicReference<>(WorkflowStatus.RUNNING);
     context = new AtomicReference<>(NullNode.getInstance());
-    WorkflowUtils.processTaskList(definition.workflow().getDo(), workflowContext, taskContext);
+    TaskExecutorHelper.processTaskList(definition.workflow().getDo(), workflowContext, taskContext);
     definition
         .outputFilter()
         .ifPresent(
             f ->
                 taskContext.output(f.apply(workflowContext, taskContext, taskContext.rawOutput())));
     definition.outputSchemaValidator().ifPresent(v -> v.validate(taskContext.output()));
+    status.compareAndSet(WorkflowStatus.RUNNING, WorkflowStatus.COMPLETED);
   }
 
   public String id() {
@@ -67,12 +69,12 @@ public JsonNode context() {
     return context.get();
   }
 
-  public WorkflowState state() {
-    return state.get();
+  public WorkflowStatus status() {
+    return status.get();
   }
 
-  public void state(WorkflowState state) {
-    this.state.set(state);
+  public void status(WorkflowStatus state) {
+    this.status.set(state);
   }
 
   public Object output() {

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowState.java renamed to impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowStatus.java

@@ -15,8 +15,11 @@
  */
 package io.serverlessworkflow.impl;
 
-public enum WorkflowState {
-  STARTED,
+public enum WorkflowStatus {
+  PENDING,
+  RUNNING,
   WAITING,
-  COMPLETED
+  COMPLETED,
+  FAULTED,
+  CANCELLED
 }

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,11 +43,12 @@ 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) {
+  private static Optional<JsonNode> schemaToNode(
+      ResourceLoader resourceLoader, SchemaUnion schema) {
     if (schema != null) {
       if (schema.getSchemaInline() != null) {
         SchemaInline inline = schema.getSchemaInline();
@@ -94,18 +90,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 +124,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) {