[Fix #1002] Implementing custom inline function call

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaConsumerCallExecutor.java

@@ -21,23 +21,28 @@
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 
-public class JavaConsumerCallExecutor<T> implements CallableTask<CallJava.CallJavaConsumer<T>> {
+public class JavaConsumerCallExecutor<T>
+    implements CallableTaskBuilder<CallJava.CallJavaConsumer<T>> {
 
   private Consumer<T> consumer;
-  private Optional<Class<T>> inputClass = Optional.empty();
+  private Optional<Class<T>> inputClass;
 
-  public void init(CallJava.CallJavaConsumer<T> task, WorkflowDefinition definition) {
+  public void init(
+      CallJava.CallJavaConsumer<T> task,
+      WorkflowDefinition definition,
+      WorkflowMutablePosition position) {
     consumer = task.consumer();
     inputClass = task.inputClass();
   }
 
-  @Override
-  public CompletableFuture<WorkflowModel> apply(
+  private CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     T typed = JavaFuncUtils.convertT(input, inputClass);
     consumer.accept(typed);
@@ -48,4 +53,9 @@ public CompletableFuture<WorkflowModel> apply(
   public boolean accept(Class<? extends TaskBase> clazz) {
     return CallJava.CallJavaConsumer.class.isAssignableFrom(clazz);
   }
+
+  @Override
+  public CallableTask build() {
+    return this::apply;
+  }
 }

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaContextFunctionCallExecutor.java

@@ -17,29 +17,44 @@
 
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.api.types.func.CallJava;
+import io.serverlessworkflow.api.types.func.CallJava.CallJavaContextFunction;
 import io.serverlessworkflow.api.types.func.JavaContextFunction;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 public class JavaContextFunctionCallExecutor<T, V>
-    implements CallableTask<CallJava.CallJavaContextFunction<T, V>> {
+    implements CallableTaskBuilder<CallJava.CallJavaContextFunction<T, V>> {
 
   private JavaContextFunction<T, V> function;
   private Optional<Class<T>> inputClass;
 
   @Override
-  public void init(CallJava.CallJavaContextFunction<T, V> task, WorkflowDefinition definition) {
+  public boolean accept(Class<? extends TaskBase> clazz) {
+    return CallJava.CallJavaContextFunction.class.isAssignableFrom(clazz);
+  }
+
+  @Override
+  public void init(
+      CallJavaContextFunction<T, V> task,
+      WorkflowDefinition definition,
+      WorkflowMutablePosition position) {
     this.function = task.function();
     this.inputClass = task.inputClass();
   }
 
   @Override
-  public CompletableFuture<WorkflowModel> apply(
+  public CallableTask build() {
+    return this::apply;
+  }
+
+  private CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     return CompletableFuture.completedFuture(
         workflowContext
@@ -49,9 +64,4 @@ public CompletableFuture<WorkflowModel> apply(
             .fromAny(
                 input, function.apply(JavaFuncUtils.convertT(input, inputClass), workflowContext)));
   }
-
-  @Override
-  public boolean accept(Class<? extends TaskBase> clazz) {
-    return CallJava.CallJavaContextFunction.class.isAssignableFrom(clazz);
-  }
 }

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaFilterFunctionCallExecutor.java

@@ -17,29 +17,25 @@
 
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.api.types.func.CallJava;
+import io.serverlessworkflow.api.types.func.CallJava.CallJavaFilterFunction;
 import io.serverlessworkflow.api.types.func.JavaFilterFunction;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 public class JavaFilterFunctionCallExecutor<T, V>
-    implements CallableTask<CallJava.CallJavaFilterFunction<T, V>> {
+    implements CallableTaskBuilder<CallJava.CallJavaFilterFunction<T, V>> {
 
   private JavaFilterFunction<T, V> function;
   private Optional<Class<T>> inputClass;
 
-  @Override
-  public void init(CallJava.CallJavaFilterFunction<T, V> task, WorkflowDefinition definition) {
-    this.function = task.function();
-    this.inputClass = task.inputClass();
-  }
-
-  @Override
-  public CompletableFuture<WorkflowModel> apply(
+  private CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     return CompletableFuture.completedFuture(
         workflowContext
@@ -56,4 +52,18 @@ public CompletableFuture<WorkflowModel> apply(
   public boolean accept(Class<? extends TaskBase> clazz) {
     return CallJava.CallJavaFilterFunction.class.isAssignableFrom(clazz);
   }
+
+  @Override
+  public void init(
+      CallJavaFilterFunction<T, V> task,
+      WorkflowDefinition definition,
+      WorkflowMutablePosition position) {
+    this.function = task.function();
+    this.inputClass = task.inputClass();
+  }
+
+  @Override
+  public CallableTask build() {
+    return this::apply;
+  }
 }

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaFunctionCallExecutor.java

@@ -17,28 +17,25 @@
 
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.api.types.func.CallJava;
+import io.serverlessworkflow.api.types.func.CallJava.CallJavaFunction;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 
 public class JavaFunctionCallExecutor<T, V>
-    implements CallableTask<CallJava.CallJavaFunction<T, V>> {
+    implements CallableTaskBuilder<CallJava.CallJavaFunction<T, V>> {
 
   private Function<T, V> function;
   private Optional<Class<T>> inputClass;
 
-  public void init(CallJava.CallJavaFunction<T, V> task, WorkflowDefinition definition) {
-    function = task.function();
-    inputClass = task.inputClass();
-  }
-
-  @Override
-  public CompletableFuture<WorkflowModel> apply(
+  private CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     return CompletableFuture.completedFuture(
         workflowContext
@@ -52,4 +49,18 @@ public CompletableFuture<WorkflowModel> apply(
   public boolean accept(Class<? extends TaskBase> clazz) {
     return CallJava.CallJavaFunction.class.isAssignableFrom(clazz);
   }
+
+  @Override
+  public void init(
+      CallJavaFunction<T, V> task,
+      WorkflowDefinition definition,
+      WorkflowMutablePosition position) {
+    function = task.function();
+    inputClass = task.inputClass();
+  }
+
+  @Override
+  public CallableTask build() {
+    return this::apply;
+  }
 }

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaLoopFunctionCallExecutor.java

@@ -19,27 +19,25 @@
 
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.api.types.func.CallJava;
+import io.serverlessworkflow.api.types.func.CallJava.CallJavaLoopFunction;
 import io.serverlessworkflow.api.types.func.LoopFunction;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelFactory;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
 import java.util.concurrent.CompletableFuture;
 
-public class JavaLoopFunctionCallExecutor implements CallableTask<CallJava.CallJavaLoopFunction> {
+public class JavaLoopFunctionCallExecutor
+    implements CallableTaskBuilder<CallJava.CallJavaLoopFunction> {
 
   private LoopFunction function;
   private String varName;
 
-  public void init(CallJava.CallJavaLoopFunction task, WorkflowDefinition definition) {
-    function = task.function();
-    varName = task.varName();
-  }
-
-  @Override
-  public CompletableFuture<WorkflowModel> apply(
+  private CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     WorkflowModelFactory modelFactory = workflowContext.definition().application().modelFactory();
     return CompletableFuture.completedFuture(
@@ -51,7 +49,18 @@ public CompletableFuture<WorkflowModel> apply(
 
   @Override
   public boolean accept(Class<? extends TaskBase> clazz) {
-
     return CallJava.CallJavaLoopFunction.class.isAssignableFrom(clazz);
   }
+
+  @Override
+  public void init(
+      CallJavaLoopFunction task, WorkflowDefinition definition, WorkflowMutablePosition position) {
+    function = task.function();
+    varName = task.varName();
+  }
+
+  @Override
+  public CallableTask build() {
+    return this::apply;
+  }
 }

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/JavaLoopFunctionIndexCallExecutor.java

@@ -19,30 +19,26 @@
 
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.api.types.func.CallJava;
+import io.serverlessworkflow.api.types.func.CallJava.CallJavaLoopFunctionIndex;
 import io.serverlessworkflow.api.types.func.LoopFunctionIndex;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelFactory;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.executors.CallableTask;
+import io.serverlessworkflow.impl.executors.CallableTaskBuilder;
 import java.util.concurrent.CompletableFuture;
 
 public class JavaLoopFunctionIndexCallExecutor
-    implements CallableTask<CallJava.CallJavaLoopFunctionIndex> {
+    implements CallableTaskBuilder<CallJava.CallJavaLoopFunctionIndex> {
 
   private LoopFunctionIndex function;
   private String varName;
   private String indexName;
 
-  public void init(CallJava.CallJavaLoopFunctionIndex task, WorkflowDefinition definition) {
-    function = task.function();
-    varName = task.varName();
-    indexName = task.indexName();
-  }
-
-  @Override
-  public CompletableFuture<WorkflowModel> apply(
+  private CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
     WorkflowModelFactory modelFactory = workflowContext.definition().application().modelFactory();
 
@@ -59,4 +55,19 @@ public CompletableFuture<WorkflowModel> apply(
   public boolean accept(Class<? extends TaskBase> clazz) {
     return CallJava.CallJavaLoopFunctionIndex.class.isAssignableFrom(clazz);
   }
+
+  @Override
+  public void init(
+      CallJavaLoopFunctionIndex task,
+      WorkflowDefinition definition,
+      WorkflowMutablePosition position) {
+    function = task.function();
+    varName = task.varName();
+    indexName = task.indexName();
+  }
+
+  @Override
+  public CallableTask build() {
+    return this::apply;
+  }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/executors/CallFunctionExecutor.java

@@ -0,0 +1,75 @@
+/*
+ * 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.executors;
+
+import io.serverlessworkflow.api.types.CallFunction;
+import io.serverlessworkflow.api.types.FunctionArguments;
+import io.serverlessworkflow.api.types.Task;
+import io.serverlessworkflow.api.types.TaskBase;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowMutablePosition;
+import io.serverlessworkflow.impl.WorkflowUtils;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import java.util.Map;
+import java.util.Optional;
+
+public class CallFunctionExecutor implements CallableTaskBuilder<CallFunction> {
+
+  private TaskExecutorBuilder<? extends TaskBase> executorBuilder;
+  private WorkflowValueResolver<Map<String, Object>> args;
+
+  @Override
+  public void init(
+      CallFunction task, WorkflowDefinition definition, WorkflowMutablePosition position) {
+    String functionName = task.getCall();
+    FunctionArguments functionArgs = task.getWith();
+    args =
+        functionArgs != null
+            ? WorkflowUtils.buildMapResolver(
+                definition.application(), functionArgs.getAdditionalProperties())
+            : (w, t, m) -> Map.of();
+    Task function = null;
+    if (definition.workflow().getUse() != null
+        && definition.workflow().getUse().getFunctions() != null
+        && definition.workflow().getUse().getFunctions().getAdditionalProperties() != null) {
+      function =
+          definition.workflow().getUse().getFunctions().getAdditionalProperties().get(functionName);
+    }
+    if (function == null) {
+      // TODO search in catalog
+      throw new UnsupportedOperationException("Function Catalog not supported yet");
+    }
+    executorBuilder =
+        definition.application().taskFactory().getTaskExecutor(position, function, definition);
+  }
+
+  @Override
+  public boolean accept(Class<? extends TaskBase> clazz) {
+    return clazz.equals(CallFunction.class);
+  }
+
+  @Override
+  public CallableTask build() {
+    TaskExecutor<? extends TaskBase> executor = executorBuilder.build();
+    return (w, t, m) ->
+        executor
+            .apply(
+                w,
+                Optional.of(t),
+                w.definition().application().modelFactory().fromAny(args.apply(w, t, m)))
+            .thenApply(o -> o.output());
+  }
+}