[Fix #693] Adding lifecycle events

[Fix #693] CloudEvent adapter for life cycle

Signed-off-by: fjtirado <[email protected]>

Author: fjtirado

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaForExecutorBuilder.java

@@ -23,7 +23,7 @@
 import io.serverlessworkflow.api.types.func.ForTaskFunction;
 import io.serverlessworkflow.api.types.func.TypedFunction;
 import io.serverlessworkflow.impl.WorkflowApplication;
-import io.serverlessworkflow.impl.WorkflowPosition;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.WorkflowValueResolver;
 import io.serverlessworkflow.impl.executors.ForExecutor.ForExecutorBuilder;
@@ -36,7 +36,7 @@
 public class JavaForExecutorBuilder extends ForExecutorBuilder {
 
   protected JavaForExecutorBuilder(
-      WorkflowPosition position,
+      WorkflowMutablePosition position,
       ForTask task,
       Workflow workflow,
       WorkflowApplication application,

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaSwitchExecutorBuilder.java

@@ -22,7 +22,7 @@
 import io.serverlessworkflow.api.types.func.SwitchCaseFunction;
 import io.serverlessworkflow.api.types.func.TypedPredicate;
 import io.serverlessworkflow.impl.WorkflowApplication;
-import io.serverlessworkflow.impl.WorkflowPosition;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.executors.SwitchExecutor.SwitchExecutorBuilder;
 import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
@@ -33,7 +33,7 @@
 public class JavaSwitchExecutorBuilder extends SwitchExecutorBuilder {
 
   protected JavaSwitchExecutorBuilder(
-      WorkflowPosition position,
+      WorkflowMutablePosition position,
       SwitchTask task,
       Workflow workflow,
       WorkflowApplication application,

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaTaskExecutorFactory.java

@@ -19,15 +19,15 @@
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.WorkflowApplication;
-import io.serverlessworkflow.impl.WorkflowPosition;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.DefaultTaskExecutorFactory;
 import io.serverlessworkflow.impl.executors.TaskExecutorBuilder;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
 
 public class JavaTaskExecutorFactory extends DefaultTaskExecutorFactory {
 
   public TaskExecutorBuilder<? extends TaskBase> getTaskExecutor(
-      WorkflowPosition position,
+      WorkflowMutablePosition position,
       Task task,
       Workflow workflow,
       WorkflowApplication application,

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

@@ -19,7 +19,7 @@
 import java.util.Deque;
 import java.util.stream.Collectors;
 
-public class QueueWorkflowPosition implements WorkflowPosition {
+public class QueueWorkflowPosition implements WorkflowMutablePosition {
 
   private Deque<Object> queue;
 
@@ -36,13 +36,13 @@ public QueueWorkflowPosition copy() {
   }
 
   @Override
-  public WorkflowPosition addIndex(int index) {
+  public WorkflowMutablePosition addIndex(int index) {
     queue.add(index);
     return this;
   }
 
   @Override
-  public WorkflowPosition addProperty(String prop) {
+  public WorkflowMutablePosition addProperty(String prop) {
     queue.add(prop);
     return this;
   }
@@ -58,7 +58,7 @@ public String toString() {
   }
 
   @Override
-  public WorkflowPosition back() {
+  public WorkflowMutablePosition back() {
     queue.removeLast();
     return this;
   }

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

@@ -15,7 +15,7 @@
  */
 package io.serverlessworkflow.impl;
 
-public class StringBufferWorkflowPosition implements WorkflowPosition {
+public class StringBufferWorkflowPosition implements WorkflowMutablePosition {
 
   private StringBuilder sb;
 
@@ -32,13 +32,13 @@ public StringBufferWorkflowPosition copy() {
   }
 
   @Override
-  public WorkflowPosition addIndex(int index) {
+  public WorkflowMutablePosition addIndex(int index) {
     sb.append('/').append(index);
     return this;
   }
 
   @Override
-  public WorkflowPosition addProperty(String prop) {
+  public WorkflowMutablePosition addProperty(String prop) {
     sb.append('/').append(prop);
     return this;
   }
@@ -54,7 +54,7 @@ public String toString() {
   }
 
   @Override
-  public WorkflowPosition back() {
+  public WorkflowMutablePosition back() {
     int indexOf = sb.lastIndexOf("/");
     if (indexOf != -1) {
       sb.substring(0, indexOf);

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

@@ -22,7 +22,7 @@
 import java.util.Map;
 import java.util.Optional;
 
-public class TaskContext {
+public class TaskContext implements TaskContextData {
 
   private final WorkflowModel rawInput;
   private final TaskBase task;
@@ -81,14 +81,17 @@ public void input(WorkflowModel input) {
     this.output = input;
   }
 
+  @Override
   public WorkflowModel input() {
     return input;
   }
 
+  @Override
   public WorkflowModel rawInput() {
     return rawInput;
   }
 
+  @Override
   public TaskBase task() {
     return task;
   }
@@ -99,6 +102,7 @@ public TaskContext rawOutput(WorkflowModel output) {
     return this;
   }
 
+  @Override
   public WorkflowModel rawOutput() {
     return rawOutput;
   }
@@ -108,26 +112,32 @@ public TaskContext output(WorkflowModel output) {
     return this;
   }
 
+  @Override
   public WorkflowModel output() {
     return output;
   }
 
+  @Override
   public WorkflowPosition position() {
     return position;
   }
 
+  @Override
   public Map<String, Object> variables() {
     return contextVariables;
   }
 
+  @Override
   public Instant startedAt() {
     return startedAt;
   }
 
+  @Override
   public Optional<TaskContext> parent() {
     return parentContext;
   }
 
+  @Override
   public String taskName() {
     return taskName;
   }
@@ -137,6 +147,7 @@ public TaskContext completedAt(Instant instant) {
     return this;
   }
 
+  @Override
   public Instant completedAt() {
     return completedAt;
   }
@@ -149,4 +160,17 @@ public TaskContext transition(TransitionInfo transition) {
     this.transition = transition;
     return this;
   }
+
+  @Override
+  public String toString() {
+    return "TaskContext [position="
+        + position
+        + ", startedAt="
+        + startedAt
+        + ", taskName="
+        + taskName
+        + ", completedAt="
+        + completedAt
+        + "]";
+  }
 }

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

@@ -0,0 +1,46 @@
+/*
+ * 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 io.serverlessworkflow.api.types.TaskBase;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+
+public interface TaskContextData {
+
+  WorkflowModel input();
+
+  WorkflowModel rawInput();
+
+  TaskBase task();
+
+  WorkflowModel rawOutput();
+
+  WorkflowModel output();
+
+  WorkflowPosition position();
+
+  Map<String, Object> variables();
+
+  Instant startedAt();
+
+  Optional<TaskContext> parent();
+
+  String taskName();
+
+  Instant completedAt();
+}

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

@@ -26,19 +26,21 @@
 import io.serverlessworkflow.impl.executors.TaskExecutorFactory;
 import io.serverlessworkflow.impl.expressions.ExpressionFactory;
 import io.serverlessworkflow.impl.expressions.RuntimeDescriptor;
+import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener;
 import io.serverlessworkflow.impl.resources.DefaultResourceLoaderFactory;
 import io.serverlessworkflow.impl.resources.ResourceLoaderFactory;
 import io.serverlessworkflow.impl.resources.StaticResource;
 import io.serverlessworkflow.impl.schema.SchemaValidator;
 import io.serverlessworkflow.impl.schema.SchemaValidatorFactory;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.Map;
 import java.util.ServiceLoader;
+import java.util.ServiceLoader.Provider;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
 
 public class WorkflowApplication implements AutoCloseable {
 
@@ -127,7 +129,10 @@ public SchemaValidator getValidator(SchemaInline inline) {
 
     private TaskExecutorFactory taskFactory;
     private ExpressionFactory exprFactory;
-    private Collection<WorkflowExecutionListener> listeners;
+    private Collection<WorkflowExecutionListener> listeners =
+        ServiceLoader.load(WorkflowExecutionListener.class).stream()
+            .map(Provider::get)
+            .collect(Collectors.toList());
     private ResourceLoaderFactory resourceLoaderFactory = DefaultResourceLoaderFactory.get();
     private SchemaValidatorFactory schemaValidatorFactory;
     private WorkflowPositionFactory positionFactory = () -> new QueueWorkflowPosition();
@@ -141,9 +146,6 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private Builder() {}
 
     public Builder withListener(WorkflowExecutionListener listener) {
-      if (listeners == null) {
-        listeners = new HashSet<>();
-      }
       listeners.add(listener);
       return this;
     }

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

@@ -15,20 +15,26 @@
  */
 package io.serverlessworkflow.impl;
 
-public class WorkflowContext {
+public class WorkflowContext implements WorkflowContextData {
   private final WorkflowDefinition definition;
-  private final WorkflowInstance instance;
+  private final WorkflowMutableInstance instance;
   private WorkflowModel context;
 
-  WorkflowContext(WorkflowDefinition definition, WorkflowInstance instance) {
+  WorkflowContext(WorkflowDefinition definition, WorkflowMutableInstance instance) {
     this.definition = definition;
     this.instance = instance;
   }
 
-  public WorkflowInstance instance() {
+  @Override
+  public WorkflowInstanceData instanceData() {
     return instance;
   }
 
+  public WorkflowMutableInstance instance() {
+    return instance;
+  }
+
+  @Override
   public WorkflowModel context() {
     return context;
   }
@@ -37,7 +43,19 @@ public void context(WorkflowModel context) {
     this.context = context;
   }
 
+  @Override
   public WorkflowDefinition definition() {
     return definition;
   }
+
+  @Override
+  public String toString() {
+    return "WorkflowContext [definition="
+        + definition.workflow().getDocument().getName()
+        + ", instance="
+        + instance
+        + ", context="
+        + context
+        + "]";
+  }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java renamed to impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowContextData.java

@@ -15,11 +15,11 @@
  */
 package io.serverlessworkflow.impl;
 
-import io.serverlessworkflow.api.types.TaskBase;
+public interface WorkflowContextData {
 
-public interface WorkflowExecutionListener {
+  WorkflowInstanceData instanceData();
 
-  void onTaskStarted(WorkflowPosition currentPos, TaskBase task);
+  WorkflowModel context();
 
-  void onTaskEnded(WorkflowPosition currentPos, TaskBase task);
+  WorkflowDefinitionData definition();
 }