KOGITO-9158 Use RestWorkItemHandler as HTTP client in Knative custom function

kogito-workitems/kogito-rest-workitem/src/main/java/org/kogito/workitem/rest/RestWorkItemHandler.java

@@ -17,6 +17,7 @@
 
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -82,6 +83,8 @@ public class RestWorkItemHandler implements KogitoWorkItemHandler {
     public static final String PATH_PARAM_RESOLVER = "PathParamResolver";
     public static final String AUTH_METHOD = "AuthMethod";
 
+    public static final String REQUEST_TIMEOUT_IN_MILLIS = "RequestTimeout";
+
     public static final int DEFAULT_PORT = 80;
     public static final int DEFAULT_SSL_PORT = 443;
 
@@ -174,14 +177,38 @@ public void executeWorkItem(KogitoWorkItem workItem, KogitoWorkItemManager manag
         requestDecorators.forEach(d -> d.decorate(workItem, parameters, request));
         authDecorators.forEach(d -> d.decorate(workItem, parameters, request));
         paramsDecorator.decorate(workItem, parameters, request);
-        HttpResponse<Buffer> response = method.equals(HttpMethod.POST) || method.equals(HttpMethod.PUT) ? request.sendJsonAndAwait(bodyBuilder.apply(parameters)) : request.sendAndAwait();
+        Duration requestTimeout = getRequestTimeout(parameters);
+        HttpResponse<Buffer> response = method.equals(HttpMethod.POST) || method.equals(HttpMethod.PUT)
+                ? sendJson(request, bodyBuilder.apply(parameters), requestTimeout)
+                : send(request, requestTimeout);
         int statusCode = response.statusCode();
         if (statusCode < 200 || statusCode >= 300) {
             throw new WorkItemExecutionException(Integer.toString(statusCode), "Request for endpoint " + endPoint + " failed with message: " + response.statusMessage());
         }
         manager.completeWorkItem(workItem.getStringId(), Collections.singletonMap(RESULT, resultHandler.apply(response, targetInfo)));
     }
 
+    private static HttpResponse<Buffer> sendJson(HttpRequest<Buffer> request, Object body, Duration requestTimeout) {
+        if (requestTimeout == null) {
+            return request.sendJsonAndAwait(body);
+        } else {
+            return request.sendJson(body).await().atMost(requestTimeout);
+        }
+    }
+
+    private static HttpResponse<Buffer> send(HttpRequest<Buffer> request, Duration requestTimeout) {
+        if (requestTimeout == null) {
+            return request.sendAndAwait();
+        } else {
+            return request.send().await().atMost(requestTimeout);
+        }
+    }
+
+    private static Duration getRequestTimeout(Map<String, Object> parameters) {
+        Long requestTimeoutInMillis = getParam(parameters, REQUEST_TIMEOUT_IN_MILLIS, Long.class, null);
+        return requestTimeoutInMillis == null ? null : Duration.ofMillis(requestTimeoutInMillis);
+    }
+
     private Class<?> getTargetInfo(KogitoWorkItem workItem) {
         String varName = ((WorkItemNode) ((WorkItemNodeInstance) workItem.getNodeInstance()).getNode()).getIoSpecification().getOutputMappingBySources().get(RESULT);
         if (varName != null) {

quarkus/addons/fabric8-kubernetes-service-catalog/test-utils/src/main/java/org/kie/kogito/addons/quarkus/k8s/test/utils/KnativeResourceDiscoveryTestUtil.java

@@ -32,7 +32,7 @@ public static void createServiceIfNotExists(KubernetesServer k8sServer, String k
 
     @SuppressWarnings("deprecation") // Quarkus LTS 2.13 compatibility
     public static void createServiceIfNotExists(KubernetesServer k8sServer, String knativeYaml, String namespace, String serviceName, String remoteServiceUrl) {
-        if (k8sServer.getClient().services().inNamespace("test").withName(serviceName).get() == null) {
+        if (k8sServer.getClient().services().inNamespace(namespace).withName(serviceName).get() == null) {
             KnativeClient knativeClient = k8sServer.getClient().adapt(KnativeClient.class);
 
             Service service = knativeClient.services()

quarkus/addons/knative/serving/deployment/pom.xml

@@ -29,6 +29,10 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-vertx-deployment</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.kie.kogito</groupId>
+            <artifactId>kogito-serverless-workflow-rest-parser</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

quarkus/addons/knative/serving/deployment/src/main/java/org/kie/kogito/addons/quarkus/knative/serving/deployment/customfunctions/KnativeTypeHandler.java

@@ -15,29 +15,86 @@
  */
 package org.kie.kogito.addons.quarkus.knative.serving.deployment.customfunctions;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+
+import org.jbpm.compiler.canonical.descriptors.TaskDescriptor;
 import org.jbpm.ruleflow.core.RuleFlowNodeContainerFactory;
+import org.jbpm.ruleflow.core.factory.NodeFactory;
 import org.jbpm.ruleflow.core.factory.WorkItemNodeFactory;
+import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.CloudEventKnativeParamsDecorator;
 import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.KnativeWorkItemHandler;
+import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.Operation;
+import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.PlainJsonKnativeParamsDecorator;
 import org.kie.kogito.serverless.workflow.parser.ParserContext;
+import org.kie.kogito.serverless.workflow.parser.VariableInfo;
 import org.kie.kogito.serverless.workflow.parser.types.WorkItemTypeHandler;
+import org.kie.kogito.serverless.workflow.suppliers.ParamsRestBodyBuilderSupplier;
+import org.kogito.workitem.rest.RestWorkItemHandler;
+
+import com.github.javaparser.ast.expr.Expression;
 
 import io.serverlessworkflow.api.Workflow;
 import io.serverlessworkflow.api.functions.FunctionDefinition;
+import io.serverlessworkflow.api.functions.FunctionRef;
+import io.vertx.core.http.HttpMethod;
 
+import static org.kie.kogito.addons.quarkus.knative.serving.customfunctions.KnativeWorkItemHandler.PAYLOAD_FIELDS_PROPERTY_NAME;
 import static org.kie.kogito.serverless.workflow.parser.FunctionTypeHandlerFactory.trimCustomOperation;
+import static org.kie.kogito.serverless.workflow.utils.ServerlessWorkflowUtils.runtimeRestApi;
 
 public class KnativeTypeHandler extends WorkItemTypeHandler {
 
+    private static final String DEFAULT_REQUEST_TIMEOUT_VALUE = "10000";
+
+    @Override
+    public NodeFactory<?, ?> getActionNode(Workflow workflow, ParserContext context,
+            RuleFlowNodeContainerFactory<?, ?> embeddedSubProcess, FunctionDefinition functionDef,
+            FunctionRef functionRef, VariableInfo varInfo) {
+        validateArgs(functionRef);
+
+        WorkItemNodeFactory<?> node = buildWorkItem(embeddedSubProcess, context, varInfo.getInputVar(), varInfo.getOutputVar())
+                .name(functionDef.getName());
+
+        if (functionRef.getArguments() != null && !functionRef.getArguments().isEmpty()) {
+            List<String> payloadFields = new ArrayList<>();
+            functionRef.getArguments().fieldNames().forEachRemaining(payloadFields::add);
+            if (!payloadFields.isEmpty()) {
+                node.workParameter(PAYLOAD_FIELDS_PROPERTY_NAME, String.join(";", payloadFields));
+            }
+        }
+
+        return addFunctionArgs(workflow,
+                fillWorkItemHandler(workflow, context, node, functionDef),
+                functionRef);
+    }
+
     @Override
-    protected <T extends RuleFlowNodeContainerFactory<T, ?>> WorkItemNodeFactory<T> fillWorkItemHandler(Workflow workflow,
-            ParserContext context,
-            WorkItemNodeFactory<T> node,
-            FunctionDefinition functionDef) {
+    protected <T extends RuleFlowNodeContainerFactory<T, ?>> WorkItemNodeFactory<T> fillWorkItemHandler(
+            Workflow workflow, ParserContext context, WorkItemNodeFactory<T> node, FunctionDefinition functionDef) {
         if (functionDef.getMetadata() != null) {
             functionDef.getMetadata().forEach(node::metaData);
         }
 
-        return node.workName(KnativeWorkItemHandler.NAME).metaData(KnativeWorkItemHandler.OPERATION_PROPERTY_NAME, trimCustomOperation(functionDef));
+        Operation operation = Operation.parse(trimCustomOperation(functionDef));
+
+        if (operation.isCloudEvent()) {
+            node.workParameter(RestWorkItemHandler.PARAMS_DECORATOR, CloudEventKnativeParamsDecorator.class.getName());
+        } else {
+            node.workParameter(RestWorkItemHandler.PARAMS_DECORATOR, PlainJsonKnativeParamsDecorator.class.getName());
+        }
+
+        Supplier<Expression> requestTimeout = runtimeRestApi(functionDef, "timeout",
+                context.getContext(), String.class, DEFAULT_REQUEST_TIMEOUT_VALUE);
+
+        return node.workParameter(KnativeWorkItemHandler.SERVICE_PROPERTY_NAME, operation.getService())
+                .workParameter(KnativeWorkItemHandler.PATH_PROPERTY_NAME, operation.getPath())
+                .workParameter(RestWorkItemHandler.BODY_BUILDER, new ParamsRestBodyBuilderSupplier())
+                .workParameter(RestWorkItemHandler.METHOD, HttpMethod.POST)
+                .workParameter(RestWorkItemHandler.REQUEST_TIMEOUT_IN_MILLIS, requestTimeout)
+                .metaData(TaskDescriptor.KEY_WORKITEM_TYPE, RestWorkItemHandler.REST_TASK_TYPE)
+                .workName(KnativeWorkItemHandler.NAME);
     }
 
     @Override

quarkus/addons/knative/serving/integration-tests/pom.xml

@@ -0,0 +1,136 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.kie.kogito</groupId>
+        <artifactId>kogito-addons-quarkus-knative-serving-parent</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>kogito-addons-quarkus-knative-serving-integration-tests</artifactId>
+    <name>Kogito Add-On Knative Serving - Integration Tests</name>
+    <description>Knative Serving Kogito Add-On Integration Tests.</description>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.kie.kogito</groupId>
+                <artifactId>kogito-quarkus-bom</artifactId>
+                <version>${project.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-smallrye-openapi</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.kie.kogito</groupId>
+            <artifactId>kogito-quarkus-serverless-workflow</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-resteasy</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-resteasy-jackson</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.kie.kogito</groupId>
+            <artifactId>kogito-addons-quarkus-knative-serving</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.kie.kogito</groupId>
+            <artifactId>kogito-addons-quarkus-kubernetes</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.kie.kogito</groupId>
+            <artifactId>kogito-addons-quarkus-fabric8-kubernetes-service-catalog</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kubernetes</artifactId>
+        </dependency>
+
+        <!-- Test -->
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-junit5</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.rest-assured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.github.tomakehurst</groupId>
+            <artifactId>wiremock-jre8</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-test-kubernetes-client</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.kie.kogito</groupId>
+            <artifactId>kogito-addons-quarkus-fabric8-kubernetes-service-catalog-test-utils</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <groupId>io.quarkus</groupId>
+                    <artifactId>quarkus-maven-plugin</artifactId>
+                    <configuration>
+                        <noDeps>true</noDeps>
+                        <skip>${skipTests}</skip>
+                    </configuration>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+        <plugins>
+            <plugin>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>build</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+    <profiles>
+        <profile>
+            <id>native</id>
+            <activation>
+                <property>
+                    <name>native</name>
+                </property>
+            </activation>
+            <properties>
+                <quarkus.package.type>native</quarkus.package.type>
+            </properties>
+        </profile>
+    </profiles>
+</project>

quarkus/addons/knative/serving/integration-tests/src/main/resources/arrayKnativeFunction.sw.json

@@ -0,0 +1,34 @@
+{
+  "id": "arrayKnativeFunction",
+  "version": "1.0",
+  "name": "Test Knative function",
+  "description": "This workflow tests a Knative function",
+  "start": "invokeFunction",
+  "functions": [
+    {
+      "name": "greet",
+      "type": "custom",
+      "operation": "knative:services.v1.serving.knative.dev/default/serverless-workflow-greeting-quarkus?path=/arrayFunction"
+    }
+  ],
+  "states": [
+    {
+      "name": "invokeFunction",
+      "type": "operation",
+      "actions": [
+        {
+          "functionRef": {
+            "refName": "greet",
+            "arguments": {
+              "array": [
+                "Javierito",
+                "Pepito"
+              ]
+            }
+          }
+        }
+      ],
+      "end": true
+    }
+  ]
+}

quarkus/addons/knative/serving/integration-tests/src/main/resources/cloudEventKnativeFunction.sw.json

@@ -0,0 +1,38 @@
+{
+  "id": "cloudEvent",
+  "version": "1.0",
+  "name": "Test Knative function",
+  "description": "This workflow tests a Knative function",
+  "start": "invokeFunction",
+  "functions": [
+    {
+      "name": "greet",
+      "type": "custom",
+      "operation": "knative:services.v1.serving.knative.dev/default/serverless-workflow-greeting-quarkus?asCloudEvent=true&path=/cloud-event"
+    }
+  ],
+  "states": [
+    {
+      "name": "invokeFunction",
+      "type": "operation",
+      "actions": [
+        {
+          "functionRef": {
+            "refName": "greet",
+            "arguments": {
+              "specversion": "1.0",
+              "id": "42",
+              "source": "https://localhost:8080",
+              "type": "org.kie.kogito.test",
+              "data": {
+                "org": "Acme",
+                "project": "Kogito"
+              }
+            }
+          }
+        }
+      ],
+      "end": true
+    }
+  ]
+}