[Fix #1283] Async Java Function Call executor (#1285)

[Fix #1283] Async Java Function Call executor


Apply suggestions from code review

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: fjtirado

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

@@ -46,7 +46,12 @@ public <T, V> FuncCallTaskBuilder function(Function<T, V> function) {
   }
 
   public <T, V> FuncCallTaskBuilder function(Function<T, V> function, Class<T> argClass) {
-    this.callTaskJava = new CallTaskJava(CallJava.function(function, argClass));
+    return function(function, argClass, null);
+  }
+
+  public <T, V> FuncCallTaskBuilder function(
+      Function<T, V> function, Class<T> argClass, Class<V> returnClass) {
+    this.callTaskJava = new CallTaskJava(CallJava.function(function, argClass, returnClass));
     super.setTask(this.callTaskJava.getCallJava());
     return this;
   }
@@ -56,7 +61,12 @@ public <T, V> FuncCallTaskBuilder function(ContextFunction<T, V> function) {
   }
 
   public <T, V> FuncCallTaskBuilder function(ContextFunction<T, V> function, Class<T> argClass) {
-    this.callTaskJava = new CallTaskJava(CallJava.function(function, argClass));
+    return function(function, argClass, null);
+  }
+
+  public <T, V> FuncCallTaskBuilder function(
+      ContextFunction<T, V> function, Class<T> argClass, Class<V> returnClass) {
+    this.callTaskJava = new CallTaskJava(CallJava.function(function, argClass, returnClass));
     super.setTask(this.callTaskJava.getCallJava());
     return this;
   }
@@ -66,7 +76,12 @@ public <T, V> FuncCallTaskBuilder function(FilterFunction<T, V> function) {
   }
 
   public <T, V> FuncCallTaskBuilder function(FilterFunction<T, V> function, Class<T> argClass) {
-    this.callTaskJava = new CallTaskJava(CallJava.function(function, argClass));
+    return function(function, argClass, null);
+  }
+
+  public <T, V> FuncCallTaskBuilder function(
+      FilterFunction<T, V> function, Class<T> argClass, Class<V> outputClass) {
+    this.callTaskJava = new CallTaskJava(CallJava.function(function, argClass, outputClass));
     super.setTask(this.callTaskJava.getCallJava());
     return this;
   }

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

@@ -16,10 +16,10 @@
 package io.serverlessworkflow.fluent.func;
 
 import io.cloudevents.CloudEventData;
+import io.serverlessworkflow.api.reflection.func.SerializableFunction;
 import io.serverlessworkflow.api.types.func.ContextFunction;
 import io.serverlessworkflow.api.types.func.EventDataFunction;
 import io.serverlessworkflow.api.types.func.FilterFunction;
-import io.serverlessworkflow.fluent.func.dsl.SerializableFunction;
 import io.serverlessworkflow.fluent.spec.AbstractEventPropertiesBuilder;
 import java.util.function.Function;
 

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

@@ -56,13 +56,20 @@ public <T, V> FuncForkTaskBuilder branch(String name, Function<T, V> function) {
 
   public <T, V> FuncForkTaskBuilder branch(
       String name, Function<T, V> function, Class<T> argParam) {
+    return branch(name, function, argParam, null);
+  }
+
+  public <T, V> FuncForkTaskBuilder branch(
+      String name, Function<T, V> function, Class<T> argParam, Class<V> returnClass) {
     if (name == null || name.isBlank()) {
       name = "branch-" + this.items.size();
     }
     this.items.add(
         new TaskItem(
             name,
-            new Task().withCallTask(new CallTaskJava(CallJava.function(function, argParam)))));
+            new Task()
+                .withCallTask(
+                    new CallTaskJava(CallJava.function(function, argParam, returnClass)))));
     return this;
   }
 

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

@@ -15,6 +15,9 @@
  */
 package io.serverlessworkflow.fluent.func.dsl;
 
+import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
+import io.serverlessworkflow.api.reflection.func.SerializableFunction;
+import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.fluent.func.FuncCallTaskBuilder;
 import io.serverlessworkflow.fluent.func.FuncSwitchTaskBuilder;

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

@@ -29,59 +29,64 @@ public final class FuncCallStep<T, R> extends Step<FuncCallStep<T, R>, FuncCallT
   private final ContextFunction<T, R> ctxFn;
   private final FilterFunction<T, R> filterFn;
   private final Class<T> argClass;
+  private final Class<R> returnClass;
 
   /** Function<T,R> variant (unnamed). */
-  FuncCallStep(Function<T, R> fn, Class<T> argClass) {
-    this(null, fn, argClass);
+  FuncCallStep(Function<T, R> fn, Class<T> argClass, Class<R> returnClass) {
+    this(null, fn, argClass, returnClass);
   }
 
   /** Function<T,R> variant (named). */
-  FuncCallStep(String name, Function<T, R> fn, Class<T> argClass) {
+  FuncCallStep(String name, Function<T, R> fn, Class<T> argClass, Class<R> returnClass) {
     this.name = name;
     this.fn = fn;
     this.ctxFn = null;
     this.filterFn = null;
     this.argClass = argClass;
+    this.returnClass = returnClass;
   }
 
   /** ContextFunction<T,R> variant (unnamed). */
-  FuncCallStep(ContextFunction<T, R> ctxFn, Class<T> argClass) {
-    this(null, ctxFn, argClass);
+  FuncCallStep(ContextFunction<T, R> ctxFn, Class<T> argClass, Class<R> returnClass) {
+    this(null, ctxFn, argClass, returnClass);
   }
 
   /** ContextFunction<T,R> variant (named). */
-  FuncCallStep(String name, ContextFunction<T, R> ctxFn, Class<T> argClass) {
+  FuncCallStep(String name, ContextFunction<T, R> ctxFn, Class<T> argClass, Class<R> returnClass) {
     this.name = name;
     this.fn = null;
     this.ctxFn = ctxFn;
     this.filterFn = null;
     this.argClass = argClass;
+    this.returnClass = returnClass;
   }
 
   /** FilterFunction<T,R> variant (unnamed). */
-  FuncCallStep(FilterFunction<T, R> filterFn, Class<T> argClass) {
-    this(null, filterFn, argClass);
+  FuncCallStep(FilterFunction<T, R> filterFn, Class<T> argClass, Class<R> returnClass) {
+    this(null, filterFn, argClass, returnClass);
   }
 
   /** FilterFunction<T,R> variant (named). */
-  FuncCallStep(String name, FilterFunction<T, R> filterFn, Class<T> argClass) {
+  FuncCallStep(
+      String name, FilterFunction<T, R> filterFn, Class<T> argClass, Class<R> returnClass) {
     this.name = name;
     this.fn = null;
     this.ctxFn = null;
     this.filterFn = filterFn;
     this.argClass = argClass;
+    this.returnClass = returnClass;
   }
 
   @Override
   protected void configure(FuncTaskItemListBuilder list, Consumer<FuncCallTaskBuilder> post) {
     final Consumer<FuncCallTaskBuilder> apply =
         cb -> {
           if (ctxFn != null) {
-            cb.function(ctxFn, argClass);
+            cb.function(ctxFn, argClass, returnClass);
           } else if (filterFn != null) {
-            cb.function(filterFn, argClass);
+            cb.function(filterFn, argClass, returnClass);
           } else {
-            cb.function(fn, argClass);
+            cb.function(fn, argClass, returnClass);
           }
           post.accept(cb);
         };

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

@@ -16,6 +16,12 @@
 package io.serverlessworkflow.fluent.func.dsl;
 
 import io.cloudevents.CloudEventData;
+import io.serverlessworkflow.api.reflection.func.InstanceIdFunction;
+import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
+import io.serverlessworkflow.api.reflection.func.SerializableConsumer;
+import io.serverlessworkflow.api.reflection.func.SerializableFunction;
+import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
+import io.serverlessworkflow.api.reflection.func.UniqueIdBiFunction;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.api.types.OAuth2AuthenticationData;
 import io.serverlessworkflow.api.types.func.ContextFunction;
@@ -103,7 +109,11 @@ public static <T, V> Consumer<FuncCallTaskBuilder> fn(
    * @return a consumer that configures a {@code FuncCallTaskBuilder}
    */
   public static <T, V> Consumer<FuncCallTaskBuilder> fn(SerializableFunction<T, V> function) {
-    return f -> f.function(function, ReflectionUtils.inferInputType(function));
+    return f ->
+        f.function(
+            function,
+            ReflectionUtils.inferInputType(function),
+            ReflectionUtils.inferResultType(function));
   }
 
   /**
@@ -348,11 +358,16 @@ public static <T, R> FuncCallStep<T, R> withContext(ContextFunction<T, R> fn) {
    */
   public static <T, R> FuncCallStep<T, R> withContext(
       String name, ContextFunction<T, R> fn, Class<T> in) {
-    return new FuncCallStep<>(name, fn, in);
+    return withContext(name, fn, in, ReflectionUtils.inferResultType(fn));
+  }
+
+  public static <T, R> FuncCallStep<T, R> withContext(
+      String name, ContextFunction<T, R> fn, Class<T> in, Class<R> out) {
+    return new FuncCallStep<>(name, fn, in, out);
   }
 
   public static <T, R> FuncCallStep<T, R> withContext(String name, ContextFunction<T, R> fn) {
-    return new FuncCallStep<>(name, fn, ReflectionUtils.inferInputType(fn));
+    return withContext(name, fn, ReflectionUtils.inferInputType(fn));
   }
 
   /**
@@ -384,7 +399,12 @@ public static <T, R> FuncCallStep<T, R> withFilter(FilterFunction<T, R> fn, Clas
    */
   public static <T, R> FuncCallStep<T, R> withFilter(
       String name, FilterFunction<T, R> fn, Class<T> in) {
-    return new FuncCallStep<>(name, fn, in);
+    return withFilter(name, fn, in, ReflectionUtils.inferResultType(fn));
+  }
+
+  public static <T, R> FuncCallStep<T, R> withFilter(
+      String name, FilterFunction<T, R> fn, Class<T> in, Class<R> out) {
+    return new FuncCallStep<>(name, fn, in, out);
   }
 
   public static <T, R> FuncCallStep<T, R> withFilter(FilterFunction<T, R> fn) {
@@ -407,8 +427,13 @@ public static <T, R> FuncCallStep<T, R> withFilter(String name, FilterFunction<T
    */
   public static <T, R> FuncCallStep<T, R> withInstanceId(
       String name, InstanceIdFunction<T, R> fn, Class<T> in) {
+    return withInstanceId(name, fn, in, ReflectionUtils.inferResultType(fn));
+  }
+
+  public static <T, R> FuncCallStep<T, R> withInstanceId(
+      String name, InstanceIdFunction<T, R> fn, Class<T> in, Class<R> out) {
     ContextFunction<T, R> jcf = (payload, wctx) -> fn.apply(wctx.instanceData().id(), payload);
-    return new FuncCallStep<>(name, jcf, in);
+    return new FuncCallStep<>(name, jcf, in, out);
   }
 
   /**
@@ -463,9 +488,14 @@ static String defaultUniqueId(WorkflowContextData wctx, TaskContextData tctx) {
    */
   public static <T, R> FuncCallStep<T, R> withUniqueId(
       String name, UniqueIdBiFunction<T, R> fn, Class<T> in) {
+    return withUniqueId(name, fn, in, ReflectionUtils.inferResultType(fn));
+  }
+
+  public static <T, R> FuncCallStep<T, R> withUniqueId(
+      String name, UniqueIdBiFunction<T, R> fn, Class<T> in, Class<R> out) {
     FilterFunction<T, R> jff =
         (payload, wctx, tctx) -> fn.apply(defaultUniqueId(wctx, tctx), payload);
-    return new FuncCallStep<>(name, jff, in);
+    return new FuncCallStep<>(name, jff, in, out);
   }
 
   public static <T, R> FuncCallStep<T, R> withUniqueId(String name, UniqueIdBiFunction<T, R> fn) {
@@ -577,7 +607,23 @@ public static <T, R> FuncCallStep<T, R> agent(String name, UniqueIdBiFunction<T,
    * @return a call step which supports chaining (e.g., {@code .exportAs(...).when(...)})
    */
   public static <T, R> FuncCallStep<T, R> function(Function<T, R> fn, Class<T> inputClass) {
-    return new FuncCallStep<>(fn, inputClass);
+    return function(fn, inputClass, null);
+  }
+
+  /**
+   * Create a {@link FuncCallStep} that calls a simple Java {@link Function} with explicit input
+   * type.
+   *
+   * @param fn the function to execute at runtime
+   * @param inputClass expected input class for model conversion
+   * @param outputClass expected outputClass class for model conversion
+   * @param <T> input type
+   * @param <R> result type
+   * @return a call step which supports chaining (e.g., {@code .exportAs(...).when(...)})
+   */
+  public static <T, R> FuncCallStep<T, R> function(
+      Function<T, R> fn, Class<T> inputClass, Class<R> outputClass) {
+    return new FuncCallStep<>(fn, inputClass, outputClass);
   }
 
   /**
@@ -590,8 +636,8 @@ public static <T, R> FuncCallStep<T, R> function(Function<T, R> fn, Class<T> inp
    * @return a call step
    */
   public static <T, R> FuncCallStep<T, R> function(SerializableFunction<T, R> fn) {
-    Class<T> inputClass = ReflectionUtils.inferInputType(fn);
-    return new FuncCallStep<>(fn, inputClass);
+    return new FuncCallStep<>(
+        fn, ReflectionUtils.inferInputType(fn), ReflectionUtils.inferResultType(fn));
   }
 
   /**
@@ -604,8 +650,8 @@ public static <T, R> FuncCallStep<T, R> function(SerializableFunction<T, R> fn)
    * @return a named call step
    */
   public static <T, R> FuncCallStep<T, R> function(String name, SerializableFunction<T, R> fn) {
-    Class<T> inputClass = ReflectionUtils.inferInputType(fn);
-    return new FuncCallStep<>(name, fn, inputClass);
+    return new FuncCallStep<>(
+        name, fn, ReflectionUtils.inferInputType(fn), ReflectionUtils.inferResultType(fn));
   }
 
   /**
@@ -620,7 +666,23 @@ public static <T, R> FuncCallStep<T, R> function(String name, SerializableFuncti
    */
   public static <T, R> FuncCallStep<T, R> function(
       String name, Function<T, R> fn, Class<T> inputClass) {
-    return new FuncCallStep<>(name, fn, inputClass);
+    return new FuncCallStep<>(name, fn, inputClass, null);
+  }
+
+  /**
+   * Named variant of {@link #function(Function, Class)} with explicit input type.
+   *
+   * @param name task name
+   * @param fn the function to execute
+   * @param inputClass expected input class
+   * @param outputClass expected output class
+   * @param <T> input type
+   * @param <R> output type
+   * @return a named call step
+   */
+  public static <T, R> FuncCallStep<T, R> function(
+      String name, Function<T, R> fn, Class<T> inputClass, Class<R> outputClass) {
+    return new FuncCallStep<>(name, fn, inputClass, outputClass);
   }
 
   // ------------------  tasks ---------------- //

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

@@ -18,6 +18,8 @@
 import io.cloudevents.CloudEventData;
 import io.cloudevents.core.data.BytesCloudEventData;
 import io.cloudevents.core.data.PojoCloudEventData;
+import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
+import io.serverlessworkflow.api.reflection.func.SerializableFunction;
 import io.serverlessworkflow.api.types.func.ContextFunction;
 import io.serverlessworkflow.api.types.func.EventDataFunction;
 import io.serverlessworkflow.fluent.func.FuncEmitEventPropertiesBuilder;

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

@@ -21,6 +21,7 @@
 import io.cloudevents.core.CloudEventUtils;
 import io.cloudevents.core.data.PojoCloudEventData;
 import io.cloudevents.jackson.PojoCloudEventDataMapper;
+import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
 import io.serverlessworkflow.api.types.func.ContextPredicate;
 import io.serverlessworkflow.api.types.func.FilterPredicate;
 import io.serverlessworkflow.fluent.func.FuncEventFilterBuilder;

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

@@ -15,6 +15,9 @@
  */
 package io.serverlessworkflow.fluent.func.dsl;
 
+import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
+import io.serverlessworkflow.api.reflection.func.SerializableFunction;
+import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.api.types.func.ContextFunction;
 import io.serverlessworkflow.api.types.func.FilterFunction;

experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLUniqueIdTest.java

@@ -23,12 +23,12 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import io.serverlessworkflow.api.reflection.func.UniqueIdBiFunction;
 import io.serverlessworkflow.api.types.Task;
 import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.api.types.func.CallJava;
 import io.serverlessworkflow.api.types.func.FilterFunction;
-import io.serverlessworkflow.fluent.func.dsl.UniqueIdBiFunction;
 import io.serverlessworkflow.impl.TaskContextData;
 import io.serverlessworkflow.impl.WorkflowContextData;
 import io.serverlessworkflow.impl.WorkflowInstanceData;