[Fix #934] Implementing function catalog

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

Author: fjtirado

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

@@ -327,7 +327,7 @@ public static <T, R> FuncCallStep<T, R> withInstanceId(
   }
 
   /**
-   * Named variant of {@link #withContext(JavaContextFunction, Class)}.
+   * NamedObject variant of {@link #withContext(JavaContextFunction, Class)}.
    *
    * @param name task name
    * @param fn context-aware function
@@ -359,7 +359,7 @@ public static <T, R> FuncCallStep<T, R> withFilter(JavaFilterFunction<T, R> fn,
   }
 
   /**
-   * Named variant of {@link #withFilter(JavaFilterFunction, Class)}.
+   * NamedObject variant of {@link #withFilter(JavaFilterFunction, Class)}.
    *
    * @param name task name
    * @param fn context-aware filter function
@@ -374,7 +374,7 @@ public static <T, R> FuncCallStep<T, R> withFilter(
   }
 
   /**
-   * Named variant of {@link #withInstanceId(InstanceIdBiFunction, Class)}.
+   * NamedObject variant of {@link #withInstanceId(InstanceIdBiFunction, Class)}.
    *
    * @param name task name
    * @param fn instance-id-aware function
@@ -450,7 +450,7 @@ public static <T> ConsumeStep<T> consume(Consumer<T> consumer, Class<T> clazz) {
   }
 
   /**
-   * Named variant of {@link #consume(Consumer, Class)}.
+   * NamedObject variant of {@link #consume(Consumer, Class)}.
    *
    * @param name task name
    * @param consumer side-effect function
@@ -480,7 +480,7 @@ public static <T, R> FuncCallStep<T, R> agent(UniqueIdBiFunction<T, R> fn, Class
   }
 
   /**
-   * Named agent-style sugar. See {@link #agent(UniqueIdBiFunction, Class)}.
+   * NamedObject agent-style sugar. See {@link #agent(UniqueIdBiFunction, Class)}.
    *
    * <p>Signature expected: {@code (uniqueId, payload) -> result}
    *
@@ -511,7 +511,7 @@ public static <T, R> FuncCallStep<T, R> function(Function<T, R> fn) {
   }
 
   /**
-   * Named variant of {@link #function(Function)} with inferred input type.
+   * NamedObject variant of {@link #function(Function)} with inferred input type.
    *
    * @param name task name
    * @param fn the function to execute
@@ -525,7 +525,7 @@ public static <T, R> FuncCallStep<T, R> function(String name, Function<T, R> fn)
   }
 
   /**
-   * Named variant of {@link #function(Function, Class)} with explicit input type.
+   * NamedObject variant of {@link #function(Function, Class)} with explicit input type.
    *
    * @param name task name
    * @param fn the function to execute
@@ -565,7 +565,7 @@ public static EmitStep emit(Consumer<FuncEmitTaskBuilder> cfg) {
   }
 
   /**
-   * Named variant of {@link #emit(Consumer)}.
+   * NamedObject variant of {@link #emit(Consumer)}.
    *
    * @param name task name
    * @param cfg emit builder configurer
@@ -588,7 +588,7 @@ public static <T> EmitStep emit(String type, Function<T, CloudEventData> fn) {
   }
 
   /**
-   * Named variant of {@link #emit(String, Function)}.
+   * NamedObject variant of {@link #emit(String, Function)}.
    *
    * @param name task name
    * @param type CloudEvent type
@@ -664,7 +664,7 @@ public static ListenStep listen(FuncListenSpec spec) {
   }
 
   /**
-   * Named variant of {@link #listen(FuncListenSpec)}.
+   * NamedObject variant of {@link #listen(FuncListenSpec)}.
    *
    * @param name task name
    * @param spec listen spec
@@ -709,7 +709,7 @@ public static FuncTaskConfigurer switchCase(SwitchCaseConfigurer... cases) {
   }
 
   /**
-   * Named variant of {@link #switchCase(SwitchCaseConfigurer...)}.
+   * NamedObject variant of {@link #switchCase(SwitchCaseConfigurer...)}.
    *
    * @param taskName task name
    * @param cases case configurers
@@ -1078,7 +1078,7 @@ public static FuncCallOpenAPIStep openapi() {
   }
 
   /**
-   * Named variant of {@link #openapi()}.
+   * NamedObject variant of {@link #openapi()}.
    *
    * @param name task name to be used when the spec is attached via {@link
    *     #call(FuncCallOpenAPIStep)}
@@ -1109,7 +1109,7 @@ public static FuncCallHttpStep http() {
   }
 
   /**
-   * Named variant of {@link #http()}.
+   * NamedObject variant of {@link #http()}.
    *
    * @param name task name to be used when the spec is attached via {@link #call(FuncCallHttpStep)}
    * @return a new named {@link FuncCallHttpStep}

impl/core/src/main/java/io/serverlessworkflow/impl/NamedObject.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;
+
+public interface NamedObject {
+  String name();
+}

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

@@ -21,6 +21,8 @@
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.additional.NamedWorkflowAdditionalObject;
 import io.serverlessworkflow.impl.additional.WorkflowAdditionalObject;
+import io.serverlessworkflow.impl.catalogs.CatalogNavigator;
+import io.serverlessworkflow.impl.catalogs.NamedCatalogNavigator;
 import io.serverlessworkflow.impl.config.ConfigManager;
 import io.serverlessworkflow.impl.config.ConfigSecretManager;
 import io.serverlessworkflow.impl.config.SecretManager;
@@ -76,6 +78,7 @@ public class WorkflowApplication implements AutoCloseable {
   private final SecretManager secretManager;
   private final SchedulerListener schedulerListener;
   private final Optional<URITemplateResolver> templateResolver;
+  private final Map<String, CatalogNavigator> catalogRegistry;
 
   private WorkflowApplication(Builder builder) {
     this.taskFactory = builder.taskFactory;
@@ -98,6 +101,7 @@ private WorkflowApplication(Builder builder) {
     this.configManager = builder.configManager;
     this.secretManager = builder.secretManager;
     this.templateResolver = builder.templateResolver;
+    this.catalogRegistry = builder.catalogRegistry;
   }
 
   public TaskExecutorFactory taskFactory() {
@@ -178,10 +182,13 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private ConfigManager configManager;
     private SchedulerListener schedulerListener;
     private Optional<URITemplateResolver> templateResolver;
+    private Map<String, CatalogNavigator> catalogRegistry = new HashMap<>();
 
     private Builder() {
       ServiceLoader.load(NamedWorkflowAdditionalObject.class)
           .forEach(a -> additionalObjects.put(a.name(), a));
+      ServiceLoader.load(NamedCatalogNavigator.class)
+          .forEach(a -> catalogRegistry.put(a.name(), a));
     }
 
     public Builder withListener(WorkflowExecutionListener listener) {
@@ -270,6 +277,11 @@ public Builder withModelFactory(WorkflowModelFactory modelFactory) {
       return this;
     }
 
+    public Builder withCatalogRegistry(String name, CatalogNavigator catalog) {
+      catalogRegistry.put(name, catalog);
+      return this;
+    }
+
     public WorkflowApplication build() {
       if (modelFactory == null) {
         modelFactory =
@@ -328,6 +340,7 @@ public WorkflowApplication build() {
                 .findFirst()
                 .orElseGet(() -> new ConfigSecretManager(configManager));
       }
+
       templateResolver = ServiceLoader.load(URITemplateResolver.class).findFirst();
       return new WorkflowApplication(this);
     }
@@ -406,6 +419,10 @@ public Optional<URITemplateResolver> templateResolver() {
     return templateResolver;
   }
 
+  public Map<String, CatalogNavigator> catalogRegistry() {
+    return catalogRegistry;
+  }
+
   public <T> Optional<T> additionalObject(
       String name, WorkflowContext workflowContext, TaskContext taskContext) {
     return Optional.ofNullable(additionalObjects.get(name))

impl/core/src/main/java/io/serverlessworkflow/impl/additional/NamedWorkflowAdditionalObject.java

@@ -15,6 +15,7 @@
  */
 package io.serverlessworkflow.impl.additional;
 
-public interface NamedWorkflowAdditionalObject<T> extends WorkflowAdditionalObject<T> {
-  String name();
-}
+import io.serverlessworkflow.impl.NamedObject;
+
+public interface NamedWorkflowAdditionalObject<T>
+    extends WorkflowAdditionalObject<T>, NamedObject {}

impl/core/src/main/java/io/serverlessworkflow/impl/catalogs/CatalogNavigator.java

@@ -0,0 +1,27 @@
+/*
+ * 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.catalogs;
+
+import io.serverlessworkflow.api.types.Catalog;
+import io.serverlessworkflow.api.types.Task;
+import io.serverlessworkflow.impl.ServicePriority;
+import java.util.Optional;
+
+public interface CatalogNavigator extends ServicePriority {
+  Optional<Task> resolveTask(Catalog catalog, String functionName);
+
+  Optional<Task> resolveTask(String functionEndpoint);
+}

impl/core/src/main/java/io/serverlessworkflow/impl/catalogs/NamedCatalogNavigator.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.catalogs;
+
+import io.serverlessworkflow.impl.NamedObject;
+
+public interface NamedCatalogNavigator extends CatalogNavigator, NamedObject {}

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

@@ -16,44 +16,85 @@
 package io.serverlessworkflow.impl.executors;
 
 import io.serverlessworkflow.api.types.CallFunction;
+import io.serverlessworkflow.api.types.Catalog;
 import io.serverlessworkflow.api.types.FunctionArguments;
 import io.serverlessworkflow.api.types.Task;
 import io.serverlessworkflow.api.types.TaskBase;
+import io.serverlessworkflow.api.types.Use;
+import io.serverlessworkflow.api.types.UseCatalogs;
+import io.serverlessworkflow.api.types.UseFunctions;
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowUtils;
 import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.catalogs.CatalogNavigator;
 import java.util.Map;
 import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class CallFunctionExecutor implements CallableTaskBuilder<CallFunction> {
 
+  private static final Logger logger = LoggerFactory.getLogger(CallFunctionExecutor.class);
   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();
+
+    Use use = definition.workflow().getUse();
     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 (use != null) {
+      UseFunctions functions = use.getFunctions();
+      if (functions != null) {
+        function = functions.getAdditionalProperties().get(functionName);
+      }
+      if (function == null) {
+        int indexOf = functionName.indexOf('@');
+        if (indexOf > 0) {
+          String catalogName = functionName.substring(indexOf + 1);
+          UseCatalogs catalogs = use.getCatalogs();
+          if (catalogs != null) {
+            Catalog catalog = catalogs.getAdditionalProperties().get(catalogName);
+            CatalogNavigator catalogNavigator =
+                definition.application().catalogRegistry().get(catalogName);
+            if (catalogNavigator == null) {
+              throw new IllegalArgumentException(
+                  "There is not catalog registered for name " + catalogName);
+            }
+            function =
+                catalogNavigator
+                    .resolveTask(catalog, functionName.substring(0, indexOf))
+                    .orElseThrow(
+                        () ->
+                            new IllegalArgumentException(
+                                "Cannot find function "
+                                    + functionName
+                                    + " in catalog "
+                                    + catalogName));
+          }
+        }
+      }
     }
     if (function == null) {
-      // TODO search in catalog
-      throw new UnsupportedOperationException("Function Catalog not supported yet");
+      // asume name is direct pointer to catalog file
+      function =
+          definition.application().catalogRegistry().values().stream()
+              .flatMap(c -> c.resolveTask(functionName).stream())
+              .findFirst()
+              .orElseThrow(
+                  () -> new IllegalArgumentException("Cannot find function name " + functionName));
     }
     executorBuilder =
         definition.application().taskFactory().getTaskExecutor(position, function, definition);
+    FunctionArguments functionArgs = task.getWith();
+    args =
+        functionArgs != null
+            ? WorkflowUtils.buildMapResolver(
+                definition.application(), functionArgs.getAdditionalProperties())
+            : (w, t, m) -> Map.of();
   }
 
   @Override