From 976729f1bb8b73e2114af268b7a87a25ee48c80b Mon Sep 17 00:00:00 2001 From: Francisco Javier Tirado Sarti Date: Mon, 2 Dec 2024 21:24:37 +0100 Subject: [PATCH] [Fix #484] Execute Fork task --- .../impl/ExecutorServiceFactory.java | 22 ++++ .../impl/QueueWorkflowPosition.java | 4 +- .../serverlessworkflow/impl/TaskContext.java | 71 +++++++++--- .../impl/WorkflowApplication.java | 23 ++++ .../impl/WorkflowDefinition.java | 5 + .../impl/WorkflowExecutionListener.java | 6 +- .../impl/WorkflowInstance.java | 22 ++-- .../impl/WorkflowUtils.java | 43 +++---- .../impl/executors/AbstractTaskExecutor.java | 15 +++ .../executors/DefaultTaskExecutorFactory.java | 2 + .../impl/executors/ForkExecutor.java | 105 ++++++++++++++++++ .../impl/WorkflowDefinitionTest.java | 31 +++++- .../src/test/resources/fork-no-compete.yaml | 18 +++ impl/core/src/test/resources/fork.yaml | 18 +++ 14 files changed, 333 insertions(+), 52 deletions(-) create mode 100644 impl/core/src/main/java/io/serverlessworkflow/impl/ExecutorServiceFactory.java create mode 100644 impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForkExecutor.java create mode 100644 impl/core/src/test/resources/fork-no-compete.yaml create mode 100644 impl/core/src/test/resources/fork.yaml diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/ExecutorServiceFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/ExecutorServiceFactory.java new file mode 100644 index 00000000..7c211149 --- /dev/null +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/ExecutorServiceFactory.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 java.util.concurrent.ExecutorService; +import java.util.function.Supplier; + +@FunctionalInterface +public interface ExecutorServiceFactory extends Supplier {} diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/QueueWorkflowPosition.java b/impl/core/src/main/java/io/serverlessworkflow/impl/QueueWorkflowPosition.java index c6d3f141..5ad4934f 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/QueueWorkflowPosition.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/QueueWorkflowPosition.java @@ -54,7 +54,7 @@ public String jsonPointer() { @Override public String toString() { - return "ListWorkflowPosition [list=" + queue + "]"; + return "QueueWorkflowPosition [queue=" + queue + "]"; } @Override @@ -65,6 +65,6 @@ public WorkflowPosition back() { @Override public Object last() { - return queue.pollLast(); + return queue.getLast(); } } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java b/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java index cadde89c..8015c687 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java @@ -28,35 +28,64 @@ public class TaskContext { private final JsonNode rawInput; private final T task; private final WorkflowPosition position; - private final Instant startedAt = Instant.now(); + private final Instant startedAt; private JsonNode input; private JsonNode output; private JsonNode rawOutput; private FlowDirective flowDirective; private Map contextVariables; + private Instant completedAt; public TaskContext(JsonNode input, WorkflowPosition position) { - this.rawInput = input; - this.position = position; - this.task = null; - this.contextVariables = new HashMap<>(); - init(); + this(input, null, position, Instant.now(), input, input, input, null, new HashMap<>()); } - public TaskContext(JsonNode input, TaskContext taskContext, T task) { - this.rawInput = input; - this.position = taskContext.position.copy(); - this.task = task; - this.flowDirective = task.getThen(); - this.contextVariables = new HashMap<>(taskContext.variables()); - init(); + public TaskContext copy() { + return new TaskContext( + rawInput, + task, + position.copy(), + startedAt, + input, + output, + rawOutput, + flowDirective, + new HashMap<>(contextVariables)); } - private void init() { - this.input = rawInput; - this.rawOutput = rawInput; - this.output = rawInput; + public TaskContext(JsonNode input, TaskContext taskContext, T task) { + this( + input, + task, + taskContext.position, + Instant.now(), + input, + input, + input, + task.getThen(), + new HashMap<>(taskContext.variables())); + } + + private TaskContext( + JsonNode rawInput, + T task, + WorkflowPosition position, + Instant startedAt, + JsonNode input, + JsonNode output, + JsonNode rawOutput, + FlowDirective flowDirective, + Map contextVariables) { + this.rawInput = rawInput; + this.task = task; + this.position = position; + this.startedAt = startedAt; + this.input = input; + this.output = output; + this.rawOutput = rawOutput; + this.flowDirective = flowDirective; + this.contextVariables = contextVariables; } public void input(JsonNode input) { @@ -115,4 +144,12 @@ public WorkflowPosition position() { public Instant startedAt() { return startedAt; } + + public void completedAt(Instant instant) { + this.completedAt = instant; + } + + public Instant completedAt() { + return completedAt; + } } 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 4f35de41..f36c23f6 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java @@ -32,6 +32,8 @@ import java.util.HashSet; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; public class WorkflowApplication implements AutoCloseable { @@ -43,8 +45,11 @@ public class WorkflowApplication implements AutoCloseable { private final Collection listeners; private final Map definitions; private final WorkflowPositionFactory positionFactory; + private final ExecutorServiceFactory executorFactory; private final RuntimeDescriptorFactory runtimeDescriptorFactory; + private ExecutorService executorService; + public WorkflowApplication( TaskExecutorFactory taskFactory, ExpressionFactory exprFactory, @@ -53,6 +58,7 @@ public WorkflowApplication( WorkflowPositionFactory positionFactory, WorkflowIdFactory idFactory, RuntimeDescriptorFactory runtimeDescriptorFactory, + ExecutorServiceFactory executorFactory, Collection listeners) { this.taskFactory = taskFactory; this.exprFactory = exprFactory; @@ -61,6 +67,7 @@ public WorkflowApplication( this.positionFactory = positionFactory; this.idFactory = idFactory; this.runtimeDescriptorFactory = runtimeDescriptorFactory; + this.executorFactory = executorFactory; this.listeners = listeners; this.definitions = new ConcurrentHashMap<>(); } @@ -101,6 +108,7 @@ public static class Builder { private SchemaValidatorFactory schemaValidatorFactory = DefaultSchemaValidatorFactory.get(); private WorkflowPositionFactory positionFactory = () -> new QueueWorkflowPosition(); private WorkflowIdFactory idFactory = () -> UlidCreator.getMonotonicUlid().toString(); + private ExecutorServiceFactory executorFactory = () -> Executors.newCachedThreadPool(); private RuntimeDescriptorFactory descriptorFactory = () -> new RuntimeDescriptor("reference impl", "1.0.0_alpha", Collections.emptyMap()); @@ -129,6 +137,11 @@ public Builder withResourceLoaderFactory(ResourceLoaderFactory resourceLoader) { return this; } + public Builder withExecutorFactory(ExecutorServiceFactory executorFactory) { + this.executorFactory = executorFactory; + return this; + } + public Builder withPositionFactory(WorkflowPositionFactory positionFactory) { this.positionFactory = positionFactory; return this; @@ -158,6 +171,7 @@ public WorkflowApplication build() { positionFactory, idFactory, descriptorFactory, + executorFactory, listeners == null ? Collections.emptySet() : Collections.unmodifiableCollection(listeners)); @@ -190,4 +204,13 @@ public WorkflowPositionFactory positionFactory() { public RuntimeDescriptorFactory runtimeDescriptorFactory() { return runtimeDescriptorFactory; } + + public ExecutorService executorService() { + synchronized (executorFactory) { + if (executorService == null) { + executorService = executorFactory.get(); + } + } + return executorService; + } } 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 db8b5e4d..39d03809 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java @@ -33,6 +33,7 @@ import java.util.Map; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; public class WorkflowDefinition implements AutoCloseable { @@ -134,6 +135,10 @@ public WorkflowPositionFactory positionFactory() { return application.positionFactory(); } + public ExecutorService executorService() { + return application.executorService(); + } + public RuntimeDescriptorFactory runtimeDescriptorFactory() { return application.runtimeDescriptorFactory(); } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java index ce72c70e..c121bb41 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowExecutionListener.java @@ -15,11 +15,11 @@ */ package io.serverlessworkflow.impl; -import io.serverlessworkflow.api.types.Task; +import io.serverlessworkflow.api.types.TaskBase; public interface WorkflowExecutionListener { - void onTaskStarted(WorkflowPosition currentPos, Task task); + void onTaskStarted(WorkflowPosition currentPos, TaskBase task); - void onTaskEnded(WorkflowPosition currentPos, Task task); + void onTaskEnded(WorkflowPosition currentPos, TaskBase task); } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java index 0f3bd410..8225439f 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java @@ -20,14 +20,15 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.NullNode; import java.time.Instant; +import java.util.concurrent.atomic.AtomicReference; public class WorkflowInstance { - private WorkflowState state; - private TaskContext taskContext; + private final AtomicReference state; + private final TaskContext taskContext; private final String id; private final JsonNode input; private final Instant startedAt; - private JsonNode context = NullNode.getInstance(); + private final AtomicReference context; WorkflowInstance(WorkflowDefinition definition, JsonNode input) { this.id = definition.idFactory().get(); @@ -39,7 +40,8 @@ public class WorkflowInstance { definition .inputFilter() .ifPresent(f -> taskContext.input(f.apply(workflowContext, taskContext, input))); - state = WorkflowState.STARTED; + state = new AtomicReference<>(WorkflowState.STARTED); + context = new AtomicReference<>(NullNode.getInstance()); WorkflowUtils.processTaskList(definition.workflow().getDo(), workflowContext, taskContext); definition .outputFilter() @@ -62,22 +64,26 @@ public JsonNode input() { } public JsonNode context() { - return context; + return context.get(); } public WorkflowState state() { - return state; + return state.get(); + } + + public void state(WorkflowState state) { + this.state.set(state); } public Object output() { return toJavaValue(taskContext.output()); } - public Object outputAsJsonNode() { + public JsonNode outputAsJsonNode() { return taskContext.output(); } void context(JsonNode context) { - this.context = context; + this.context.set(context); } } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java index 01486d5f..56b3499a 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java @@ -153,23 +153,8 @@ public static void processTaskList( TaskItem nextTask = iter.next(); while (nextTask != null) { TaskItem task = nextTask; - parentTask.position().addIndex(iter.previousIndex()).addProperty(task.getName()); - context - .definition() - .listeners() - .forEach(l -> l.onTaskStarted(parentTask.position(), task.getTask())); - currentContext = - context - .definition() - .taskExecutors() - .computeIfAbsent( - parentTask.position().jsonPointer(), - k -> - context - .definition() - .taskFactory() - .getTaskExecutor(task.getTask(), context.definition())) - .apply(context, parentTask, currentContext.output()); + parentTask.position().addIndex(iter.previousIndex()); + currentContext = executeTask(context, parentTask, task, currentContext.output()); FlowDirective flowDirective = currentContext.flowDirective(); if (flowDirective.getFlowDirectiveEnum() != null) { switch (flowDirective.getFlowDirectiveEnum()) { @@ -177,6 +162,7 @@ public static void processTaskList( nextTask = iter.hasNext() ? iter.next() : null; break; case END: + context.instance().state(WorkflowState.COMPLETED); case EXIT: nextTask = null; break; @@ -184,10 +170,6 @@ public static void processTaskList( } else { nextTask = WorkflowUtils.findTaskByName(iter, flowDirective.getString()); } - context - .definition() - .listeners() - .forEach(l -> l.onTaskEnded(parentTask.position(), task.getTask())); parentTask.position().back(); } } @@ -195,6 +177,25 @@ public static void processTaskList( parentTask.rawOutput(currentContext.output()); } + public static TaskContext executeTask( + WorkflowContext context, TaskContext parentTask, TaskItem task, JsonNode input) { + parentTask.position().addProperty(task.getName()); + TaskContext result = + context + .definition() + .taskExecutors() + .computeIfAbsent( + parentTask.position().jsonPointer(), + k -> + context + .definition() + .taskFactory() + .getTaskExecutor(task.getTask(), context.definition())) + .apply(context, parentTask, input); + parentTask.position().back(); + return result; + } + public static WorkflowFilter buildWorkflowFilter(ExpressionFactory exprFactory, String str) { assert str != null; Expression expression = exprFactory.getExpression(str); diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java index 086c7c51..44c6a7fd 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java @@ -26,7 +26,9 @@ import io.serverlessworkflow.impl.WorkflowContext; import io.serverlessworkflow.impl.WorkflowDefinition; import io.serverlessworkflow.impl.WorkflowFilter; +import io.serverlessworkflow.impl.WorkflowState; import io.serverlessworkflow.impl.jsonschema.SchemaValidator; +import java.time.Instant; import java.util.Optional; public abstract class AbstractTaskExecutor implements TaskExecutor { @@ -86,6 +88,14 @@ private void buildContextProcessors(WorkflowDefinition definition) { public TaskContext apply( WorkflowContext workflowContext, TaskContext parentContext, JsonNode input) { TaskContext taskContext = new TaskContext<>(input, parentContext, task); + if (workflowContext.instance().state() == WorkflowState.COMPLETED) { + return taskContext; + } + workflowContext + .definition() + .listeners() + .forEach(l -> l.onTaskStarted(parentContext.position(), task)); + inputSchemaValidator.ifPresent(s -> s.validate(taskContext.rawInput())); inputProcessor.ifPresent( p -> taskContext.input(p.apply(workflowContext, taskContext, taskContext.rawInput()))); @@ -98,6 +108,11 @@ public TaskContext apply( workflowContext.context( p.apply(workflowContext, taskContext, workflowContext.context()))); contextSchemaValidator.ifPresent(s -> s.validate(workflowContext.context())); + taskContext.completedAt(Instant.now()); + workflowContext + .definition() + .listeners() + .forEach(l -> l.onTaskEnded(parentContext.position(), task)); return taskContext; } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java index 58bd18ac..89c099b6 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java @@ -71,6 +71,8 @@ public TaskExecutor getTaskExecutor( return new RaiseExecutor(task.getRaiseTask(), definition); } else if (task.getTryTask() != null) { return new TryExecutor(task.getTryTask(), definition); + } else if (task.getForkTask() != null) { + return new ForkExecutor(task.getForkTask(), definition); } throw new UnsupportedOperationException(task.get().getClass().getName() + " not supported yet"); } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForkExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForkExecutor.java new file mode 100644 index 00000000..0193765c --- /dev/null +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForkExecutor.java @@ -0,0 +1,105 @@ +/* + * Copyright 2020-Present The Serverless Workflow Specification Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.serverlessworkflow.impl.executors; + +import io.serverlessworkflow.api.types.FlowDirectiveEnum; +import io.serverlessworkflow.api.types.ForkTask; +import io.serverlessworkflow.api.types.ForkTaskConfiguration; +import io.serverlessworkflow.api.types.TaskItem; +import io.serverlessworkflow.impl.TaskContext; +import io.serverlessworkflow.impl.WorkflowContext; +import io.serverlessworkflow.impl.WorkflowDefinition; +import io.serverlessworkflow.impl.WorkflowState; +import io.serverlessworkflow.impl.WorkflowUtils; +import io.serverlessworkflow.impl.json.JsonUtils; +import java.util.Comparator; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public class ForkExecutor extends AbstractTaskExecutor { + + private final ExecutorService service; + + protected ForkExecutor(ForkTask task, WorkflowDefinition definition) { + super(task, definition); + service = definition.executorService(); + } + + private record BranchContext(String taskName, TaskContext taskContext) {} + + @Override + protected void internalExecute(WorkflowContext workflow, TaskContext taskContext) { + ForkTaskConfiguration forkConfig = task.getFork(); + Map>> futures = new HashMap<>(); + int index = 0; + for (TaskItem item : forkConfig.getBranches()) { + final int i = index++; + futures.put( + item.getName(), + service.submit(() -> executeBranch(workflow, taskContext.copy(), item, i))); + } + + Stream sortedStream = + futures.entrySet().stream() + .map( + e -> { + try { + return new BranchContext(e.getKey(), e.getValue().get()); + } catch (ExecutionException ex) { + // todo handle error + throw new IllegalStateException(ex.getCause()); + } catch (InterruptedException ex) { + throw new IllegalStateException(ex); + } + }) + .sorted( + new Comparator() { + public int compare(BranchContext arg1, BranchContext arg2) { + return arg1.taskContext.completedAt().compareTo(arg2.taskContext.completedAt()); + } + }); + + taskContext.rawOutput( + forkConfig.isCompete() + ? sortedStream.map(e -> e.taskContext().output()).findFirst().orElseThrow() + : JsonUtils.fromValue( + sortedStream + .map( + e -> + JsonUtils.mapper() + .createObjectNode() + .set(e.taskName(), e.taskContext().output())) + .collect(Collectors.toList()))); + } + + private TaskContext executeBranch( + WorkflowContext workflow, TaskContext taskContext, TaskItem taskItem, int index) { + taskContext.position().addIndex(index); + TaskContext result = + WorkflowUtils.executeTask(workflow, taskContext, taskItem, taskContext.input()); + if (result.flowDirective() != null + && result.flowDirective().getFlowDirectiveEnum() == FlowDirectiveEnum.END) { + workflow.instance().state(WorkflowState.COMPLETED); + } + taskContext.position().back(); + return result; + } +} diff --git a/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java b/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java index 6b29bac1..e324de2a 100644 --- a/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java +++ b/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java @@ -19,6 +19,10 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.catchThrowableOfType; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import io.serverlessworkflow.impl.json.JsonUtils; import java.io.IOException; import java.time.Instant; import java.util.Arrays; @@ -108,7 +112,14 @@ private static Stream provideParameters() { args( "raise-reusable.yaml", WorkflowDefinitionTest::checkWorkflowException, - WorkflowException.class)); + WorkflowException.class), + args( + "fork.yaml", + Map.of(), + o -> + assertThat(((ObjectNode) o.outputAsJsonNode()).get("patientId").asText()) + .isIn("John", "Smith")), + args("fork-no-compete.yaml", Map.of(), WorkflowDefinitionTest::checkNotCompeteOuput)); } private static Arguments args( @@ -125,6 +136,24 @@ private static Arguments args( d -> consumer.accept(catchThrowableOfType(clazz, () -> d.execute(Map.of())))); } + private static void checkNotCompeteOuput(WorkflowInstance instance) { + JsonNode out = instance.outputAsJsonNode(); + assertThat(out).isInstanceOf(ArrayNode.class); + assertThat(out).hasSize(2); + ArrayNode array = (ArrayNode) out; + assertThat(array) + .containsExactlyInAnyOrder( + createObjectNode("callNurse", "patientId", "John", "room", 1), + createObjectNode("callDoctor", "patientId", "Smith", "room", 2)); + } + + private static JsonNode createObjectNode( + String parent, String key1, String value1, String key2, int value2) { + return JsonUtils.mapper() + .createObjectNode() + .set(parent, JsonUtils.mapper().createObjectNode().put(key1, value1).put(key2, value2)); + } + private static void checkWorkflowException(WorkflowException ex) { assertThat(ex.getWorflowError().type()) .isEqualTo("https://serverlessworkflow.io/errors/not-implemented"); diff --git a/impl/core/src/test/resources/fork-no-compete.yaml b/impl/core/src/test/resources/fork-no-compete.yaml new file mode 100644 index 00000000..5e78acf2 --- /dev/null +++ b/impl/core/src/test/resources/fork-no-compete.yaml @@ -0,0 +1,18 @@ +document: + dsl: '1.0.0-alpha5' + namespace: test + name: fork-example + version: '0.1.0' +do: + - callSomeone: + fork: + compete: false + branches: + - callNurse: + set: + patientId: John + room: 1 + - callDoctor: + set: + patientId: Smith + room: 2 \ No newline at end of file diff --git a/impl/core/src/test/resources/fork.yaml b/impl/core/src/test/resources/fork.yaml new file mode 100644 index 00000000..dfde183b --- /dev/null +++ b/impl/core/src/test/resources/fork.yaml @@ -0,0 +1,18 @@ +document: + dsl: '1.0.0-alpha5' + namespace: test + name: fork-no-compete + version: '0.1.0' +do: + - callSomeone: + fork: + compete: true + branches: + - callNurse: + set: + patientId: John + room: 1 + - callDoctor: + set: + patientId: Smith + room: 2 \ No newline at end of file