Repository: serverlessworkflow/sdk-java
Pull request #450 (2024-10-28 17:25:57)

Serverless Workflow reference implementation (skeleton)
This is the skeleton of a reference implementation of Serverless Workflows spec, which just implement the CallHttp task. 
The idea is to incrementally add functionality to it in baby steps. 
The purpose of the exercise is to verify that the SDK is usable. 


Patch:
diff --git a/impl/pom.xml b/impl/pom.xml
new file mode 100644
index 0000000..32ee86a
--- /dev/null
+++ b/impl/pom.xml
@@ -0,0 +1,73 @@
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>io.serverlessworkflow</groupId>
+    <artifactId>serverlessworkflow-parent</artifactId>
+    <version>7.0.0-SNAPSHOT</version>
+  </parent>
+  <artifactId>serverlessworkflow-impl</artifactId>
+  <properties>
+      <version.org.glassfish.jersey>3.1.9</version.org.glassfish.jersey>
+  </properties>
+  <dependencies>
+    <dependency>
+      <groupId>io.serverlessworkflow</groupId>
+      <artifactId>serverlessworkflow-api</artifactId>
+      <version>7.0.0-SNAPSHOT</version>
+    </dependency>
+    <dependency>
+      <groupId>org.glassfish.jersey.core</groupId>
+      <artifactId>jersey-client</artifactId>
+      <version>${version.org.glassfish.jersey}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.glassfish.jersey.media</groupId>
+      <artifactId>jersey-media-json-jackson</artifactId>
+      <version>${version.org.glassfish.jersey}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-params</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+  <build>
+    <plugins>
+        <plugin>
+            <groupId>com.spotify.fmt</groupId>
+            <artifactId>fmt-maven-plugin</artifactId>
+            <configuration>
+                <sourceDirectory>src/main/java</sourceDirectory>
+                <testSourceDirectory>src/test/java</testSourceDirectory>
+                <verbose>false</verbose>
+                <filesNamePattern>.*\.java</filesNamePattern>
+                <skip>false</skip>
+                <skipSortingImports>false</skipSortingImports>
+                <style>google</style>
+            </configuration>
+            <executions>
+                <execution>
+                    <goals>
+                        <goal>format</goal>
+                    </goals>
+                </execution>
+            </executions>
+        </plugin>
+    </plugins>
+  </build>
+</project>
\ No newline at end of file
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/AbstractTaskExecutor.java b/impl/src/main/java/io/serverlessworkflow/impl/AbstractTaskExecutor.java
new file mode 100644
index 0000000..f377b3f
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/AbstractTaskExecutor.java
@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.serverlessworkflow.api.types.TaskBase;
+
+public abstract class AbstractTaskExecutor<T extends TaskBase> implements TaskExecutor<T> {
+
+  protected final T task;
+
+  protected AbstractTaskExecutor(T task) {
+    this.task = task;
+  }
+
+  @Override
+  public JsonNode apply(JsonNode node) {
+
+    // do input filtering
+    return internalExecute(node);
+    // do output filtering
+
+  }
+
+  protected abstract JsonNode internalExecute(JsonNode node);
+}
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/HttpExecutor.java b/impl/src/main/java/io/serverlessworkflow/impl/HttpExecutor.java
new file mode 100644
index 0000000..7d0f89e
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/HttpExecutor.java
@@ -0,0 +1,76 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.serverlessworkflow.api.types.CallHTTP;
+import io.serverlessworkflow.api.types.HTTPArguments;
+import io.serverlessworkflow.api.types.WithHTTPHeaders;
+import io.serverlessworkflow.api.types.WithHTTPQuery;
+import jakarta.ws.rs.client.Client;
+import jakarta.ws.rs.client.ClientBuilder;
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.client.Invocation.Builder;
+import jakarta.ws.rs.client.WebTarget;
+import jakarta.ws.rs.core.MediaType;
+import java.util.Map;
+import java.util.Map.Entry;
+
+public class HttpExecutor extends AbstractTaskExecutor<CallHTTP> {
+
+  private static final Client client = ClientBuilder.newClient();
+
+  public HttpExecutor(CallHTTP task) {
+    super(task);
+  }
+
+  @Override
+  protected JsonNode internalExecute(JsonNode node) {
+    HTTPArguments httpArgs = task.getWith();
+    // missing checks
+    String uri =
+        httpArgs
+            .getEndpoint()
+            .getEndpointConfiguration()
+            .getUri()
+            .getLiteralEndpointURI()
+            .getLiteralUriTemplate();
+    WebTarget target = client.target(uri);
+    WithHTTPQuery query = httpArgs.getQuery();
+    if (query != null) {
+      for (Entry<String, Object> entry : query.getAdditionalProperties().entrySet()) {
+        target = target.queryParam(entry.getKey(), entry.getValue());
+      }
+    }
+    Builder request =
+        target
+            .resolveTemplates(
+                JsonUtils.mapper().convertValue(node, new TypeReference<Map<String, Object>>() {}))
+            .request(MediaType.APPLICATION_JSON);
+    WithHTTPHeaders headers = httpArgs.getHeaders();
+    if (headers != null) {
+      headers.getAdditionalProperties().forEach(request::header);
+    }
+    switch (httpArgs.getMethod().toLowerCase()) {
+      case "get":
+      default:
+        return request.get(JsonNode.class);
+      case "post":
+        return request.post(Entity.json(httpArgs.getBody()), JsonNode.class);
+    }
+  }
+}
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/JsonUtils.java b/impl/src/main/java/io/serverlessworkflow/impl/JsonUtils.java
new file mode 100644
index 0000000..b00b14f
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/JsonUtils.java
@@ -0,0 +1,209 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.BigIntegerNode;
+import com.fasterxml.jackson.databind.node.BinaryNode;
+import com.fasterxml.jackson.databind.node.BooleanNode;
+import com.fasterxml.jackson.databind.node.DecimalNode;
+import com.fasterxml.jackson.databind.node.DoubleNode;
+import com.fasterxml.jackson.databind.node.FloatNode;
+import com.fasterxml.jackson.databind.node.IntNode;
+import com.fasterxml.jackson.databind.node.LongNode;
+import com.fasterxml.jackson.databind.node.NullNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.ShortNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+public class JsonUtils {
+
+  private static ObjectMapper mapper = new ObjectMapper();
+
+  public static ObjectMapper mapper() {
+    return mapper;
+  }
+
+  /*
+   * Implementation note:
+   * Although we can use directly ObjectMapper.convertValue for implementing fromValue and toJavaValue methods,
+   * the performance gain of avoiding an intermediate buffer is so tempting that we cannot avoid it
+   */
+  public static JsonNode fromValue(Object value) {
+    if (value == null) {
+      return NullNode.instance;
+    } else if (value instanceof JsonNode) {
+      return (JsonNode) value;
+    } else if (value instanceof Boolean) {
+      return BooleanNode.valueOf((Boolean) value);
+    } else if (value instanceof String) {
+      return fromString((String) value);
+    } else if (value instanceof Short) {
+      return new ShortNode((Short) value);
+    } else if (value instanceof Integer) {
+      return new IntNode((Integer) value);
+    } else if (value instanceof Long) {
+      return new LongNode((Long) value);
+    } else if (value instanceof Float) {
+      return new FloatNode((Float) value);
+    } else if (value instanceof Double) {
+      return new DoubleNode((Double) value);
+    } else if (value instanceof BigDecimal) {
+      return DecimalNode.valueOf((BigDecimal) value);
+    } else if (value instanceof BigInteger) {
+      return BigIntegerNode.valueOf((BigInteger) value);
+    } else if (value instanceof byte[]) {
+      return BinaryNode.valueOf((byte[]) value);
+    } else if (value instanceof Collection) {
+      return mapToArray((Collection<?>) value);
+    } else if (value instanceof Map) {
+      return mapToNode((Map<String, Object>) value);
+    } else {
+      return mapper.convertValue(value, JsonNode.class);
+    }
+  }
+
+  public static JsonNode fromString(String value) {
+    String trimmedValue = value.trim();
+    if (trimmedValue.startsWith("{") && trimmedValue.endsWith("}")) {
+      try {
+        return mapper.readTree(trimmedValue);
+      } catch (IOException ex) {
+        // ignore and return test node
+      }
+    }
+    return new TextNode(value);
+  }
+
+  private static Object toJavaValue(ObjectNode node) {
+    Map<String, Object> result = new HashMap<>();
+    node.fields().forEachRemaining(iter -> result.put(iter.getKey(), toJavaValue(iter.getValue())));
+    return result;
+  }
+
+  private static Collection toJavaValue(ArrayNode node) {
+    Collection result = new ArrayList<>();
+    for (JsonNode item : node) {
+      result.add(internalToJavaValue(item, JsonUtils::toJavaValue, JsonUtils::toJavaValue));
+    }
+    return result;
+  }
+
+  public static Object toJavaValue(JsonNode jsonNode) {
+    return internalToJavaValue(jsonNode, JsonUtils::toJavaValue, JsonUtils::toJavaValue);
+  }
+
+  public static <T> T convertValue(Object obj, Class<T> returnType) {
+    if (returnType.isInstance(obj)) {
+      return returnType.cast(obj);
+    } else if (obj instanceof JsonNode) {
+      return convertValue((JsonNode) obj, returnType);
+    } else {
+      return mapper.convertValue(obj, returnType);
+    }
+  }
+
+  public static <T> T convertValue(JsonNode jsonNode, Class<T> returnType) {
+    Object obj;
+    if (Boolean.class.isAssignableFrom(returnType)) {
+      obj = jsonNode.asBoolean();
+    } else if (Integer.class.isAssignableFrom(returnType)) {
+      obj = jsonNode.asInt();
+    } else if (Double.class.isAssignableFrom(returnType)) {
+      obj = jsonNode.asDouble();
+    } else if (Long.class.isAssignableFrom(returnType)) {
+      obj = jsonNode.asLong();
+    } else if (String.class.isAssignableFrom(returnType)) {
+      obj = jsonNode.asText();
+    } else {
+      obj = mapper.convertValue(jsonNode, returnType);
+    }
+    return returnType.cast(obj);
+  }
+
+  public static Object simpleToJavaValue(JsonNode jsonNode) {
+    return internalToJavaValue(jsonNode, node -> node, node -> node);
+  }
+
+  private static Object internalToJavaValue(
+      JsonNode jsonNode,
+      Function<ObjectNode, Object> objectFunction,
+      Function<ArrayNode, Object> arrayFunction) {
+    if (jsonNode.isNull()) {
+      return null;
+    } else if (jsonNode.isTextual()) {
+      return jsonNode.asText();
+    } else if (jsonNode.isBoolean()) {
+      return jsonNode.asBoolean();
+    } else if (jsonNode.isInt()) {
+      return jsonNode.asInt();
+    } else if (jsonNode.isDouble()) {
+      return jsonNode.asDouble();
+    } else if (jsonNode.isNumber()) {
+      return jsonNode.numberValue();
+    } else if (jsonNode.isArray()) {
+      return arrayFunction.apply((ArrayNode) jsonNode);
+    } else if (jsonNode.isObject()) {
+      return objectFunction.apply((ObjectNode) jsonNode);
+    } else {
+      return mapper.convertValue(jsonNode, Object.class);
+    }
+  }
+
+  public static String toString(JsonNode node) throws JsonProcessingException {
+    return mapper.writeValueAsString(node);
+  }
+
+  public static void addToNode(String name, Object value, ObjectNode dest) {
+    dest.set(name, fromValue(value));
+  }
+
+  private static ObjectNode mapToNode(Map<String, Object> value) {
+    ObjectNode objectNode = mapper.createObjectNode();
+    for (Map.Entry<String, Object> entry : value.entrySet()) {
+      addToNode(entry.getKey(), entry.getValue(), objectNode);
+    }
+    return objectNode;
+  }
+
+  private static ArrayNode mapToArray(Collection<?> collection) {
+    return mapToArray(collection, mapper.createArrayNode());
+  }
+
+  private static ArrayNode mapToArray(Collection<?> collection, ArrayNode arrayNode) {
+    for (Object item : collection) {
+      arrayNode.add(fromValue(item));
+    }
+    return arrayNode;
+  }
+
+  static ObjectNode object() {
+    return mapper.createObjectNode();
+  }
+
+  private JsonUtils() {}
+}
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/MergeUtils.java b/impl/src/main/java/io/serverlessworkflow/impl/MergeUtils.java
new file mode 100644
index 0000000..8c1ec1d
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/MergeUtils.java
@@ -0,0 +1,107 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+
+public class MergeUtils {
+  /**
+   * Merge two JSON documents.
+   *
+   * @param src JsonNode to be merged
+   * @param target JsonNode to merge to
+   */
+  public static JsonNode merge(JsonNode src, JsonNode target) {
+    return merge(src, target, false);
+  }
+
+  public static JsonNode merge(JsonNode src, JsonNode target, boolean mergeArray) {
+    if (target == null
+        || target.isNull()
+        || target.isObject() && target.isEmpty() && src != null && !src.isNull()) {
+      return src;
+    } else if (target.isArray()) {
+      return mergeArray(src, (ArrayNode) target, mergeArray);
+    } else if (target.isObject()) {
+      return mergeObject(src, (ObjectNode) target, mergeArray);
+    } else {
+      if (src.isArray()) {
+        ArrayNode srcArray = (ArrayNode) src;
+        insert(srcArray, target, getExistingNodes(srcArray));
+      } else if (src.isObject()) {
+        ((ObjectNode) src).set("_target", target);
+      }
+      return src;
+    }
+  }
+
+  private static ObjectNode mergeObject(JsonNode src, ObjectNode target, boolean mergeArray) {
+    if (src.isObject()) {
+      Iterator<Map.Entry<String, JsonNode>> mergedIterator = src.fields();
+      while (mergedIterator.hasNext()) {
+        Map.Entry<String, JsonNode> entry = mergedIterator.next();
+        JsonNode found = target.get(entry.getKey());
+        target.set(
+            entry.getKey(),
+            found != null ? merge(entry.getValue(), found, mergeArray) : entry.getValue());
+      }
+    } else if (!src.isNull()) {
+      target.set("response", src);
+    }
+    return target;
+  }
+
+  private static JsonNode mergeArray(JsonNode src, ArrayNode target, boolean mergeArray) {
+    if (src != target) {
+      if (src.isArray()) {
+        if (mergeArray) {
+          ((ArrayNode) src).forEach(node -> add(target, node, getExistingNodes(target)));
+        } else {
+          return src;
+        }
+      } else {
+        add(target, src, getExistingNodes(target));
+      }
+    }
+    return target;
+  }
+
+  private static void add(ArrayNode array, JsonNode node, Set<JsonNode> existingNodes) {
+    if (!existingNodes.contains(node)) {
+      array.add(node);
+    }
+  }
+
+  private static void insert(ArrayNode array, JsonNode node, Set<JsonNode> existingNodes) {
+    if (!existingNodes.contains(node)) {
+      array.insert(0, node);
+    }
+  }
+
+  private static Set<JsonNode> getExistingNodes(ArrayNode arrayNode) {
+    Set<JsonNode> existingNodes = new HashSet<>();
+    arrayNode.forEach(existingNodes::add);
+    return existingNodes;
+  }
+
+  private MergeUtils() {}
+}
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/TaskExecutor.java b/impl/src/main/java/io/serverlessworkflow/impl/TaskExecutor.java
new file mode 100644
index 0000000..83c4bd1
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/TaskExecutor.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.serverlessworkflow.api.types.TaskBase;
+import java.util.function.UnaryOperator;
+
+public interface TaskExecutor<T extends TaskBase> extends UnaryOperator<JsonNode> {}
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/TaskExecutorFactory.java b/impl/src/main/java/io/serverlessworkflow/impl/TaskExecutorFactory.java
new file mode 100644
index 0000000..a45f945
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/TaskExecutorFactory.java
@@ -0,0 +1,36 @@
+/*
+ * 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;
+
+import io.serverlessworkflow.api.types.CallTask;
+import io.serverlessworkflow.api.types.Task;
+import io.serverlessworkflow.api.types.TaskBase;
+
+public class TaskExecutorFactory {
+
+  private TaskExecutorFactory() {}
+
+  static TaskExecutor<? extends TaskBase> buildExecutor(Task task) {
+
+    if (task.getCallTask() != null) {
+      CallTask callTask = task.getCallTask();
+      if (callTask.getCallHTTP() != null) {
+        return new HttpExecutor(callTask.getCallHTTP());
+      }
+    }
+    throw new UnsupportedOperationException(task + " not supported yet");
+  }
+}
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java b/impl/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java
new file mode 100644
index 0000000..bb453a8
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java
@@ -0,0 +1,134 @@
+/*
+ * 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;
+
+import static io.serverlessworkflow.impl.JsonUtils.*;
+
+import com.fasterxml.jackson.core.JsonPointer;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.serverlessworkflow.api.types.TaskBase;
+import io.serverlessworkflow.api.types.TaskItem;
+import io.serverlessworkflow.api.types.Workflow;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class WorkflowDefinition {
+
+  private WorkflowDefinition(Workflow workflow, Collection<WorkflowExecutionListener> listeners) {
+    this.workflow = workflow;
+    this.listeners = listeners;
+  }
+
+  private final Workflow workflow;
+  private final Collection<WorkflowExecutionListener> listeners;
+  private final Map<JsonPointer, TaskExecutor<? extends TaskBase>> taskExecutors =
+      new ConcurrentHashMap<>();
+
+  public static class Builder {
+    private final Workflow workflow;
+    private Collection<WorkflowExecutionListener> listeners;
+
+    private Builder(Workflow workflow) {
+      this.workflow = workflow;
+    }
+
+    public Builder withListener(WorkflowExecutionListener listener) {
+      if (listeners == null) {
+        listeners = new HashSet<>();
+      }
+      listeners.add(listener);
+      return this;
+    }
+
+    public WorkflowDefinition build() {
+      return new WorkflowDefinition(
+          workflow,
+          listeners == null
+              ? Collections.emptySet()
+              : Collections.unmodifiableCollection(listeners));
+    }
+  }
+
+  public static Builder builder(Workflow workflow) {
+    return new Builder(workflow);
+  }
+
+  public WorkflowInstance execute(Object input) {
+    return new WorkflowInstance(JsonUtils.fromValue(input));
+  }
+
+  enum State {
+    STARTED,
+    WAITING,
+    FINISHED
+  };
+
+  public class WorkflowInstance {
+
+    private final JsonNode input;
+    private JsonNode output;
+    private State state;
+
+    private JsonPointer currentPos;
+
+    private WorkflowInstance(JsonNode input) {
+      this.input = input;
+      this.output = object();
+      this.state = State.STARTED;
+      this.currentPos = JsonPointer.compile("/");
+      processDo(workflow.getDo());
+    }
+
+    private void processDo(List<TaskItem> tasks) {
+      currentPos = currentPos.appendProperty("do");
+      int index = 0;
+      for (TaskItem task : tasks) {
+        currentPos = currentPos.appendIndex(index).appendProperty(task.getName());
+        listeners.forEach(l -> l.onTaskStarted(currentPos, task.getTask()));
+        this.output =
+            MergeUtils.merge(
+                taskExecutors
+                    .computeIfAbsent(
+                        currentPos, k -> TaskExecutorFactory.buildExecutor(task.getTask()))
+                    .apply(input),
+                output);
+        listeners.forEach(l -> l.onTaskEnded(currentPos, task.getTask()));
+        currentPos = currentPos.head().head();
+      }
+      currentPos = currentPos.head();
+    }
+
+    public String currentPos() {
+      return currentPos.toString();
+    }
+
+    public State state() {
+      return state;
+    }
+
+    public Object output() {
+      return toJavaValue(output);
+    }
+
+    public Object outputAsJsonNode() {
+      return output;
+    }
+  }
+}
diff --git a/impl/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java b/impl/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java
new file mode 100644
index 0000000..700c6aa
--- /dev/null
+++ b/impl/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java
@@ -0,0 +1,26 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonPointer;
+import io.serverlessworkflow.api.types.Task;
+
+public interface WorkflowExecutionListener {
+
+  void onTaskStarted(JsonPointer currentPos, Task task);
+
+  void onTaskEnded(JsonPointer currentPos, Task task);
+}
diff --git a/impl/src/main/resources/callHttp.yaml b/impl/src/main/resources/callHttp.yaml
new file mode 100644
index 0000000..4022e38
--- /dev/null
+++ b/impl/src/main/resources/callHttp.yaml
@@ -0,0 +1,13 @@
+document:
+  dsl: 1.0.0-alpha1
+  namespace: default
+  name: http-call-with-response-output
+  version: 1.0.0
+do:
+  - getPet:
+      call: http
+      with:
+        method: get
+        endpoint:
+          uri: https://petstore.swagger.io/v2/pet/{petId}
+        output: response
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 78378df..8f904f7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -39,6 +39,7 @@
   <modules>
     <module>api</module>
     <module>custom-generator</module>
+    <module>impl</module>
   </modules>
 
   <properties>


Test patch:
diff --git a/api/src/test/java/io/serverlessworkflow/api/FeaturesTest.java b/api/src/test/java/io/serverlessworkflow/api/FeaturesTest.java
index af11a49..fd16b95 100644
--- a/api/src/test/java/io/serverlessworkflow/api/FeaturesTest.java
+++ b/api/src/test/java/io/serverlessworkflow/api/FeaturesTest.java
@@ -44,7 +44,8 @@ public class FeaturesTest {
         "features/switch.yaml",
         "features/try.yaml",
         "features/listen.yaml",
-        "features/callFunction.yaml"
+        "features/callFunction.yaml",
+        "features/callCustomFunction.yaml"
       })
   public void testSpecFeaturesParsing(String workflowLocation) throws IOException {
     Workflow workflow = readWorkflowFromClasspath(workflowLocation);
diff --git a/api/src/test/resources/features/callCustomFunction.yaml b/api/src/test/resources/features/callCustomFunction.yaml
new file mode 100644
index 0000000..4161cf4
--- /dev/null
+++ b/api/src/test/resources/features/callCustomFunction.yaml
@@ -0,0 +1,27 @@
+document:
+  dsl: 1.0.0-alpha5
+  namespace: test
+  name: call-example
+  version: 0.1.0
+schedule:
+  cron: 0 8 * * *
+do:
+- getData:
+    call: http
+    with:
+      method: get
+      endpoint: https://api.agify.io?name=meelad
+  output:
+    as: ".data.reading"
+- filterData:
+    for:
+      in: ".data.reading"
+      each: reading
+    do:
+    - log:
+        call: https://raw.githubusercontent.com/serverlessworkflow/catalog/main/functions/log/1.0.0/function.yaml
+        with:
+          level: information
+          format: "{TIMESTAMP} [{LEVEL}] ({CONTEXT}): {MESSAGE}"
+          message: Hello, world!
+          timestamp: true
\ No newline at end of file
diff --git a/impl/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java b/impl/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java
new file mode 100644
index 0000000..91b1a25
--- /dev/null
+++ b/impl/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java
@@ -0,0 +1,51 @@
+package io.serverlessworkflow.impl;
+
+/*
+ * 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.
+ */
+import static io.serverlessworkflow.api.WorkflowReader.readWorkflowFromClasspath;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.stream.Stream;
+import org.assertj.core.api.Condition;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class WorkflowDefinitionTest {
+
+  @ParameterizedTest
+  @MethodSource("provideParameters")
+  void testWorkflowExecution(String fileName, Object input, Condition<Object> condition)
+      throws IOException {
+    assertThat(
+            WorkflowDefinition.builder(readWorkflowFromClasspath(fileName))
+                .build()
+                .execute(input)
+                .output())
+        .is(condition);
+  }
+
+  private static Stream<Arguments> provideParameters() {
+    return Stream.of(
+        Arguments.of(
+            "callHttp.yaml",
+            Map.of("petId", 10),
+            new Condition<>(
+                o -> ((Map<String, Object>) o).containsKey("photoUrls"), "callHttpCondition")));
+  }
+}