[Fix #490] Listen & Wait Events unit test

fjtirado · fjtirado · commit 41d9e941236e · 2025-01-24T14:36:39.000+01:00
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/AbstractTypeConsumer.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/AbstractTypeConsumer.java
@@ -24,34 +24,50 @@
 import java.util.AbstractCollection;
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class AbstractTypeConsumer
     implements EventConsumer<TypeEventRegistration, TypeEventRegistrationBuilder> {
 
+  private static final Logger logger = LoggerFactory.getLogger(AbstractTypeConsumer.class);
+
+  protected abstract void registerToAll(Consumer<CloudEvent> consumer);
+
+  protected abstract void unregisterFromAll();
+
   protected abstract void register(String topicName, Consumer<CloudEvent> consumer);
 
   protected abstract void unregister(String topicName);
 
   private Map<String, CloudEventConsumer> registrations = new ConcurrentHashMap<>();
 
   @Override
-  public TypeEventRegistrationBuilder build(EventFilter register, WorkflowApplication application) {
+  public TypeEventRegistrationBuilder listen(
+      EventFilter register, WorkflowApplication application) {
     EventProperties properties = register.getWith();
     String type = properties.getType();
     return new TypeEventRegistrationBuilder(
         type, new DefaultCloudEventPredicate(properties, application.expressionFactory()));
   }
 
+  @Override
+  public Collection<TypeEventRegistrationBuilder> listenToAll(WorkflowApplication application) {
+    return List.of(new TypeEventRegistrationBuilder(null, null));
+  }
+
   private static class CloudEventConsumer extends AbstractCollection<TypeEventRegistration>
       implements Consumer<CloudEvent> {
     private Collection<TypeEventRegistration> registrations = new CopyOnWriteArrayList<>();
 
     @Override
     public void accept(CloudEvent ce) {
+      logger.debug("Received cloud event {}", ce);
       for (TypeEventRegistration registration : registrations) {
         if (registration.predicate().test(ce, registration.workflow(), registration.task())) {
           registration.consumer().accept(ce);
@@ -80,32 +96,42 @@ public TypeEventRegistration register(
       Consumer<CloudEvent> ce,
       WorkflowContext workflow,
       TaskContext taskContext) {
-    TypeEventRegistration registration =
-        new TypeEventRegistration(builder.type(), ce, builder.cePredicate(), workflow, taskContext);
-    registrations
-        .computeIfAbsent(
-            registration.type(),
-            k -> {
-              CloudEventConsumer consumer = new CloudEventConsumer();
-              register(k, consumer);
-              return consumer;
-            })
-        .add(registration);
-    return registration;
+    if (builder.type() == null) {
+      registerToAll(ce);
+      return new TypeEventRegistration(null, ce, null, workflow, taskContext);
+    } else {
+      TypeEventRegistration registration =
+          new TypeEventRegistration(
+              builder.type(), ce, builder.cePredicate(), workflow, taskContext);
+      registrations
+          .computeIfAbsent(
+              registration.type(),
+              k -> {
+                CloudEventConsumer consumer = new CloudEventConsumer();
+                register(k, consumer);
+                return consumer;
+              })
+          .add(registration);
+      return registration;
+    }
   }
 
   @Override
   public void unregister(TypeEventRegistration registration) {
-    registrations.computeIfPresent(
-        registration.type(),
-        (k, v) -> {
-          v.remove(registration);
-          if (v.isEmpty()) {
-            unregister(registration.type());
-            return null;
-          } else {
-            return v;
-          }
-        });
+    if (registration.type() == null) {
+      unregisterFromAll();
+    } else {
+      registrations.computeIfPresent(
+          registration.type(),
+          (k, v) -> {
+            v.remove(registration);
+            if (v.isEmpty()) {
+              unregister(registration.type());
+              return null;
+            } else {
+              return v;
+            }
+          });
+    }
   }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventConsumer.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/EventConsumer.java
@@ -20,11 +20,14 @@
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
+import java.util.Collection;
 import java.util.function.Consumer;
 
 public interface EventConsumer<T extends EventRegistration, V extends EventRegistrationBuilder> {
 
-  V build(EventFilter filter, WorkflowApplication workflowApplication);
+  V listen(EventFilter filter, WorkflowApplication workflowApplication);
+
+  Collection<V> listenToAll(WorkflowApplication workflowApplication);
 
   T register(V builder, Consumer<CloudEvent> consumer, WorkflowContext context, TaskContext task);
 
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java
@@ -19,10 +19,11 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 /*
- * Straighforward implementation of in memory event broker.
+ * Straightforward implementation of in memory event broker.
  * User might invoke notifyCE to simulate event reception.
  */
 public class InMemoryEvents extends AbstractTypeConsumer implements EventPublisher {
@@ -37,6 +38,8 @@ public static InMemoryEvents get() {
 
   private Map<String, Consumer<CloudEvent>> topicMap = new ConcurrentHashMap<>();
 
+  private AtomicReference<Consumer<CloudEvent>> allConsumerRef = new AtomicReference<>();
+
   @Override
   protected void register(String topicName, Consumer<CloudEvent> consumer) {
     topicMap.put(topicName, consumer);
@@ -51,10 +54,24 @@ protected void unregister(String topicName) {
   public CompletableFuture<Void> publish(CloudEvent ce) {
     return CompletableFuture.runAsync(
         () -> {
+          Consumer<CloudEvent> allConsumer = allConsumerRef.get();
+          if (allConsumer != null) {
+            allConsumer.accept(ce);
+          }
           Consumer<CloudEvent> consumer = topicMap.get(ce.getType());
           if (consumer != null) {
             consumer.accept(ce);
           }
         });
   }
+
+  @Override
+  protected void registerToAll(Consumer<CloudEvent> consumer) {
+    allConsumerRef.set(consumer);
+  }
+
+  @Override
+  protected void unregisterFromAll() {
+    allConsumerRef.set(null);
+  }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java
@@ -29,6 +29,7 @@
 import io.serverlessworkflow.impl.events.EventPropertiesFilter;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
 import java.net.URI;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 
 public class EmitExecutor extends RegularTaskExecutor<EmitTask> {
@@ -75,22 +76,27 @@ protected CompletableFuture<JsonNode> internalExecute(
 
   private CloudEvent buildCloudEvent(WorkflowContext workflow, TaskContext taskContext) {
     io.cloudevents.core.v1.CloudEventBuilder ceBuilder = CloudEventBuilder.v1();
-    props
-        .idFilter()
-        .map(filter -> filter.apply(workflow, taskContext))
-        .ifPresent(value -> ceBuilder.withId(value));
+    ceBuilder.withId(
+        props
+            .idFilter()
+            .map(filter -> filter.apply(workflow, taskContext))
+            .orElse(UUID.randomUUID().toString()));
+    ceBuilder.withSource(
+        props
+            .sourceFilter()
+            .map(filter -> filter.apply(workflow, taskContext))
+            .map(URI::create)
+            .orElse(URI.create("reference-impl")));
+    ceBuilder.withType(
+        props
+            .typeFilter()
+            .map(filter -> filter.apply(workflow, taskContext))
+            .orElseThrow(
+                () -> new IllegalArgumentException("Type is required for emitting events")));
     props
         .timeFilter()
         .map(filter -> filter.apply(workflow, taskContext))
         .ifPresent(value -> ceBuilder.withTime(value));
-    props
-        .sourceFilter()
-        .map(filter -> filter.apply(workflow, taskContext))
-        .ifPresent(value -> ceBuilder.withSource(URI.create(value)));
-    props
-        .typeFilter()
-        .map(filter -> filter.apply(workflow, taskContext))
-        .ifPresent(value -> ceBuilder.withType(value));
     props
         .subjectFilter()
         .map(filter -> filter.apply(workflow, taskContext))
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java
@@ -73,7 +73,8 @@ protected ListenExecutorBuilder(
         registrations = from(to.getAllEventConsumptionStrategy().getAll());
       } else if (to.getAnyEventConsumptionStrategy() != null) {
         isAnd = false;
-        registrations = from(to.getAnyEventConsumptionStrategy().getAny());
+        List<EventFilter> eventFilters = to.getAnyEventConsumptionStrategy().getAny();
+        registrations = eventFilters.isEmpty() ? registerToAll() : from(eventFilters);
       } else if (to.getOneEventConsumptionStrategy() != null) {
         isAnd = false;
         registrations = List.of(from(to.getOneEventConsumptionStrategy().getOne()));
@@ -97,6 +98,10 @@ protected ListenExecutorBuilder(
       }
     }
 
+    private Collection<EventRegistrationBuilder> registerToAll() {
+      return application.eventConsumer().listenToAll(application);
+    }
+
     private JsonNode defaultCEConverter(CloudEvent ce) {
       return CloudEventUtils.toJsonNode(ce.getData());
     }
@@ -106,7 +111,7 @@ private Collection<EventRegistrationBuilder> from(List<EventFilter> filters) {
     }
 
     private EventRegistrationBuilder from(EventFilter filter) {
-      return application.eventConsumer().build(filter, application);
+      return application.eventConsumer().listen(filter, application);
     }
 
     @Override
@@ -175,7 +180,7 @@ private void processCe(
       WorkflowContext workflow,
       TaskContext taskContext,
       CompletableFuture<JsonNode> future) {
-    arrayNode.add(arrayNode);
+    arrayNode.add(node);
     loop.ifPresentOrElse(
         t -> {
           SubscriptionIterator forEach = task.getForeach();
diff --git a/impl/core/src/test/java/io/serverlessworkflow/impl/EventDefinitionTest.java b/impl/core/src/test/java/io/serverlessworkflow/impl/EventDefinitionTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.serverlessworkflow.api.WorkflowReader;
+import io.serverlessworkflow.impl.json.JsonUtils;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+public class EventDefinitionTest {
+
+  private static WorkflowApplication appl;
+
+  @BeforeAll
+  static void init() {
+    appl = WorkflowApplication.builder().build();
+  }
+
+  @Test
+  void testAnyEvent() throws IOException {
+    WorkflowDefinition listenDefinition =
+        appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("listen.yaml"));
+    WorkflowDefinition emitDefinition =
+        appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("emit.yaml"));
+    WorkflowInstance waitingInstance = listenDefinition.instance(Map.of());
+    CompletableFuture<JsonNode> future = waitingInstance.start();
+    assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.RUNNING);
+    emitDefinition.instance(Map.of()).start().join();
+    assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.COMPLETED);
+    assertThat(future).isCompleted();
+    assertThat(waitingInstance.outputAsJsonNode()).isEqualTo(cruellaDeVil());
+  }
+
+  private static JsonNode cruellaDeVil() {
+    ObjectMapper mapper = JsonUtils.mapper();
+    ObjectNode node = mapper.createObjectNode();
+    node.set(
+        "client", mapper.createObjectNode().put("firstName", "Cruella").put("lastName", "de Vil"));
+    node.set(
+        "items",
+        mapper
+            .createArrayNode()
+            .add(mapper.createObjectNode().put("breed", "dalmatian").put("quantity", 101)));
+    return mapper.createArrayNode().add(node);
+  }
+}
diff --git a/impl/core/src/test/resources/emit.yaml b/impl/core/src/test/resources/emit.yaml
@@ -0,0 +1,19 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: emit
+  version: '0.1.0'
+do:
+  - emitEvent:
+      emit:
+        event:
+          with:
+            source: https://petstore.com
+            type: com.petstore.order.placed.v1
+            data:
+              client:
+                firstName: Cruella
+                lastName: de Vil
+              items:
+                - breed: dalmatian
+                  quantity: 101
diff --git a/impl/core/src/test/resources/listen.yaml b/impl/core/src/test/resources/listen.yaml
@@ -0,0 +1,10 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: listen-to-any
+  version: '0.1.0'
+do:
+  - callDoctor:
+      listen:
+        to:
+          any: []
