[Fix serverlessworkflow#490] Delaying registration

Author: fjtirado

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'
@@ -1710,6 +1720,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 +1754,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.

api/src/test/resources/features/emit.yaml

@@ -10,5 +10,5 @@ do:
           with:
             source: https://fake-source.com
             type: com.fake-source.user.greeted.v1
-            data:
-              greetings: ${ "Hello \(.user.firstName) \(.user.lastName)!" }
+            data: [1,2,3,4]
+              

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

@@ -0,0 +1,33 @@
+/*
+ * 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.EventFilter;
+
+public abstract class AbstractTypeConsumer implements EventConsumer<TypeEventRegistration> {
+
+  @Override
+  public EventRegistrationBuilder<TypeEventRegistration> register(EventFilter register) {
+    // TODO Auto-generated method stub
+    return null;
+  }
+
+  @Override
+  public void unregister(TypeEventRegistration register) {
+    // TODO Auto-generated method stub
+
+  }
+}

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

@@ -15,7 +15,7 @@
  */
 package io.serverlessworkflow.impl;
 
-public class DefaultEventConsumer implements EventConsumer {
+public class DefaultEventConsumer extends AbstractTypeConsumer {
 
   private static DefaultEventConsumer instance = new DefaultEventConsumer();
 
@@ -24,10 +24,4 @@ private DefaultEventConsumer() {}
   public static DefaultEventConsumer get() {
     return instance;
   }
-
-  @Override
-  public void register(EventRegistration register) {}
-
-  @Override
-  public void unregister(EventRegistration register) {}
 }

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

@@ -15,8 +15,10 @@
  */
 package io.serverlessworkflow.impl;
 
-public interface EventConsumer {
-  void register(EventRegistration register);
+import io.serverlessworkflow.api.types.EventFilter;
 
-  void unregister(EventRegistration register);
+public interface EventConsumer<T extends EventRegistration> {
+  EventRegistrationBuilder<T> register(EventFilter register);
+
+  void unregister(T register);
 }

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

@@ -15,42 +15,4 @@
  */
 package io.serverlessworkflow.impl;
 
-import io.cloudevents.CloudEvent;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.function.Consumer;
-import java.util.function.Predicate;
-
-public record EventRegistration(
-    Consumer<CloudEvent> consumer, String type, Optional<Predicate<CloudEvent>> filter) {
-
-  public static EventRegistrationBuilder builder(String type) {
-    return new EventRegistrationBuilder(type);
-  }
-
-  public static class EventRegistrationBuilder {
-
-    private final String type;
-    private Consumer<CloudEvent> consumer;
-    private Predicate<CloudEvent> predicate;
-
-    private EventRegistrationBuilder(String type) {
-      this.type = type;
-    }
-
-    public EventRegistrationBuilder withConsumer(Consumer<CloudEvent> consumer) {
-      this.consumer = consumer;
-      return this;
-    }
-
-    public EventRegistrationBuilder withFilter(Predicate<CloudEvent> predicate) {
-      this.predicate = predicate;
-      return this;
-    }
-
-    public EventRegistration build() {
-      return new EventRegistration(
-          Objects.requireNonNull(consumer), type, Optional.ofNullable(predicate));
-    }
-  }
-}
+public interface EventRegistration {}

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

@@ -0,0 +1,23 @@
+/*
+ * 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.cloudevents.CloudEvent;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+public interface EventRegistrationBuilder<T extends EventRegistration>
+    extends Function<Consumer<CloudEvent>, T> {}

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

@@ -0,0 +1,22 @@
+/*
+ * 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.cloudevents.CloudEvent;
+import java.util.function.Consumer;
+
+public record TypeEventRegistration(String type, Consumer<CloudEvent> consumer)
+    implements EventRegistration {}

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

@@ -47,7 +47,7 @@ public class WorkflowApplication implements AutoCloseable {
   private final WorkflowPositionFactory positionFactory;
   private final ExecutorServiceFactory executorFactory;
   private final RuntimeDescriptorFactory runtimeDescriptorFactory;
-  private final EventConsumer eventConsumer;
+  private final EventConsumer<?> eventConsumer;
 
   private ExecutorService executorService;
 
@@ -102,7 +102,7 @@ public static class Builder {
     private WorkflowPositionFactory positionFactory = () -> new QueueWorkflowPosition();
     private WorkflowIdFactory idFactory = () -> UlidCreator.getMonotonicUlid().toString();
     private ExecutorServiceFactory executorFactory = () -> Executors.newCachedThreadPool();
-    private EventConsumer eventConsumer = DefaultEventConsumer.get();
+    private EventConsumer<?> eventConsumer = DefaultEventConsumer.get();
     private RuntimeDescriptorFactory descriptorFactory =
         () -> new RuntimeDescriptor("reference impl", "1.0.0_alpha", Collections.emptyMap());
 
@@ -156,6 +156,11 @@ public Builder withDescriptorFactory(RuntimeDescriptorFactory factory) {
       return this;
     }
 
+    public Builder withEventConsumer(EventConsumer<?> eventConsumer) {
+      this.eventConsumer = eventConsumer;
+      return this;
+    }
+
     public WorkflowApplication build() {
       return new WorkflowApplication(this);
     }
@@ -188,6 +193,7 @@ public RuntimeDescriptorFactory runtimeDescriptorFactory() {
     return runtimeDescriptorFactory;
   }
 
+  @SuppressWarnings("rawtypes")
   public EventConsumer eventConsumer() {
     return eventConsumer;
   }

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

@@ -101,10 +101,6 @@ public Optional<WorkflowFilter> outputFilter() {
     return outputFilter;
   }
 
-  public WorkflowIdFactory idFactory() {
-    return application.idFactory();
-  }
-
   public Optional<SchemaValidator> outputSchemaValidator() {
     return outputSchemaValidator;
   }
@@ -113,6 +109,10 @@ public RuntimeDescriptorFactory runtimeDescriptorFactory() {
     return application.runtimeDescriptorFactory();
   }
 
+  public WorkflowApplication application() {
+    return application;
+  }
+
   @Override
   public void close() {
     // TODO close resourcers hold for uncompleted process instances, if any

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

@@ -36,7 +36,7 @@ public class WorkflowInstance {
   private CompletableFuture<JsonNode> completableFuture;
 
   WorkflowInstance(WorkflowDefinition definition, JsonNode input) {
-    this.id = definition.idFactory().get();
+    this.id = definition.application().idFactory().get();
     this.input = input;
     this.definition = definition;
     this.status = new AtomicReference<>(WorkflowStatus.PENDING);