[Fix #1339] Add SerializableFunction support to ForEach on DSL (#1340)

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncForTaskBuilder.java

@@ -64,11 +64,17 @@ public <T, V> FuncForTaskBuilder whileC(LoopPredicateIndex<T, V> predicate) {
     return this;
   }
 
-  public <T> FuncForTaskBuilder collection(Function<T, Collection<?>> collectionF) {
+  public <T, V> FuncForTaskBuilder collection(Function<T, Collection<V>> collectionF) {
     this.forTaskFunction.withCollection(collectionF);
     return this;
   }
 
+  public <T, V> FuncForTaskBuilder collection(
+      Function<T, Collection<V>> collectionF, Class<T> clazz) {
+    this.forTaskFunction.withCollection(collectionF, clazz);
+    return this;
+  }
+
   public <T, V, R> FuncForTaskBuilder tasks(String name, LoopFunction<T, V, R> function) {
     if (name == null || name.isBlank()) {
       name = "for-task-" + this.items.size();

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncDSL.java

@@ -26,6 +26,7 @@
 import io.serverlessworkflow.api.types.OAuth2AuthenticationData;
 import io.serverlessworkflow.api.types.func.ContextFunction;
 import io.serverlessworkflow.api.types.func.FilterFunction;
+import io.serverlessworkflow.api.types.func.LoopFunction;
 import io.serverlessworkflow.fluent.func.FuncCallTaskBuilder;
 import io.serverlessworkflow.fluent.func.FuncEmitTaskBuilder;
 import io.serverlessworkflow.fluent.func.FuncRaiseTaskBuilder;
@@ -1064,9 +1065,25 @@ public static FuncTaskConfigurer switchWhenOrElse(
    * @param <T> input type for the collection function
    * @return list configurer
    */
-  public static <T> FuncTaskConfigurer forEach(
-      Function<T, Collection<?>> collection, Consumer<FuncTaskItemListBuilder> body) {
-    return list -> list.forEach(j -> j.collection(collection).tasks(body));
+  public static <T, V> FuncTaskConfigurer forEach(
+      SerializableFunction<T, Collection<V>> collection, Consumer<FuncTaskItemListBuilder> body) {
+    return list ->
+        list.forEach(
+            j -> j.collection(collection, ReflectionUtils.inferInputType(collection)).tasks(body));
+  }
+
+  public static <T, V> FuncTaskConfigurer forEach(
+      SerializableFunction<T, Collection<V>> collection, LoopFunction<T, V, ?> function) {
+    return list ->
+        list.forEach(
+            j ->
+                j.collection(collection, ReflectionUtils.inferInputType(collection))
+                    .tasks(function));
+  }
+
+  public static <T, V> FuncTaskConfigurer forEachItem(
+      SerializableFunction<T, Collection<V>> collection, Function<V, ?> function) {
+    return forEach(collection, ((t, v) -> function.apply((V) v)));
   }
 
   /**
@@ -1077,9 +1094,9 @@ public static <T> FuncTaskConfigurer forEach(
    * @param <T> ignored (kept for signature consistency)
    * @return list configurer
    */
-  public static <T> FuncTaskConfigurer forEach(
-      Collection<?> collection, Consumer<FuncTaskItemListBuilder> body) {
-    Function<T, Collection<?>> f = ctx -> (Collection<?>) collection;
+  public static <T, V> FuncTaskConfigurer forEach(
+      Collection<V> collection, Consumer<FuncTaskItemListBuilder> body) {
+    Function<T, Collection<V>> f = ctx -> collection;
     return list -> list.forEach(j -> j.collection(f).tasks(body));
   }
 

experimental/test/src/test/java/io/serverlessworkflow/fluent/test/ForEachFuncTest.java

@@ -0,0 +1,57 @@
+/*
+ * 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.fluent.test;
+
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.*;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowModel;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+public class ForEachFuncTest {
+
+  private static record Order(String id) {}
+
+  private static record EnhancedOrder(String id, int salary) {}
+
+  private static record OrdersPayload(List<Order> orders) {}
+
+  @Test
+  void testForEachIteration() throws Exception {
+
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("foreach-workflow")
+            .tasks(forEachItem(OrdersPayload::orders, ForEachFuncTest::enhace))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      OrdersPayload input =
+          new OrdersPayload(
+              List.of(new Order("ORD-001"), new Order("ORD-002"), new Order("ORD-003")));
+      WorkflowModel result = app.workflowDefinition(workflow).instance(input).start().join();
+      assertThat(result.as(EnhancedOrder.class).orElseThrow().id())
+          .isEqualTo(input.orders().get(input.orders.size() - 1).id());
+    }
+  }
+
+  private static EnhancedOrder enhace(Order order) {
+    return new EnhancedOrder(order.id(), 1000);
+  }
+}

experimental/types/src/main/java/io/serverlessworkflow/api/types/func/ForTaskFunction.java

@@ -29,7 +29,7 @@ public class ForTaskFunction extends ForTask {
   private Optional<Class<?>> whileClass = Optional.empty();
   private Optional<Class<?>> itemClass = Optional.empty();
   private Optional<Class<?>> forClass = Optional.empty();
-  private Function<?, Collection<?>> collection;
+  private Function collection;
 
   public <T, V> ForTaskFunction withWhile(LoopPredicate<T, V> whilePredicate) {
     return withWhile(toPredicate(whilePredicate));
@@ -119,12 +119,12 @@ private <T, V> ForTaskFunction withWhile(
     return this;
   }
 
-  public <T> ForTaskFunction withCollection(Function<T, Collection<?>> collection) {
+  public <T, V> ForTaskFunction withCollection(Function<T, Collection<V>> collection) {
     return withCollection(collection, null);
   }
 
-  public <T> ForTaskFunction withCollection(
-      Function<T, Collection<?>> collection, Class<T> colArgClass) {
+  public <T, V> ForTaskFunction withCollection(
+      Function<T, Collection<V>> collection, Class<T> colArgClass) {
     this.collection = collection;
     this.forClass = Optional.ofNullable(colArgClass);
     return this;