[Fix #490] Implement until condition for any

Signed-off-by: Francisco Javier Tirado Sarti <[email protected]>

Author: fjtirado

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

@@ -18,19 +18,26 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.api.types.AllEventConsumptionStrategy;
+import io.serverlessworkflow.api.types.AnyEventConsumptionStrategy;
+import io.serverlessworkflow.api.types.EventConsumptionStrategy;
 import io.serverlessworkflow.api.types.EventFilter;
 import io.serverlessworkflow.api.types.ListenTask;
 import io.serverlessworkflow.api.types.ListenTaskConfiguration;
 import io.serverlessworkflow.api.types.ListenTaskConfiguration.ListenAndReadAs;
 import io.serverlessworkflow.api.types.ListenTo;
+import io.serverlessworkflow.api.types.OneEventConsumptionStrategy;
 import io.serverlessworkflow.api.types.SubscriptionIterator;
+import io.serverlessworkflow.api.types.Until;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowFilter;
 import io.serverlessworkflow.impl.WorkflowPosition;
+import io.serverlessworkflow.impl.WorkflowUtils;
 import io.serverlessworkflow.impl.events.CloudEventUtils;
+import io.serverlessworkflow.impl.events.EventConsumer;
 import io.serverlessworkflow.impl.events.EventRegistration;
 import io.serverlessworkflow.impl.events.EventRegistrationBuilder;
 import io.serverlessworkflow.impl.json.JsonUtils;
@@ -40,24 +47,43 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public abstract class ListenExecutor extends RegularTaskExecutor<ListenTask> {
 
-  protected final Collection<EventRegistrationBuilder> regBuilders;
+  protected final EventRegistrationBuilderCollection regBuilders;
+  protected final EventRegistrationBuilderCollection untilRegBuilders;
   protected final Optional<WorkflowFilter> until;
   protected final Optional<TaskExecutor<?>> loop;
   protected final Function<CloudEvent, JsonNode> converter;
+  protected final EventConsumer eventConsumer;
+
+  private static record EventRegistrationBuilderCollection(
+      Collection<EventRegistrationBuilder> registrations, boolean isAnd) {}
 
   public static class ListenExecutorBuilder extends RegularTaskExecutorBuilder<ListenTask> {
 
-    private Collection<EventRegistrationBuilder> registrations;
+    private EventRegistrationBuilderCollection registrations;
     private WorkflowFilter until;
+    private EventRegistrationBuilderCollection untilRegistrations;
     private TaskExecutor<?> loop;
     private Function<CloudEvent, JsonNode> converter = this::defaultCEConverter;
-    private boolean isAnd;
+
+    private EventRegistrationBuilderCollection allEvents(AllEventConsumptionStrategy allStrategy) {
+      return new EventRegistrationBuilderCollection(from(allStrategy.getAll()), true);
+    }
+
+    private EventRegistrationBuilderCollection anyEvents(AnyEventConsumptionStrategy anyStrategy) {
+      List<EventFilter> eventFilters = anyStrategy.getAny();
+      return new EventRegistrationBuilderCollection(
+          eventFilters.isEmpty() ? registerToAll() : from(eventFilters), false);
+    }
+
+    private EventRegistrationBuilderCollection oneEvent(OneEventConsumptionStrategy oneStrategy) {
+      return new EventRegistrationBuilderCollection(List.of(from(oneStrategy.getOne())), false);
+    }
 
     protected ListenExecutorBuilder(
         WorkflowPosition position,
@@ -69,15 +95,29 @@ protected ListenExecutorBuilder(
       ListenTaskConfiguration listen = task.getListen();
       ListenTo to = listen.getTo();
       if (to.getAllEventConsumptionStrategy() != null) {
-        isAnd = true;
-        registrations = from(to.getAllEventConsumptionStrategy().getAll());
+        registrations = allEvents(to.getAllEventConsumptionStrategy());
       } else if (to.getAnyEventConsumptionStrategy() != null) {
-        isAnd = false;
-        List<EventFilter> eventFilters = to.getAnyEventConsumptionStrategy().getAny();
-        registrations = eventFilters.isEmpty() ? registerToAll() : from(eventFilters);
+        AnyEventConsumptionStrategy any = to.getAnyEventConsumptionStrategy();
+        registrations = anyEvents(any);
+        Until untilDesc = any.getUntil();
+        if (untilDesc != null) {
+          if (untilDesc.getAnyEventUntilCondition() != null) {
+            until =
+                WorkflowUtils.buildWorkflowFilter(
+                    application.expressionFactory(), untilDesc.getAnyEventUntilCondition());
+          } else if (untilDesc.getAnyEventUntilConsumed() != null) {
+            EventConsumptionStrategy strategy = untilDesc.getAnyEventUntilConsumed();
+            if (strategy.getAllEventConsumptionStrategy() != null) {
+              untilRegistrations = allEvents(strategy.getAllEventConsumptionStrategy());
+            } else if (strategy.getAnyEventConsumptionStrategy() != null) {
+              untilRegistrations = anyEvents(strategy.getAnyEventConsumptionStrategy());
+            } else if (strategy.getOneEventConsumptionStrategy() != null) {
+              untilRegistrations = oneEvent(strategy.getOneEventConsumptionStrategy());
+            }
+          }
+        }
       } else if (to.getOneEventConsumptionStrategy() != null) {
-        isAnd = false;
-        registrations = List.of(from(to.getOneEventConsumptionStrategy().getOne()));
+        registrations = oneEvent(to.getOneEventConsumptionStrategy());
       }
       SubscriptionIterator forEach = task.getForeach();
       if (forEach != null) {
@@ -116,7 +156,7 @@ private EventRegistrationBuilder from(EventFilter filter) {
 
     @Override
     public TaskExecutor<ListenTask> buildInstance() {
-      return isAnd ? new AndListenExecutor(this) : new OrListenExecutor(this);
+      return registrations.isAnd() ? new AndListenExecutor(this) : new OrListenExecutor(this);
     }
   }
 
@@ -176,6 +216,59 @@ protected abstract void internalProcessCe(
       TaskContext taskContext,
       CompletableFuture<JsonNode> future);
 
+  @Override
+  protected CompletableFuture<JsonNode> internalExecute(
+      WorkflowContext workflow, TaskContext taskContext) {
+    ArrayNode output = JsonUtils.mapper().createArrayNode();
+    Collection<EventRegistration> registrations = new ArrayList<>();
+    CompletableFuture<?> combinedFuture =
+        combine(
+            toCompletables(
+                regBuilders,
+                registrations,
+                (ce, future) ->
+                    processCe(converter.apply(ce), output, workflow, taskContext, future)));
+    CompletableFuture<JsonNode> resultFuture =
+        combinedFuture.thenApply(
+            v -> {
+              registrations.forEach(reg -> eventConsumer.unregister(reg));
+              return output;
+            });
+    if (untilRegBuilders != null) {
+      Collection<EventRegistration> untilRegistrations = new ArrayList<>();
+      CompletableFuture<?>[] futures =
+          toCompletables(
+              untilRegBuilders, untilRegistrations, (ce, future) -> future.complete(null));
+      CompletableFuture<?> untilFuture =
+          untilRegBuilders.isAnd()
+              ? CompletableFuture.allOf(futures)
+              : CompletableFuture.anyOf(futures);
+      untilFuture
+          .thenAccept(v -> combinedFuture.complete(null))
+          .thenAccept(v -> untilRegistrations.forEach(reg -> eventConsumer.unregister(reg)));
+    }
+    return resultFuture;
+  }
+
+  private <T> CompletableFuture<T>[] toCompletables(
+      EventRegistrationBuilderCollection regCollection,
+      Collection<EventRegistration> registrations,
+      BiConsumer<CloudEvent, CompletableFuture<T>> consumer) {
+    return regCollection.registrations().stream()
+        .map(reg -> toCompletable(reg, registrations, consumer))
+        .toArray(size -> new CompletableFuture[size]);
+  }
+
+  private <T> CompletableFuture<T> toCompletable(
+      EventRegistrationBuilder regBuilder,
+      Collection<EventRegistration> registrations,
+      BiConsumer<CloudEvent, CompletableFuture<T>> ceConsumer) {
+    final CompletableFuture<T> future = new CompletableFuture<>();
+    registrations.add(
+        eventConsumer.register(regBuilder, ce -> ceConsumer.accept((CloudEvent) ce, future)));
+    return future;
+  }
+
   private void processCe(
       JsonNode node,
       ArrayNode arrayNode,
@@ -199,48 +292,13 @@ private void processCe(
         () -> internalProcessCe(node, arrayNode, workflow, taskContext, future));
   }
 
-  protected CompletableFuture<JsonNode> toCompletable(
-      WorkflowContext workflow,
-      TaskContext taskContext,
-      EventRegistrationBuilder regBuilder,
-      Collection<EventRegistration> registrations,
-      ArrayNode arrayNode) {
-    final CompletableFuture<JsonNode> future = new CompletableFuture<>();
-    registrations.add(
-        workflow
-            .definition()
-            .application()
-            .eventConsumer()
-            .register(
-                regBuilder,
-                (Consumer<CloudEvent>)
-                    (ce ->
-                        processCe(converter.apply(ce), arrayNode, workflow, taskContext, future))));
-    return future;
-  }
-
-  @Override
-  protected CompletableFuture<JsonNode> internalExecute(
-      WorkflowContext workflow, TaskContext taskContext) {
-    ArrayNode output = JsonUtils.mapper().createArrayNode();
-    Collection<EventRegistration> registrations = new ArrayList<>();
-    return combine(
-            regBuilders.stream()
-                .map(reg -> toCompletable(workflow, taskContext, reg, registrations, output))
-                .toArray(size -> new CompletableFuture[size]))
-        .thenApply(
-            v -> {
-              registrations.forEach(
-                  reg -> workflow.definition().application().eventConsumer().unregister(reg));
-              return output;
-            });
-  }
-
   protected ListenExecutor(ListenExecutorBuilder builder) {
     super(builder);
+    this.eventConsumer = builder.application.eventConsumer();
     this.regBuilders = builder.registrations;
     this.until = Optional.ofNullable(builder.until);
     this.loop = Optional.ofNullable(builder.loop);
     this.converter = builder.converter;
+    this.untilRegBuilders = builder.untilRegistrations;
   }
 }

impl/core/src/test/resources/listen-to-any-filter.yaml

@@ -14,10 +14,12 @@ do:
           - with:
               type: com.fake-hospital.vitals.measurements.bpm
               data: ${ .bpm < 60 or .bpm > 100 }
+          until: ( . | length ) > 0
       foreach:
         item: event
         do:
           - isSick: 
               set: 
                 temperature: ${$event.temperature}
-                isSick: true
+                isSick: true
+