Merge pull request #489 from fjtirado/Fix_#484

[Fix #484] Execute Fork task

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

@@ -0,0 +1,22 @@
+/*
+ * 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.concurrent.ExecutorService;
+import java.util.function.Supplier;
+
+@FunctionalInterface
+public interface ExecutorServiceFactory extends Supplier<ExecutorService> {}

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

@@ -54,7 +54,7 @@ public String jsonPointer() {
 
   @Override
   public String toString() {
-    return "ListWorkflowPosition [list=" + queue + "]";
+    return "QueueWorkflowPosition [queue=" + queue + "]";
   }
 
   @Override
@@ -65,6 +65,6 @@ public WorkflowPosition back() {
 
   @Override
   public Object last() {
-    return queue.pollLast();
+    return queue.getLast();
   }
 }

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

@@ -28,35 +28,64 @@ public class TaskContext<T extends TaskBase> {
   private final JsonNode rawInput;
   private final T task;
   private final WorkflowPosition position;
-  private final Instant startedAt = Instant.now();
+  private final Instant startedAt;
 
   private JsonNode input;
   private JsonNode output;
   private JsonNode rawOutput;
   private FlowDirective flowDirective;
   private Map<String, Object> contextVariables;
+  private Instant completedAt;
 
   public TaskContext(JsonNode input, WorkflowPosition position) {
-    this.rawInput = input;
-    this.position = position;
-    this.task = null;
-    this.contextVariables = new HashMap<>();
-    init();
+    this(input, null, position, Instant.now(), input, input, input, null, new HashMap<>());
   }
 
   public TaskContext(JsonNode input, TaskContext<?> taskContext, T task) {
-    this.rawInput = input;
-    this.position = taskContext.position.copy();
+    this(
+        input,
+        task,
+        taskContext.position,
+        Instant.now(),
+        input,
+        input,
+        input,
+        task.getThen(),
+        new HashMap<>(taskContext.variables()));
+  }
+
+  private TaskContext(
+      JsonNode rawInput,
+      T task,
+      WorkflowPosition position,
+      Instant startedAt,
+      JsonNode input,
+      JsonNode output,
+      JsonNode rawOutput,
+      FlowDirective flowDirective,
+      Map<String, Object> contextVariables) {
+    this.rawInput = rawInput;
     this.task = task;
-    this.flowDirective = task.getThen();
-    this.contextVariables = new HashMap<>(taskContext.variables());
-    init();
+    this.position = position;
+    this.startedAt = startedAt;
+    this.input = input;
+    this.output = output;
+    this.rawOutput = rawOutput;
+    this.flowDirective = flowDirective;
+    this.contextVariables = contextVariables;
   }
 
-  private void init() {
-    this.input = rawInput;
-    this.rawOutput = rawInput;
-    this.output = rawInput;
+  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) {
@@ -115,4 +144,12 @@ public WorkflowPosition position() {
   public Instant startedAt() {
     return startedAt;
   }
+
+  public void completedAt(Instant instant) {
+    this.completedAt = instant;
+  }
+
+  public Instant completedAt() {
+    return completedAt;
+  }
 }

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

@@ -32,6 +32,8 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 public class WorkflowApplication implements AutoCloseable {
 
@@ -43,8 +45,11 @@ public class WorkflowApplication implements AutoCloseable {
   private final Collection<WorkflowExecutionListener> listeners;
   private final Map<WorkflowId, WorkflowDefinition> definitions;
   private final WorkflowPositionFactory positionFactory;
+  private final ExecutorServiceFactory executorFactory;
   private final RuntimeDescriptorFactory runtimeDescriptorFactory;
 
+  private ExecutorService executorService;
+
   public WorkflowApplication(
       TaskExecutorFactory taskFactory,
       ExpressionFactory exprFactory,
@@ -53,6 +58,7 @@ public WorkflowApplication(
       WorkflowPositionFactory positionFactory,
       WorkflowIdFactory idFactory,
       RuntimeDescriptorFactory runtimeDescriptorFactory,
+      ExecutorServiceFactory executorFactory,
       Collection<WorkflowExecutionListener> listeners) {
     this.taskFactory = taskFactory;
     this.exprFactory = exprFactory;
@@ -61,6 +67,7 @@ public WorkflowApplication(
     this.positionFactory = positionFactory;
     this.idFactory = idFactory;
     this.runtimeDescriptorFactory = runtimeDescriptorFactory;
+    this.executorFactory = executorFactory;
     this.listeners = listeners;
     this.definitions = new ConcurrentHashMap<>();
   }
@@ -101,6 +108,7 @@ public static class Builder {
     private SchemaValidatorFactory schemaValidatorFactory = DefaultSchemaValidatorFactory.get();
     private WorkflowPositionFactory positionFactory = () -> new QueueWorkflowPosition();
     private WorkflowIdFactory idFactory = () -> UlidCreator.getMonotonicUlid().toString();
+    private ExecutorServiceFactory executorFactory = () -> Executors.newCachedThreadPool();
     private RuntimeDescriptorFactory descriptorFactory =
         () -> new RuntimeDescriptor("reference impl", "1.0.0_alpha", Collections.emptyMap());
 
@@ -129,6 +137,11 @@ public Builder withResourceLoaderFactory(ResourceLoaderFactory resourceLoader) {
       return this;
     }
 
+    public Builder withExecutorFactory(ExecutorServiceFactory executorFactory) {
+      this.executorFactory = executorFactory;
+      return this;
+    }
+
     public Builder withPositionFactory(WorkflowPositionFactory positionFactory) {
       this.positionFactory = positionFactory;
       return this;
@@ -158,6 +171,7 @@ public WorkflowApplication build() {
           positionFactory,
           idFactory,
           descriptorFactory,
+          executorFactory,
           listeners == null
               ? Collections.emptySet()
               : Collections.unmodifiableCollection(listeners));
@@ -190,4 +204,13 @@ public WorkflowPositionFactory positionFactory() {
   public RuntimeDescriptorFactory runtimeDescriptorFactory() {
     return runtimeDescriptorFactory;
   }
+
+  public ExecutorService executorService() {
+    synchronized (executorFactory) {
+      if (executorService == null) {
+        executorService = executorFactory.get();
+      }
+    }
+    return executorService;
+  }
 }

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

@@ -33,6 +33,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
 
 public class WorkflowDefinition implements AutoCloseable {
 
@@ -48,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());
     }
   }
@@ -134,6 +132,10 @@ public WorkflowPositionFactory positionFactory() {
     return application.positionFactory();
   }
 
+  public ExecutorService executorService() {
+    return application.executorService();
+  }
+
   public RuntimeDescriptorFactory runtimeDescriptorFactory() {
     return application.runtimeDescriptorFactory();
   }

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

@@ -15,11 +15,11 @@
  */
 package io.serverlessworkflow.impl;
 
-import io.serverlessworkflow.api.types.Task;
+import io.serverlessworkflow.api.types.TaskBase;
 
 public interface WorkflowExecutionListener {
 
-  void onTaskStarted(WorkflowPosition currentPos, Task task);
+  void onTaskStarted(WorkflowPosition currentPos, TaskBase task);
 
-  void onTaskEnded(WorkflowPosition currentPos, Task task);
+  void onTaskEnded(WorkflowPosition currentPos, TaskBase task);
 }

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

@@ -19,15 +19,17 @@
 
 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 WorkflowState state;
-  private TaskContext<?> taskContext;
+  private final AtomicReference<WorkflowStatus> status;
+  private final TaskContext<?> taskContext;
   private final String id;
   private final JsonNode input;
   private final Instant startedAt;
-  private JsonNode context = NullNode.getInstance();
+  private final AtomicReference<JsonNode> context;
 
   WorkflowInstance(WorkflowDefinition definition, JsonNode input) {
     this.id = definition.idFactory().get();
@@ -39,14 +41,16 @@ public class WorkflowInstance {
     definition
         .inputFilter()
         .ifPresent(f -> taskContext.input(f.apply(workflowContext, taskContext, input)));
-    state = WorkflowState.STARTED;
-    WorkflowUtils.processTaskList(definition.workflow().getDo(), workflowContext, taskContext);
+    status = new AtomicReference<>(WorkflowStatus.RUNNING);
+    context = new AtomicReference<>(NullNode.getInstance());
+    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() {
@@ -62,22 +66,26 @@ public JsonNode input() {
   }
 
   public JsonNode context() {
-    return context;
+    return context.get();
   }
 
-  public WorkflowState state() {
-    return state;
+  public WorkflowStatus status() {
+    return status.get();
+  }
+
+  public void status(WorkflowStatus state) {
+    this.status.set(state);
   }
 
   public Object output() {
     return toJavaValue(taskContext.output());
   }
 
-  public Object outputAsJsonNode() {
+  public JsonNode outputAsJsonNode() {
     return taskContext.output();
   }
 
   void context(JsonNode context) {
-    this.context = context;
+    this.context.set(context);
   }
 }

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
 }