[Fix_#549] Implement if for all task

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

Author: fjtirado

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

@@ -49,12 +49,14 @@ public abstract class AbstractTaskExecutor<T extends TaskBase> implements TaskEx
   private final Optional<SchemaValidator> inputSchemaValidator;
   private final Optional<SchemaValidator> outputSchemaValidator;
   private final Optional<SchemaValidator> contextSchemaValidator;
+  private final Optional<WorkflowFilter> ifFilter;
 
   public abstract static class AbstractTaskExecutorBuilder<T extends TaskBase>
       implements TaskExecutorBuilder<T> {
     private Optional<WorkflowFilter> inputProcessor = Optional.empty();
     private Optional<WorkflowFilter> outputProcessor = Optional.empty();
     private Optional<WorkflowFilter> contextProcessor = Optional.empty();
+    private Optional<WorkflowFilter> ifFilter = Optional.empty();
     private Optional<SchemaValidator> inputSchemaValidator = Optional.empty();
     private Optional<SchemaValidator> outputSchemaValidator = Optional.empty();
     private Optional<SchemaValidator> contextSchemaValidator = Optional.empty();
@@ -100,6 +102,7 @@ protected AbstractTaskExecutorBuilder(
         this.contextSchemaValidator =
             getSchemaValidator(application.validatorFactory(), resourceLoader, export.getSchema());
       }
+      this.ifFilter = optionalFilter(application.expressionFactory(), task.getIf());
     }
 
     protected final TransitionInfoBuilder next(
@@ -153,6 +156,7 @@ protected AbstractTaskExecutor(AbstractTaskExecutorBuilder<T> builder) {
     this.inputSchemaValidator = builder.inputSchemaValidator;
     this.outputSchemaValidator = builder.outputSchemaValidator;
     this.contextSchemaValidator = builder.contextSchemaValidator;
+    this.ifFilter = builder.ifFilter;
   }
 
   protected final CompletableFuture<TaskContext> executeNext(
@@ -177,40 +181,49 @@ public CompletableFuture<TaskContext> apply(
     if (!TaskExecutorHelper.isActive(workflowContext)) {
       return completable;
     }
-    return executeNext(
-        completable
-            .thenApply(
-                t -> {
-                  workflowContext
-                      .definition()
-                      .listeners()
-                      .forEach(l -> l.onTaskStarted(position, task));
-                  inputSchemaValidator.ifPresent(s -> s.validate(t.rawInput()));
-                  inputProcessor.ifPresent(
-                      p -> taskContext.input(p.apply(workflowContext, t, t.rawInput())));
-                  return t;
-                })
-            .thenCompose(t -> execute(workflowContext, t))
-            .thenApply(
-                t -> {
-                  outputProcessor.ifPresent(
-                      p -> t.output(p.apply(workflowContext, t, t.rawOutput())));
-                  outputSchemaValidator.ifPresent(s -> s.validate(t.output()));
-                  contextProcessor.ifPresent(
-                      p ->
-                          workflowContext.context(
-                              p.apply(workflowContext, t, workflowContext.context())));
-                  contextSchemaValidator.ifPresent(s -> s.validate(workflowContext.context()));
-                  t.completedAt(Instant.now());
-                  workflowContext
-                      .definition()
-                      .listeners()
-                      .forEach(l -> l.onTaskEnded(position, task));
-                  return t;
-                }),
-        workflowContext);
+    if (ifFilter
+        .map(f -> f.apply(workflowContext, taskContext, input).asBoolean(true))
+        .orElse(true)) {
+      return executeNext(
+          completable
+              .thenApply(
+                  t -> {
+                    workflowContext
+                        .definition()
+                        .listeners()
+                        .forEach(l -> l.onTaskStarted(position, task));
+                    inputSchemaValidator.ifPresent(s -> s.validate(t.rawInput()));
+                    inputProcessor.ifPresent(
+                        p -> taskContext.input(p.apply(workflowContext, t, t.rawInput())));
+                    return t;
+                  })
+              .thenCompose(t -> execute(workflowContext, t))
+              .thenApply(
+                  t -> {
+                    outputProcessor.ifPresent(
+                        p -> t.output(p.apply(workflowContext, t, t.rawOutput())));
+                    outputSchemaValidator.ifPresent(s -> s.validate(t.output()));
+                    contextProcessor.ifPresent(
+                        p ->
+                            workflowContext.context(
+                                p.apply(workflowContext, t, workflowContext.context())));
+                    contextSchemaValidator.ifPresent(s -> s.validate(workflowContext.context()));
+                    t.completedAt(Instant.now());
+                    workflowContext
+                        .definition()
+                        .listeners()
+                        .forEach(l -> l.onTaskEnded(position, task));
+                    return t;
+                  }),
+          workflowContext);
+    } else {
+      taskContext.transition(getSkipTransition());
+      return executeNext(completable, workflowContext);
+    }
   }
 
+  protected abstract TransitionInfo getSkipTransition();
+
   protected abstract CompletableFuture<TaskContext> execute(
       WorkflowContext workflow, TaskContext taskContext);
 }

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

@@ -54,6 +54,11 @@ public void connect(Map<String, TaskExecutorBuilder<?>> connections) {
     }
   }
 
+  @Override
+  protected TransitionInfo getSkipTransition() {
+    return transition;
+  }
+
   protected CompletableFuture<TaskContext> execute(
       WorkflowContext workflow, TaskContext taskContext) {
     CompletableFuture<TaskContext> future =

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

@@ -79,6 +79,11 @@ protected TaskExecutor<SwitchTask> buildInstance() {
     }
   }
 
+  @Override
+  protected TransitionInfo getSkipTransition() {
+    return defaultTask;
+  }
+
   private SwitchExecutor(SwitchExecutorBuilder builder) {
     super(builder);
     this.defaultTask = TransitionInfo.build(builder.defaultTask);

impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java

@@ -81,6 +81,14 @@ private static Stream<Arguments> provideParameters() {
             "simple-expression.yaml",
             Map.of("input", Arrays.asList(1, 2, 3)),
             WorkflowDefinitionTest::checkSpecialKeywords),
+        args(
+            "conditional-set.yaml",
+            Map.of("enabled", true),
+            WorkflowDefinitionTest::checkEnableCondition),
+        args(
+            "conditional-set.yaml",
+            Map.of("enabled", false),
+            WorkflowDefinitionTest::checkDisableCondition),
         args(
             "raise-inline copy.yaml",
             WorkflowDefinitionTest::checkWorkflowException,
@@ -166,4 +174,14 @@ private static void checkSpecialKeywords(Object obj) {
     assertThat(result.get("id").toString()).hasSize(26);
     assertThat(result.get("version").toString()).contains("alpha");
   }
+
+  private static void checkEnableCondition(Object obj) {
+    Map<String, Object> result = (Map<String, Object>) obj;
+    assertThat(result.get("name")).isEqualTo("javierito");
+  }
+
+  private static void checkDisableCondition(Object obj) {
+    Map<String, Object> result = (Map<String, Object>) obj;
+    assertThat(result.get("enabled")).isEqualTo(false);
+  }
 }

impl/core/src/test/resources/conditional-set.yaml

@@ -0,0 +1,10 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: conditional-set
+  version: '0.1.0'
+do: 
+  - conditionalExpression:
+      if: .enabled
+      set:
+        name: javierito