diff --git a/README.md b/README.md index c0b4df70..b641b483 100644 --- a/README.md +++ b/README.md @@ -8,11 +8,11 @@ Provides the Java API for the [Serverless Workflow Specification](https://github With the SDK you can: * Read workflow JSON and YAML definitions -* Write workflow in JSON and YAML format. +* Write workflow definitions in JSON and YAML formats. +* Test your workflow definitions using the reference implementation. -Serverless Workflow Java SDK is **not** a workflow runtime implementation but can be used by Java runtime implementations to parse workflow definitions. -### Status +## Status | Latest Releases | Conformance to spec version | | :---: | :---: | @@ -25,17 +25,18 @@ Serverless Workflow Java SDK is **not** a workflow runtime implementation but ca Note that 6.0.0.Final, which will be the one for specification version 0.9, is skipped intentionally in case someone want to work on it. -### JDK Version +## JDK Version | SDK Version | JDK Version | | :---: | :---: | +| 7.0.0 and after | 17 | | 5.0.0 and after | 11 | | 4.0.x and before | 8 | -### Getting Started +## Getting Started -#### Building SNAPSHOT locally +### Building SNAPSHOT locally To build project and run tests locally: @@ -47,7 +48,7 @@ mvn clean install The project uses [Google's code styleguide](https://google.github.io/styleguide/javaguide.html). Your changes should be automatically formatted during the build. -#### Maven projects: +### Maven projects: Add the following dependencies to your pom.xml `dependencies` section: @@ -59,7 +60,7 @@ Add the following dependencies to your pom.xml `dependencies` section: ``` -#### Gradle projects: +### Gradle projects: Add the following dependencies to your build.gradle `dependencies` section: @@ -67,11 +68,20 @@ Add the following dependencies to your pom.xml `dependencies` section: implementation("io.serverlessworkflow:serverlessworkflow-api:7.0.0-SNAPSHOT") ``` -### How to Use +## How to Use -#### Creating from JSON/YAML source +There are, roughly speaking, two kind of users of this SDK: + * Those ones interested on implementing their own runtime using Java. + * Those ones interested on using the provided runtime reference implementation. -You can create a Workflow instance from JSON/YAML source: +### Implementing your own runtime + +For those ones interested on implementing their own runtime, this SDK provides an easy way to load an in memory representation of a given workflow definition. +This in-memory representation consists of a hierarchy of POJOS directly generated from the Serverless Workflow specification [schema](api/src/main/resources/schema/workflow.yaml), which ensures the internal representation is aligned with the specification schema. The root of the hierarchy is `io.serverlessworkflow.api.types.Workflow` class + +### Reading workflow definition from JSON/YAML source + +You can read a Workflow definition from JSON/YAML source: Let's say you have a simple YAML based workflow definition in a file name `simple.yaml` located in your working dir: @@ -93,7 +103,7 @@ do: ``` -To parse it and create a Workflow instance you can do: +To parse it and get a Workflow instance you can do: ``` java @@ -102,10 +112,20 @@ try (InputStream in = new FileInputStream("simple.yaml")) { // Once you have the Workflow instance you can use its API to inspect it } ``` +By default, Workflows are not validated against the schema (performance being the priority). If you want to enable validation, you can do that by using: + +``` java +try (InputStream in = new FileInputStream("simple.yaml")) { + Workflow workflow = WorkflowReader.validation().readWorkflow (in, WorkflowFormat.YAML); + // Once you have the Workflow instance you can use its API to inspect it +} +``` -#### Writing a workflow +For additional reading helper methods, including the one to read a workflow definition from classpath, check [WorkflowReader](api/src/main/java/io/serverlessworkflow/api/WorkflowReader.java) class. -Given a workflow definition, you can store it using JSON or YAML format. +### Writing workflow definition to a JSON/YAML target + +Given a Workflow instance, you can store it using JSON or YAML format. For example, to store a workflow using json format in a file called `simple.json`, you write ``` java @@ -113,4 +133,10 @@ try (OutputStream out = new FileOutputStream("simple.json")) { WorkflowWriter.writeWorkflow(out, workflow, WorkflowFormat.JSON); } -``` \ No newline at end of file +``` +For additional writing helper methods, check [WorkflowWriter](api/src/main/java/io/serverlessworkflow/api/WorkflowWriter.java) class. + +### Reference implementation + +The reference implementation provides a ready-to-use runtime that supports the Serverless Workflow Specification. It includes a workflow execution engine, validation utilities, and illustrative examples to help you quickly test and deploy your workflows. For details on usage, configuration, and supported features, see [readme](impl/README.md). + diff --git a/examples/events/pom.xml b/examples/events/pom.xml new file mode 100644 index 00000000..245459ed --- /dev/null +++ b/examples/events/pom.xml @@ -0,0 +1,19 @@ + + 4.0.0 + + io.serverlessworkflow + serverlessworkflow-examples + 7.0.0-SNAPSHOT + + events + + + io.serverlessworkflow + serverlessworkflow-impl-core + + + org.slf4j + slf4j-simple + + + \ No newline at end of file diff --git a/examples/events/src/main/java/events/EventExample.java b/examples/events/src/main/java/events/EventExample.java new file mode 100644 index 00000000..628782fb --- /dev/null +++ b/examples/events/src/main/java/events/EventExample.java @@ -0,0 +1,50 @@ +/* + * 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 events; + +import io.serverlessworkflow.api.WorkflowReader; +import io.serverlessworkflow.impl.WorkflowApplication; +import io.serverlessworkflow.impl.WorkflowDefinition; +import io.serverlessworkflow.impl.WorkflowInstance; +import java.io.IOException; +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class EventExample { + + private static final Logger logger = LoggerFactory.getLogger(EventExample.class); + + public static void main(String[] args) throws IOException { + try (WorkflowApplication appl = WorkflowApplication.builder().build()) { + WorkflowDefinition listenDefinition = + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("listen.yaml")); + WorkflowDefinition emitDefinition = + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("emit.yaml")); + WorkflowInstance waitingInstance = listenDefinition.instance(Map.of()); + waitingInstance + .start() + .thenAccept(node -> logger.info("Waiting instance completed with result {}", node)); + logger.info("Listen instance waiting for proper event, Status {}", waitingInstance.status()); + logger.info("Publishing event with temperature 35"); + emitDefinition.instance(Map.of("temperature", 35)).start().join(); + logger.info( + "Listen instance still waiting for proper event, Status {}", waitingInstance.status()); + logger.info("Publishing event with temperature 39"); + emitDefinition.instance(Map.of("temperature", 39)).start().join(); + } + } +} diff --git a/examples/events/src/main/resources/emit.yaml b/examples/events/src/main/resources/emit.yaml new file mode 100644 index 00000000..4d14b030 --- /dev/null +++ b/examples/events/src/main/resources/emit.yaml @@ -0,0 +1,14 @@ +document: + dsl: '1.0.0-alpha5' + namespace: test + name: emit + version: '0.1.0' +do: + - emitEvent: + emit: + event: + with: + source: https://hospital.com + type: com.fake-hospital.vitals.measurements.temperature + data: + temperature: ${.temperature} \ No newline at end of file diff --git a/examples/events/src/main/resources/listen.yaml b/examples/events/src/main/resources/listen.yaml new file mode 100644 index 00000000..e49cea92 --- /dev/null +++ b/examples/events/src/main/resources/listen.yaml @@ -0,0 +1,13 @@ +document: + dsl: '1.0.0-alpha5' + namespace: examples + name: listen + version: '0.1.0' +do: + - callDoctor: + listen: + to: + one: + with: + type: com.fake-hospital.vitals.measurements.temperature + data: ${ .temperature > 38 } \ No newline at end of file diff --git a/examples/pom.xml b/examples/pom.xml new file mode 100644 index 00000000..54c88571 --- /dev/null +++ b/examples/pom.xml @@ -0,0 +1,33 @@ + + 4.0.0 + + io.serverlessworkflow + serverlessworkflow-parent + 7.0.0-SNAPSHOT + + serverlessworkflow-examples + pom + + + + io.serverlessworkflow + serverlessworkflow-impl-core + ${project.version} + + + io.serverlessworkflow + serverlessworkflow-impl-http + ${project.version} + + + org.slf4j + slf4j-simple + 2.0.16 + + + + + simpleGet + events + + \ No newline at end of file diff --git a/impl/bom/pom.xml b/examples/simpleGet/pom.xml similarity index 76% rename from impl/bom/pom.xml rename to examples/simpleGet/pom.xml index 63ef0fe3..34ad62c7 100644 --- a/impl/bom/pom.xml +++ b/examples/simpleGet/pom.xml @@ -2,11 +2,10 @@ 4.0.0 io.serverlessworkflow - serverlessworkflow-impl + serverlessworkflow-examples 7.0.0-SNAPSHOT - serverlessworkflow-impl-bom - pom + simpleGet io.serverlessworkflow @@ -16,5 +15,9 @@ io.serverlessworkflow serverlessworkflow-impl-http + + org.slf4j + slf4j-simple + \ No newline at end of file diff --git a/examples/simpleGet/src/main/java/io/serverlessworkflow/impl/BlockingExample.java b/examples/simpleGet/src/main/java/io/serverlessworkflow/impl/BlockingExample.java new file mode 100644 index 00000000..233d121f --- /dev/null +++ b/examples/simpleGet/src/main/java/io/serverlessworkflow/impl/BlockingExample.java @@ -0,0 +1,38 @@ +/* + * 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.WorkflowReader; +import java.io.IOException; +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class BlockingExample { + + private static final Logger logger = LoggerFactory.getLogger(BlockingExample.class); + + public static void main(String[] args) throws IOException { + try (WorkflowApplication appl = WorkflowApplication.builder().build()) { + logger.info( + "Workflow output is {}", + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("get.yaml")) + .instance(Map.of("petId", 10)) + .start() + .join()); + } + } +} diff --git a/examples/simpleGet/src/main/java/io/serverlessworkflow/impl/NotBlockingExample.java b/examples/simpleGet/src/main/java/io/serverlessworkflow/impl/NotBlockingExample.java new file mode 100644 index 00000000..cb663c1a --- /dev/null +++ b/examples/simpleGet/src/main/java/io/serverlessworkflow/impl/NotBlockingExample.java @@ -0,0 +1,37 @@ +/* + * 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.WorkflowReader; +import java.io.IOException; +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class NotBlockingExample { + + private static final Logger logger = LoggerFactory.getLogger(NotBlockingExample.class); + + public static void main(String[] args) throws IOException { + try (WorkflowApplication appl = WorkflowApplication.builder().build()) { + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("get.yaml")) + .instance(Map.of("petId", 10)) + .start() + .thenAccept(node -> logger.info("Workflow output is {}", node)); + logger.info("The request has been sent, this thread might continue doing stuff"); + } + } +} diff --git a/examples/simpleGet/src/main/resources/get.yaml b/examples/simpleGet/src/main/resources/get.yaml new file mode 100644 index 00000000..7adf3132 --- /dev/null +++ b/examples/simpleGet/src/main/resources/get.yaml @@ -0,0 +1,11 @@ +document: + dsl: '1.0.0-alpha5' + namespace: examples + name: call-http-shorthand-endpoint + version: '0.1.0' +do: + - getPet: + call: http + with: + method: get + endpoint: https://petstore.swagger.io/v2/pet/{petId} diff --git a/impl/README.md b/impl/README.md new file mode 100644 index 00000000..f581a398 --- /dev/null +++ b/impl/README.md @@ -0,0 +1,194 @@ +![Verify JAVA SDK](https://github.com/serverlessworkflow/sdk-java/workflows/Verify%20JAVA%20SDK/badge.svg) +![Deploy JAVA SDK](https://github.com/serverlessworkflow/sdk-java/workflows/Deploy%20JAVA%20SDK/badge.svg) [![Gitpod ready-to-code](https://img.shields.io/badge/Gitpod-ready--to--code-blue?logo=gitpod)](https://gitpod.io/#https://github.com/serverlessworkflow/sdk-java) + +# Serverless Workflow Specification - Java SDK- Reference Implementation + +Welcome to Java SDK runtime reference implementation, a lightweight implementation of the Serverless Workflow specification which provides a simple, non blocking, reactive API for workflow execution. + +Although initially conceived mainly for testing purposes, it was designed to be easily expanded, so it can eventually become production ready. + +## Status + +This reference implementation is currently capable of running workflows consisting of: + + +* Tasks + * Switch + * Set + * Do + * Raise + * Listen + * Emit + * Fork + * For + * Try + * Wait + * Call + * HTTP +* Schema Validation + * Input + * Output +* Expressions + * Input + * Output + * Export + * Special keywords: runtime, workflow, task... +* Error definitions + + +## Setup + +Before getting started, ensure you have Java 17+ and Maven or Gradle installed. + +Install [Java 17](https://openjdk.org/projects/jdk/17/) +Install [Maven](https://maven.apache.org/install.html) (if using Maven) +Install [Gradle](https://gradle.org/install) (if using Gradle) + +### Dependencies + +This implementation follows a modular approach, keeping dependencies minimal: +- The core library is always required. +- Additional dependencies must be explicitly included if your workflow interacts with external services (e.g., HTTP). +This ensures you only include what you need, preventing unnecessary dependencies. + +#### Maven + +You always need to add this dependency to your pom.xml `dependencies` section: + +```xml + + io.serverlessworkflow + serverlessworkflow-impl-core + 7.0.0 + +``` + +And only if your workflow is using HTTP calls, you must add: + +```xml + + io.serverlessworkflow + serverlessworkflow-impl-http + 7.0.0 + +``` + +#### Gradle projects: + +You always need to add this dependency to your build.gradle `dependencies` section: + +```text +implementation("io.serverlessworkflow:serverlessworkflow-impl-core:7.0.0") +``` + +And only if your workflow is using HTTP calls, you must add: + +```text +implementation("io.serverlessworkflow:serverlessworkflow-impl-http:7.0.0") +``` + +## How to use + +The quick version is intended for impatient users who want to try something as soon as possible. + +The detailed version is more suitable for those users interested in a more thoughtful discussion of the API. + +### Quick version + +For a quick introduction, we will use a simple workflow [definition](../examples/simpleGet/src/main/resources/get.yaml) that performs a get call. +We are going to show two ways of invoking the workflow: + - blocking the thread till the get request goes through + - returning control to the caller, so the main thread continues while the get is executed + +In order to execute the workflow, blocking the thread till the HTTP request is completed, you should write + +``` java +try (WorkflowApplication appl = WorkflowApplication.builder().build()) { + logger.info( + "Workflow output is {}", + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("get.yaml")) + .instance(Map.of("petId", 10)) + .start() + .join()); + } +``` +You can find the complete java code [here](../examples/simpleGet/src/main/java/BlockingExample.java) + +In order to execute the workflow without blocking the calling thread till the HTTP request is completed, you should write + +``` java + try (WorkflowApplication appl = WorkflowApplication.builder().build()) { + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("get.yaml")) + .instance(Map.of("petId", 10)) + .start() + .thenAccept(node -> logger.info("Workflow output is {}", node)); + } +``` +When the HTTP request is done, both examples will print a similar output + + +```shell +Workflow output is {"id":10,"category":{"id":10,"name":"string"},"name":"doggie","photoUrls":["string"],"tags":[{"id":10,"name":"string"}],"status":"string"} +``` + +You can find the complete java code [here](../examples/simpleGet/src/main/java/NotBlockingExample.java) + +### Detailed version + +To discuss runtime API we are going to use a couple of workflow: +- [listen.yaml](../examples/events/src/main/listen.yaml), which waits for an event reporting a temperature greater than 38 +- [emit.yaml](../examples/events/src/main/emit.yaml), which emits events with a certain temperature, specified as workflow parameter. + +Here is a summary of what we are trying to do: + +- The listen.yaml workflow waits for an event (not-blocking). +- We send an event with a low temperature (ignored). +- We send an event with a high temperature (completes the workflow). + +The first step is to create a [WorkflowApplication](core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java) instance. An application is an abstraction that allows customization of different aspects of the workflow execution (for example, change the default `ExecutorService` for thread spawning) + +Since `WorkflowApplication` implements `Autocloseable`, we better use a **try-with-resources** block, ensuring any resource that the workflow might have used is freed when done. + +`try (WorkflowApplication appl = WorkflowApplication.builder().build())` + +Once we have the application object, we use it to parse our definition examples. To load each workflow definition, we use the `readFromClasspath` helper method defined in [WorkflowReader](api/src/main/java/io/serverlessworkflow/api/WorkflowReader.java) class. + +```java + WorkflowDefinition listenDefinition = + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("listen.yaml")); + WorkflowDefinition emitDefinition = + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath("emit.yaml")); +``` + +A [WorkflowDefinition](core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java) object is immutable and, therefore, thread-safe. It is used to execute as many workflow instances as desired. + +To execute a workflow, we first create a [WorkflowInstance](core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java) object (its initial status is PENDING) and then invoke the `start` method on it (its status is changed to RUNNING). The `start` method returns a [CompletableFuture](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html), which we use to indicate that a log message should be printed when the workflow is completed. + +```java + WorkflowInstance waitingInstance = listenDefinition.instance(Map.of()); + waitingInstance + .start() + .thenAccept(node -> logger.info("Waiting instance completed with result {}", node)); +``` + +As soon as the workflow execution reach the point where it waits for events to arrive, control is returned to the calling thread. Since the execution is not blocking, we can execute another workflow instance while the first one is waiting. + +We will send an event with a temperature that does not satisfy the criteria, so the listen instance will continue waiting. We use a regular Java `Map` to pass parameters to the workflow instance that sends the event. Note that since we want to wait till the event is published, we call `join` after `start`, telling the `CompletableFuture` to wait for workflow completion. + +```java + emitDefinition.instance(Map.of("temperature", 35)).start().join(); + ``` + + It's time to complete the waiting instance and send an event with the expected temperature. We do so by reusing `emitDefinition`. + +```java + emitDefinition.instance(Map.of("temperature", 39)).start().join(); + ``` + +After that, listen instance will be completed and we will see this log message + +```java +[pool-1-thread-1] INFO events.EventExample - Waiting instance completed with result [{"temperature":39}] +``` +The source code of the example is [here](../examples/events/src/main/java/EventExample.java) + diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.java new file mode 100644 index 00000000..1ac1f759 --- /dev/null +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.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 java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +public class DefaultExecutorServiceFactory implements ExecutorServiceFactory { + + private static final ExecutorServiceFactory instance = new DefaultExecutorServiceFactory(); + + public static ExecutorServiceFactory instance() { + return instance; + } + + private static class ExecutorServiceHolder { + private static ExecutorService instance = Executors.newCachedThreadPool(); + } + + @Override + public ExecutorService get() { + return ExecutorServiceHolder.instance; + } + + private DefaultExecutorServiceFactory() {} +} diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java index 23597057..b998c57d 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java @@ -53,8 +53,6 @@ public class WorkflowApplication implements AutoCloseable { private final EventConsumer eventConsumer; private final EventPublisher eventPublisher; - private ExecutorService executorService; - private WorkflowApplication(Builder builder) { this.taskFactory = builder.taskFactory; this.exprFactory = builder.exprFactory; @@ -193,7 +191,7 @@ public WorkflowDefinition workflowDefinition(Workflow workflow) { } @Override - public void close() throws Exception { + public void close() { for (WorkflowDefinition definition : definitions.values()) { definition.close(); } @@ -214,11 +212,6 @@ public EventConsumer eventConsumer() { } public ExecutorService executorService() { - synchronized (executorFactory) { - if (executorService == null) { - executorService = executorFactory.get(); - } - } - return executorService; + return executorFactory.get(); } } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java index 6566cf6b..1a789616 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java @@ -77,15 +77,15 @@ public WorkflowInstance instance(Object input) { return new WorkflowInstance(this, JsonUtils.fromValue(input)); } - public Optional inputSchemaValidator() { + Optional inputSchemaValidator() { return inputSchemaValidator; } - public TaskExecutor startTask() { + TaskExecutor startTask() { return taskExecutor; } - public Optional inputFilter() { + Optional inputFilter() { return inputFilter; } @@ -97,11 +97,11 @@ public Collection listeners() { return application.listeners(); } - public Optional outputFilter() { + Optional outputFilter() { return outputFilter; } - public Optional outputSchemaValidator() { + Optional outputSchemaValidator() { return outputSchemaValidator; } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java index 3993f8fe..714d89d0 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java @@ -16,6 +16,7 @@ package io.serverlessworkflow.impl.events; import io.cloudevents.CloudEvent; +import io.serverlessworkflow.impl.DefaultExecutorServiceFactory; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -62,7 +63,8 @@ public CompletableFuture publish(CloudEvent ce) { if (consumer != null) { consumer.accept(ce); } - }); + }, + DefaultExecutorServiceFactory.instance().get()); } @Override diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java index 1b53dea3..58472289 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ListenExecutor.java @@ -35,6 +35,7 @@ import io.serverlessworkflow.impl.WorkflowContext; import io.serverlessworkflow.impl.WorkflowFilter; import io.serverlessworkflow.impl.WorkflowPosition; +import io.serverlessworkflow.impl.WorkflowStatus; import io.serverlessworkflow.impl.WorkflowUtils; import io.serverlessworkflow.impl.events.CloudEventUtils; import io.serverlessworkflow.impl.events.EventConsumer; @@ -235,6 +236,7 @@ protected CompletableFuture internalExecute( WorkflowContext workflow, TaskContext taskContext) { ArrayNode output = JsonUtils.mapper().createArrayNode(); Collection registrations = new ArrayList<>(); + workflow.instance().status(WorkflowStatus.WAITING); return buildFuture( regBuilders, registrations, @@ -243,6 +245,7 @@ protected CompletableFuture internalExecute( processCe(converter.apply(ce), output, workflow, taskContext, future))) .thenApply( v -> { + workflow.instance().status(WorkflowStatus.RUNNING); registrations.forEach(reg -> eventConsumer.unregister(reg)); return output; }); diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutorHelper.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutorHelper.java index ba6c33b2..3fa77f5f 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutorHelper.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutorHelper.java @@ -54,7 +54,7 @@ public static boolean isActive(WorkflowContext context) { } public static boolean isActive(WorkflowStatus status) { - return status == WorkflowStatus.RUNNING; + return status == WorkflowStatus.RUNNING || status == WorkflowStatus.WAITING; } public static TaskExecutor createExecutorList( diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/WaitExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/WaitExecutor.java index 2f1ea1b6..42e648aa 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/WaitExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/WaitExecutor.java @@ -23,6 +23,7 @@ import io.serverlessworkflow.impl.WorkflowApplication; import io.serverlessworkflow.impl.WorkflowContext; import io.serverlessworkflow.impl.WorkflowPosition; +import io.serverlessworkflow.impl.WorkflowStatus; import io.serverlessworkflow.impl.executors.RegularTaskExecutor.RegularTaskExecutorBuilder; import io.serverlessworkflow.impl.resources.ResourceLoader; import java.time.Duration; @@ -72,7 +73,13 @@ protected WaitExecutor(WaitExecutorBuilder builder) { @Override protected CompletableFuture internalExecute( WorkflowContext workflow, TaskContext taskContext) { + workflow.instance().status(WorkflowStatus.WAITING); return new CompletableFuture() - .completeOnTimeout(taskContext.output(), millisToWait.toMillis(), TimeUnit.MILLISECONDS); + .completeOnTimeout(taskContext.output(), millisToWait.toMillis(), TimeUnit.MILLISECONDS) + .thenApply( + node -> { + workflow.instance().status(WorkflowStatus.RUNNING); + return node; + }); } } diff --git a/impl/core/src/test/java/io/serverlessworkflow/impl/EventDefinitionTest.java b/impl/core/src/test/java/io/serverlessworkflow/impl/EventDefinitionTest.java index 20f78b65..981b149d 100644 --- a/impl/core/src/test/java/io/serverlessworkflow/impl/EventDefinitionTest.java +++ b/impl/core/src/test/java/io/serverlessworkflow/impl/EventDefinitionTest.java @@ -51,7 +51,7 @@ void testEventListened(String listen, String emit, JsonNode expectedResult, Obje appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(emit)); WorkflowInstance waitingInstance = listenDefinition.instance(Map.of()); CompletableFuture future = waitingInstance.start(); - assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.RUNNING); + assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.WAITING); emitDefinition.instance(emitInput).start().join(); assertThat(future).isCompleted(); assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.COMPLETED); @@ -70,11 +70,11 @@ void testEventsListened(String listen, String emit1, String emit2, JsonNode expe appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(emit2)); WorkflowInstance waitingInstance = listenDefinition.instance(Map.of()); CompletableFuture future = waitingInstance.start(); - assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.RUNNING); + assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.WAITING); emitDoctorDefinition.instance(Map.of("temperature", 35)).start().join(); - assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.RUNNING); + assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.WAITING); emitDoctorDefinition.instance(Map.of("temperature", 39)).start().join(); - assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.RUNNING); + assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.WAITING); emitOutDefinition.instance(Map.of()).start().join(); assertThat(future).isCompleted(); assertThat(waitingInstance.status()).isEqualTo(WorkflowStatus.COMPLETED); diff --git a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java index 79a57156..3c078309 100644 --- a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java +++ b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java @@ -107,14 +107,18 @@ public CompletableFuture apply( Builder request = target.request(); ExpressionUtils.evaluateExpressionMap(headersMap, workflow, taskContext, input) .forEach(request::header); - try { - return CompletableFuture.completedFuture( - requestFunction.apply(request, workflow, taskContext, input)); - } catch (WebApplicationException exception) { - throw new WorkflowException( - WorkflowError.communication(exception.getResponse().getStatus(), taskContext, exception) - .build()); - } + return CompletableFuture.supplyAsync( + () -> { + try { + return requestFunction.apply(request, workflow, taskContext, input); + } catch (WebApplicationException exception) { + throw new WorkflowException( + WorkflowError.communication( + exception.getResponse().getStatus(), taskContext, exception) + .build()); + } + }, + workflow.definition().application().executorService()); } @Override diff --git a/impl/pom.xml b/impl/pom.xml index 0f0a224d..d2efb4f7 100644 --- a/impl/pom.xml +++ b/impl/pom.xml @@ -60,6 +60,5 @@ http core - bom \ No newline at end of file diff --git a/pom.xml b/pom.xml index 28314d62..27ef242f 100644 --- a/pom.xml +++ b/pom.xml @@ -40,6 +40,7 @@ api custom-generator impl + examples