[Fix #490] Implement Listen& Emit Task

api/src/main/resources/schema/workflow.yaml

@@ -558,7 +558,17 @@ $defs:
             $ref: '#/$defs/eventConsumptionStrategy'
             title: ListenTo
             description: Defines the event(s) to listen to.
+          read:
+            type: string
+            enum: [ data, envelope, raw ]
+            default: data
+            title: ListenAndReadAs
+            description: Specifies how events are read during the listen operation.
         required: [ to ]
+      foreach:
+        $ref: '#/$defs/subscriptionIterator'
+        title: ListenIterator
+        description: Configures the iterator, if any, for processing consumed event(s).
   raiseTask:
     type: object
     $ref: '#/$defs/taskBase'
@@ -1315,47 +1325,25 @@ $defs:
               $ref: '#/$defs/eventFilter'
         required: [ all ]
       - title: AnyEventConsumptionStrategy
-        oneOf:
-          - properties:
-              any:
-                type: array
-                title: AnyEventConsumptionStrategyConfiguration
-                description: A list containing any of the events to consume.
-                items:
-                  $ref: '#/$defs/eventFilter'
-                minItems: 1
-              until:
-                oneOf:
-                  - type: string
-                    title: AnyEventUntilCondition
-                    description: A runtime expression condition evaluated after consuming an event and which determines whether or not to continue listening.
-                  - allOf:
-                      - $ref: '#/$defs/eventConsumptionStrategy'
-                        title: AnyEventUntilConsumed
-                        description: The strategy that defines the event(s) to consume to stop listening.
-                      - properties:
-                          until: false
-            required: [ any ]
-          - properties:
-              any:
-                type: array
-                title: AnyEventConsumptionStrategyConfiguration
-                description: A list containing any of the events to consume.
-                items:
-                  $ref: '#/$defs/eventFilter'
-                maxItems: 0
-              until:
-                oneOf:
-                  - type: string
-                    title: AnyEventUntilCondition
-                    description: A runtime expression condition evaluated after consuming an event and which determines whether or not to continue listening.
-                  - allOf:
-                      - $ref: '#/$defs/eventConsumptionStrategy'
-                        title: AnyEventUntilConsumed
-                        description: The strategy that defines the event(s) to consume to stop listening.
-                      - properties:
-                          until: false
-            required: [ any, until ]
+        properties:
+          any:
+              type: array
+              title: AnyEventConsumptionStrategyConfiguration
+              description: A list containing any of the events to consume.
+              items:
+                $ref: '#/$defs/eventFilter'
+          until:
+            oneOf:
+              - type: string
+                title: AnyEventUntilCondition
+                description: A runtime expression condition evaluated after consuming an event and which determines whether or not to continue listening.
+              - allOf:
+                  - $ref: '#/$defs/eventConsumptionStrategy'
+                    description: The strategy that defines the event(s) to consume to stop listening.
+                  - properties:
+                      until: false
+                title: AnyEventUntilConsumed
+        required: [ any ]  
       - title: OneEventConsumptionStrategy
         properties:
           one:
@@ -1710,6 +1698,10 @@ $defs:
         $ref: '#/$defs/asyncApiMessageConsumptionPolicy'
         title: AsyncApiMessageConsumptionPolicy
         description: An object used to configure the subscription's message consumption policy.
+      foreach:
+        $ref: '#/$defs/subscriptionIterator'
+        title: AsyncApiSubscriptionIterator
+        description: Configures the iterator, if any, for processing consumed messages(s).
     required: [ consume ]
   asyncApiMessageConsumptionPolicy:
     type: object
@@ -1740,3 +1732,31 @@ $defs:
             title: AsyncApiMessageConsumptionPolicyUntil
             description: A runtime expression evaluated before each consumed (filtered) message to decide if message consumption should continue.
         required: [ until ]
+  subscriptionIterator:
+    type: object
+    title: SubscriptionIterator
+    description: Configures the iteration over each item (event or message) consumed by a subscription.
+    unevaluatedProperties: false
+    properties:
+      item:
+        type: string
+        title: SubscriptionIteratorItem
+        description: The name of the variable used to store the current item being enumerated.
+        default: item
+      at:
+        type: string
+        title: SubscriptionIteratorIndex
+        description: The name of the variable used to store the index of the current item being enumerated.
+        default: index
+      do:
+        $ref: '#/$defs/taskList'
+        title: SubscriptionIteratorTasks
+        description: The tasks to perform for each consumed item.
+      output:
+        $ref: '#/$defs/output'
+        title: SubscriptionIteratorOutput
+        description: An object, if any, used to customize the item's output and to document its schema.
+      export:
+        $ref: '#/$defs/export'
+        title: SubscriptionIteratorExport
+        description: An object, if any, used to customize the content of the workflow context.

custom-generator/src/main/java/io/serverlessworkflow/generator/AllAnyOneOfSchemaRule.java

@@ -60,6 +60,7 @@ class AllAnyOneOfSchemaRule extends SchemaRule {
   }
 
   private static final String REF = "$ref";
+  private static final String TITLE = "title";
   private static final String PATTERN = "pattern";
 
   private enum Format {
@@ -154,6 +155,16 @@ public JType apply(
         && allOfTypes.isEmpty()
         && refType.isPresent()) {
       javaType = refType.get();
+    } else if (!schemaNode.has("properties")
+        && oneOfTypes.isEmpty()
+        && allOfTypes.size() == 1
+        && refType.isEmpty()) {
+      javaType = allOfTypes.get(0).getType();
+    } else if (!schemaNode.has("properties")
+        && oneOfTypes.size() == 1
+        && allOfTypes.isEmpty()
+        && refType.isEmpty()) {
+      javaType = oneOfTypes.get(0).getType();
     } else {
       JPackage container = generatableType.getPackage();
       javaType = ruleFactory.getTypeRule().apply(nodeName, schemaNode, parent, container, schema);
@@ -469,6 +480,9 @@ private void unionType(
       Collection<JTypeWrapper> types) {
     if (schemaNode.has(prefix)) {
       ArrayNode array = (ArrayNode) schemaNode.get(prefix);
+      if (schemaNode.has(TITLE)) {
+        nodeName = schemaNode.get(TITLE).asText();
+      }
       int i = 0;
       for (JsonNode oneOf : array) {
         if (!ignoreNode(oneOf)) {
@@ -491,6 +505,23 @@ private void unionType(
   }
 
   private static boolean ignoreNode(JsonNode node) {
+    return allRequired(node) || allRemoveProperties(node);
+  }
+
+  private static boolean allRemoveProperties(JsonNode node) {
+    if (node.size() == 1 && node.has("properties")) {
+      JsonNode propsNode = node.get("properties");
+      for (JsonNode propNode : propsNode) {
+        if (!propNode.isBoolean() || propNode.asBoolean()) {
+          return false;
+        }
+      }
+      return true;
+    }
+    return false;
+  }
+
+  private static boolean allRequired(JsonNode node) {
     return node.size() == 1 && node.has("required");
   }
 
@@ -514,7 +545,7 @@ private Optional<JType> refType(
           schema.isGenerated()
               ? schema.getJavaType()
               : apply(
-                  nameFromRef(ref, nodeName),
+                  nameFromRef(ref, nodeName, schemaNode),
                   schema.getContent(),
                   parent,
                   generatableType,
@@ -556,7 +587,10 @@ private String pattern(JsonNode node) {
     return format != null ? format.pattern() : getFromNode(node, PATTERN);
   }
 
-  private String nameFromRef(String ref, String nodeName) {
+  private String nameFromRef(String ref, String nodeName, JsonNode schemaNode) {
+    if (schemaNode.has(TITLE)) {
+      return schemaNode.get(TITLE).asText();
+    }
     if ("#".equals(ref)) {
       return nodeName;
     }

custom-generator/src/main/java/io/serverlessworkflow/generator/RefNameHelper.java

@@ -0,0 +1,42 @@
+/*
+ * 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.generator;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.sun.codemodel.JClassAlreadyExistsException;
+import com.sun.codemodel.JDefinedClass;
+import com.sun.codemodel.JPackage;
+import org.jsonschema2pojo.GenerationConfig;
+import org.jsonschema2pojo.util.NameHelper;
+
+public class RefNameHelper extends NameHelper {
+
+  public RefNameHelper(GenerationConfig generationConfig) {
+    super(generationConfig);
+  }
+
+  @Override
+  public String getUniqueClassName(String nodeName, JsonNode node, JPackage _package) {
+    String className = getClassName(nodeName, node, _package);
+    try {
+      JDefinedClass _class = _package._class(className);
+      _package.remove(_class);
+      return className;
+    } catch (JClassAlreadyExistsException ex) {
+      return super.getUniqueClassName(nodeName, null, _package);
+    }
+  }
+}

custom-generator/src/main/java/io/serverlessworkflow/generator/UnreferencedFactory.java

@@ -18,10 +18,25 @@
 import com.sun.codemodel.JClassContainer;
 import com.sun.codemodel.JDefinedClass;
 import com.sun.codemodel.JType;
+import org.jsonschema2pojo.GenerationConfig;
 import org.jsonschema2pojo.rules.Rule;
 import org.jsonschema2pojo.rules.RuleFactory;
+import org.jsonschema2pojo.util.NameHelper;
 
 public class UnreferencedFactory extends RuleFactory {
+
+  private NameHelper refNameHelper;
+
+  public UnreferencedFactory() {
+    this.refNameHelper = new RefNameHelper(getGenerationConfig());
+  }
+
+  @Override
+  public void setGenerationConfig(final GenerationConfig generationConfig) {
+    super.setGenerationConfig(generationConfig);
+    this.refNameHelper = new RefNameHelper(generationConfig);
+  }
+
   @Override
   public Rule<JClassContainer, JType> getSchemaRule() {
     return new AllAnyOneOfSchemaRule(this);
@@ -36,4 +51,9 @@ public Rule<JClassContainer, JType> getTypeRule() {
   public Rule<JDefinedClass, JDefinedClass> getAdditionalPropertiesRule() {
     return new UnevaluatedPropertiesRule(this);
   }
+
+  @Override
+  public NameHelper getNameHelper() {
+    return refNameHelper;
+  }
 }

impl/core/pom.xml

@@ -6,20 +6,23 @@
     <version>7.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>serverlessworkflow-impl-core</artifactId>
-  <properties>
-      <version.net.thisptr>1.2.0</version.net.thisptr>
-      <version.com.github.f4b6a3>5.2.3</version.com.github.f4b6a3>
-  </properties>
   <dependencies>
     <dependency>
       <groupId>io.serverlessworkflow</groupId>
       <artifactId>serverlessworkflow-api</artifactId>
-      <version>7.0.0-SNAPSHOT</version>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-json-jackson</artifactId>
     </dependency>
     <dependency>
       <groupId>com.github.f4b6a3</groupId>
       <artifactId>ulid-creator</artifactId>
-      <version>${version.com.github.f4b6a3}</version>
     </dependency>
     <dependency>
      <groupId>com.networknt</groupId>
@@ -28,7 +31,6 @@
     <dependency>
       <groupId>net.thisptr</groupId>
       <artifactId>jackson-jq</artifactId>
-      <version>${version.net.thisptr}</version>
     </dependency>
     <dependency>
       <groupId>org.junit.jupiter</groupId>

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

@@ -0,0 +1,20 @@
+/*
+ * 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;
+
+public interface ExpressionHolder<T> extends BiFunction<WorkflowContext, TaskContext, T> {}

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

@@ -15,7 +15,5 @@
  */
 package io.serverlessworkflow.impl;
 
-import java.util.function.BiFunction;
-
 @FunctionalInterface
-public interface LongFilter extends BiFunction<WorkflowContext, TaskContext, Long> {}
+public interface LongFilter extends ExpressionHolder<Long> {}

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

@@ -15,7 +15,5 @@
  */
 package io.serverlessworkflow.impl;
 
-import java.util.function.BiFunction;
-
 @FunctionalInterface
-public interface StringFilter extends BiFunction<WorkflowContext, TaskContext, String> {}
+public interface StringFilter extends ExpressionHolder<String> {}