Skip to content

Commit 8fd32b7

Browse files
matheusandre1matheusandre1
committed
feat: add sub worflows run on FuncDSL
Signed-off-by: Matheus Andre <matheusandr2@gmail.com>
1 parent 3dfd814 commit 8fd32b7

5 files changed

Lines changed: 235 additions & 11 deletions

File tree

‎experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncDoTaskBuilder.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import io.serverlessworkflow.fluent.func.spi.FuncDoFluent;
2020
import io.serverlessworkflow.fluent.func.spi.FuncTaskTransformations;
2121
import io.serverlessworkflow.fluent.spec.BaseDoTaskBuilder;
22+
import io.serverlessworkflow.fluent.spec.WorkflowTaskBuilder;
2223
import java.util.function.Consumer;
2324

2425
public class FuncDoTaskBuilder extends BaseDoTaskBuilder<FuncDoTaskBuilder, FuncTaskItemListBuilder>
@@ -96,4 +97,22 @@ public FuncDoTaskBuilder openapi(
9697
this.listBuilder().openapi(name, itemsConfigurer);
9798
return this;
9899
}
100+
101+
@Override
102+
public FuncDoTaskBuilder workflow(String name, Consumer<WorkflowTaskBuilder> itemsConfigurer) {
103+
this.listBuilder().workflow(name, itemsConfigurer);
104+
return this;
105+
}
106+
107+
@Override
108+
public FuncDoTaskBuilder subflow(String name, Consumer<WorkflowTaskBuilder> itemsConfigurer) {
109+
this.listBuilder().subflow(name, itemsConfigurer);
110+
return this;
111+
}
112+
113+
@Override
114+
public FuncDoTaskBuilder subflow(Consumer<WorkflowTaskBuilder> itemsConfigurer) {
115+
this.listBuilder().subflow(itemsConfigurer);
116+
return this;
117+
}
99118
}

‎experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncTaskItemListBuilder.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
import io.serverlessworkflow.api.types.TaskItem;
2323
import io.serverlessworkflow.fluent.func.spi.FuncDoFluent;
2424
import io.serverlessworkflow.fluent.spec.BaseTaskItemListBuilder;
25+
import io.serverlessworkflow.fluent.spec.TaskItemListBuilder;
26+
import io.serverlessworkflow.fluent.spec.WorkflowTaskBuilder;
2527
import java.util.List;
2628
import java.util.function.Consumer;
2729

@@ -154,4 +156,28 @@ public FuncTaskItemListBuilder openapi(
154156

155157
return this.addTaskItem(new TaskItem(name, task));
156158
}
159+
160+
@Override
161+
public FuncTaskItemListBuilder workflow(
162+
String name, Consumer<WorkflowTaskBuilder> itemsConfigurer) {
163+
final TaskItemListBuilder delegate = new TaskItemListBuilder(this.mutableList().size());
164+
delegate.workflow(name, itemsConfigurer);
165+
final List<TaskItem> taskItems = delegate.build();
166+
if (taskItems.size() != 1) {
167+
throw new IllegalStateException(
168+
"Expected workflow delegate to build exactly 1 TaskItem, but got " + taskItems.size());
169+
}
170+
return addTaskItem(taskItems.get(0));
171+
}
172+
173+
@Override
174+
public FuncTaskItemListBuilder subflow(
175+
String name, Consumer<WorkflowTaskBuilder> itemsConfigurer) {
176+
return this.workflow(name, itemsConfigurer);
177+
}
178+
179+
@Override
180+
public FuncTaskItemListBuilder subflow(Consumer<WorkflowTaskBuilder> itemsConfigurer) {
181+
return this.workflow(itemsConfigurer);
182+
}
157183
}

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

Lines changed: 118 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -38,9 +38,12 @@
3838
import io.serverlessworkflow.fluent.spec.EventFilterBuilder;
3939
import io.serverlessworkflow.fluent.spec.ScheduleBuilder;
4040
import io.serverlessworkflow.fluent.spec.TimeoutBuilder;
41+
import io.serverlessworkflow.fluent.spec.WorkflowTaskBuilder;
4142
import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer;
43+
import io.serverlessworkflow.fluent.spec.configurers.WorkflowConfigurer;
4244
import io.serverlessworkflow.fluent.spec.dsl.DSL;
4345
import io.serverlessworkflow.fluent.spec.dsl.UseSpec;
46+
import io.serverlessworkflow.fluent.spec.dsl.WorkflowSpec;
4447
import io.serverlessworkflow.impl.TaskContextData;
4548
import io.serverlessworkflow.impl.WorkflowContextData;
4649
import java.net.URI;
@@ -927,14 +930,19 @@ public static FuncTaskConfigurer switchWhen(String jqExpression, String thenTask
927930
* @return list configurer
928931
*/
929932
public static <T> FuncTaskConfigurer switchWhenOrElse(
930-
Predicate<T> pred, String thenTask, FlowDirectiveEnum otherwise, Class<T> predClass) {
933+
Predicate<T> pred,
934+
String thenTask,
935+
io.serverlessworkflow.api.types.FlowDirectiveEnum otherwise,
936+
Class<T> predClass) {
931937
return list ->
932938
list.switchCase(
933939
FuncDSL.cases(caseOf(pred, predClass).then(thenTask), caseDefault(otherwise)));
934940
}
935941

936942
public static <T> FuncTaskConfigurer switchWhenOrElse(
937-
SerializablePredicate<T> pred, String thenTask, FlowDirectiveEnum otherwise) {
943+
SerializablePredicate<T> pred,
944+
String thenTask,
945+
io.serverlessworkflow.api.types.FlowDirectiveEnum otherwise) {
938946
return switchWhenOrElse(pred, thenTask, otherwise, ReflectionUtils.inferInputType(pred));
939947
}
940948

@@ -975,7 +983,9 @@ public static <T> FuncTaskConfigurer switchWhenOrElse(
975983
* @return list configurer
976984
*/
977985
public static FuncTaskConfigurer switchWhenOrElse(
978-
String jqExpression, String thenTask, FlowDirectiveEnum otherwise) {
986+
String jqExpression,
987+
String thenTask,
988+
io.serverlessworkflow.api.types.FlowDirectiveEnum otherwise) {
979989

980990
Objects.requireNonNull(jqExpression, "jqExpression");
981991
Objects.requireNonNull(thenTask, "thenTask");
@@ -1072,6 +1082,111 @@ public static FuncTaskConfigurer set(Map<String, Object> map) {
10721082
return list -> list.set(s -> s.expr(map));
10731083
}
10741084

1085+
/**
1086+
* Create a {@link FuncTaskConfigurer} that adds a workflow subflow task.
1087+
*
1088+
* @param configurer configurer for the nested workflow task
1089+
* @return a {@link FuncTaskConfigurer} that adds a workflow task to the tasks list
1090+
*/
1091+
/**
1092+
* Create a {@link FuncTaskConfigurer} that adds a sub-workflow call task using a {@link
1093+
* WorkflowConfigurer}.
1094+
*
1095+
* <p>
1096+
*
1097+
* <pre>{@code
1098+
* tasks(
1099+
* subflow(workflow("org.acme", "sub-workflow", "0.1.0").input(...)
1100+
* );
1101+
* }</pre>
1102+
*
1103+
* @param configurer nested workflow configurer
1104+
* @return a {@link FuncTaskConfigurer} that adds a workflow task to the tasks list
1105+
*/
1106+
public static FuncTaskConfigurer subflow(WorkflowConfigurer configurer) {
1107+
Objects.requireNonNull(configurer, "configurer");
1108+
return list -> list.subflow(configurer);
1109+
}
1110+
1111+
/**
1112+
* Create a {@link FuncTaskConfigurer} that adds a named sub-workflow call task.
1113+
*
1114+
* @param name task name
1115+
* @param configurer nested workflow configurer
1116+
* @return a {@link FuncTaskConfigurer} that adds a workflow task to the tasks list
1117+
*/
1118+
public static FuncTaskConfigurer subflow(String name, Consumer<WorkflowTaskBuilder> configurer) {
1119+
Objects.requireNonNull(name, "name");
1120+
Objects.requireNonNull(configurer, "configurer");
1121+
return list -> list.subflow(name, configurer);
1122+
}
1123+
1124+
/**
1125+
* Create a {@link FuncTaskConfigurer} that adds an unnamed sub-workflow call task.
1126+
*
1127+
* @param configurer nested workflow configurer
1128+
* @return a {@link FuncTaskConfigurer} that adds a workflow task to the tasks list
1129+
*/
1130+
public static FuncTaskConfigurer subflow(Consumer<WorkflowTaskBuilder> configurer) {
1131+
Objects.requireNonNull(configurer, "configurer");
1132+
return list -> list.subflow(configurer);
1133+
}
1134+
1135+
/**
1136+
* Alias for {@link #subflow(WorkflowConfigurer)}.
1137+
*
1138+
* @param configurer nested workflow configurer
1139+
* @return a {@link FuncTaskConfigurer} that adds a workflow task to the tasks list
1140+
*/
1141+
public static FuncTaskConfigurer workflowTask(WorkflowConfigurer configurer) {
1142+
return subflow(configurer);
1143+
}
1144+
1145+
/**
1146+
* Create a {@link FuncTaskConfigurer} that adds a workflow subflow task.
1147+
*
1148+
* @param configurer configurer for the nested workflow task
1149+
* @return a {@link FuncTaskConfigurer} that adds a workflow task to the tasks list
1150+
* @deprecated use {@link #subflow(WorkflowConfigurer)} to avoid ambiguity with spec-side factory
1151+
* methods
1152+
*/
1153+
@Deprecated
1154+
public static FuncTaskConfigurer workflow(WorkflowConfigurer configurer) {
1155+
return subflow(configurer);
1156+
}
1157+
1158+
/**
1159+
* Create a new {@link WorkflowSpec} to be used as a factory for workflow definitions.
1160+
*
1161+
* @param namespace workflow namespace
1162+
* @param name workflow name
1163+
* @param version workflow version
1164+
* @return a new {@link WorkflowSpec} instance
1165+
*/
1166+
public static WorkflowSpec workflow(String namespace, String name, String version) {
1167+
return DSL.workflow(namespace, name, version);
1168+
}
1169+
1170+
/**
1171+
* Create a new {@link WorkflowSpec} to be used as a factory for workflow definitions.
1172+
*
1173+
* @param namespace workflow namespace
1174+
* @param name workflow name
1175+
* @return a new {@link WorkflowSpec} instance
1176+
*/
1177+
public static WorkflowSpec workflow(String namespace, String name) {
1178+
return DSL.workflow(namespace, name);
1179+
}
1180+
1181+
/**
1182+
* Create a new {@link WorkflowSpec} to be used as a factory for workflow definitions.
1183+
*
1184+
* @return a new {@link WorkflowSpec} instance
1185+
*/
1186+
public static WorkflowSpec workflow() {
1187+
return DSL.workflow();
1188+
}
1189+
10751190
// ---------------------------------------------------------------------------
10761191
// HTTP / OpenAPI
10771192
// ---------------------------------------------------------------------------

‎experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/spi/FuncDoFluent.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import io.serverlessworkflow.fluent.func.FuncListenTaskBuilder;
2525
import io.serverlessworkflow.fluent.func.FuncSetTaskBuilder;
2626
import io.serverlessworkflow.fluent.func.FuncSwitchTaskBuilder;
27+
import io.serverlessworkflow.fluent.spec.WorkflowTaskBuilder;
2728
import io.serverlessworkflow.fluent.spec.spi.CallHttpFluent;
2829
import io.serverlessworkflow.fluent.spec.spi.CallOpenAPIFluent;
2930
import io.serverlessworkflow.fluent.spec.spi.EmitFluent;
@@ -32,6 +33,8 @@
3233
import io.serverlessworkflow.fluent.spec.spi.ListenFluent;
3334
import io.serverlessworkflow.fluent.spec.spi.SetFluent;
3435
import io.serverlessworkflow.fluent.spec.spi.SwitchFluent;
36+
import io.serverlessworkflow.fluent.spec.spi.WorkflowFluent;
37+
import java.util.function.Consumer;
3538

3639
public interface FuncDoFluent<SELF extends FuncDoFluent<SELF>>
3740
extends SetFluent<FuncSetTaskBuilder, SELF>,
@@ -42,4 +45,14 @@ public interface FuncDoFluent<SELF extends FuncDoFluent<SELF>>
4245
ListenFluent<FuncListenTaskBuilder, SELF>,
4346
CallFnFluent<FuncCallTaskBuilder, SELF>,
4447
CallHttpFluent<FuncCallHttpTaskBuilder, SELF>,
45-
CallOpenAPIFluent<FuncCallOpenAPITaskBuilder, SELF> {}
48+
CallOpenAPIFluent<FuncCallOpenAPITaskBuilder, SELF>,
49+
WorkflowFluent<WorkflowTaskBuilder, SELF> {
50+
51+
default SELF subflow(String name, Consumer<WorkflowTaskBuilder> itemsConfigurer) {
52+
return this.workflow(name, itemsConfigurer);
53+
}
54+
55+
default SELF subflow(Consumer<WorkflowTaskBuilder> itemsConfigurer) {
56+
return this.workflow(itemsConfigurer);
57+
}
58+
}

‎experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java‎

Lines changed: 58 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -23,17 +23,19 @@
2323
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.http;
2424
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.listen;
2525
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.produced;
26+
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.subflow;
2627
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.switchWhenOrElse;
2728
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.toOne;
29+
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.workflow;
2830
import static io.serverlessworkflow.fluent.spec.dsl.DSL.use;
2931
import static org.junit.jupiter.api.Assertions.assertEquals;
32+
import static org.junit.jupiter.api.Assertions.assertFalse;
3033
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
3134
import static org.junit.jupiter.api.Assertions.assertNotNull;
3235
import static org.junit.jupiter.api.Assertions.assertNull;
3336

3437
import io.serverlessworkflow.api.types.CallHTTP;
3538
import io.serverlessworkflow.api.types.Export;
36-
import io.serverlessworkflow.api.types.FlowDirectiveEnum;
3739
import io.serverlessworkflow.api.types.Task;
3840
import io.serverlessworkflow.api.types.TaskItem;
3941
import io.serverlessworkflow.api.types.Workflow;
@@ -174,6 +176,50 @@ void mixed_chaining_order_and_exports() {
174176
assertNotNull(t2.getListenTask().getExport(), "listen step should carry export");
175177
}
176178

179+
@Test
180+
void subflow_task_builds_from_func_dsl() {
181+
Workflow wf =
182+
FuncWorkflowBuilder.workflow("step-subflow")
183+
.tasks(
184+
subflow(workflow("child.ns", "child-flow", "2.3.4").input("id", 99).await(false)))
185+
.build();
186+
187+
List<TaskItem> items = wf.getDo();
188+
assertEquals(1, items.size());
189+
190+
Task t = items.get(0).getTask();
191+
assertNotNull(t.getRunTask(), "RunTask expected");
192+
var run = t.getRunTask().getRun().getRunWorkflow();
193+
assertNotNull(run, "RunWorkflow should be present");
194+
assertEquals("child.ns", run.getWorkflow().getNamespace());
195+
assertEquals("child-flow", run.getWorkflow().getName());
196+
assertEquals("2.3.4", run.getWorkflow().getVersion());
197+
assertEquals(99, run.getWorkflow().getInput().getAdditionalProperties().get("id"));
198+
assertFalse(run.isAwait(), "await(false) should be preserved as false");
199+
}
200+
201+
@Test
202+
void subflow_builder_style_builds_correctly() {
203+
Workflow wf =
204+
FuncWorkflowBuilder.workflow("subflow-builder")
205+
.tasks(
206+
subflow(
207+
"my-subflow",
208+
sub ->
209+
sub.namespace("child.ns")
210+
.name("child-flow")
211+
.version("1.0.0")
212+
.input(Map.of("key", "val"))))
213+
.build();
214+
215+
Task t = wf.getDo().get(0).getTask();
216+
assertNotNull(t.getRunTask());
217+
var run = t.getRunTask().getRun().getRunWorkflow();
218+
assertEquals("my-subflow", wf.getDo().get(0).getName());
219+
assertEquals("child-flow", run.getWorkflow().getName());
220+
assertEquals("val", run.getWorkflow().getInput().getAdditionalProperties().get("key"));
221+
}
222+
177223
@Test
178224
void switchWhenOrElse_jq_to_taskName() {
179225
Workflow wf =
@@ -191,13 +237,16 @@ void switchWhenOrElse_jq_to_taskName() {
191237
void switchWhenOrElse_jq_to_directive() {
192238
Workflow wf =
193239
FuncWorkflowBuilder.workflow("jqSwitchDir")
194-
.tasks(switchWhenOrElse(".score >= 80", "pass", FlowDirectiveEnum.END))
240+
.tasks(
241+
switchWhenOrElse(
242+
".score >= 80", "pass", io.serverlessworkflow.api.types.FlowDirectiveEnum.END))
195243
.build();
196244
Task switchTask = wf.getDo().get(0).getTask();
197245
var items = switchTask.getSwitchTask().getSwitch();
198246
assertEquals(".score >= 80", items.get(0).getSwitchCase().getWhen());
199247
assertEquals(
200-
FlowDirectiveEnum.END, items.get(1).getSwitchCase().getThen().getFlowDirectiveEnum());
248+
io.serverlessworkflow.api.types.FlowDirectiveEnum.END,
249+
items.get(1).getSwitchCase().getThen().getFlowDirectiveEnum());
201250
}
202251

203252
@Test
@@ -447,7 +496,9 @@ void function_step_then_task_name_sets_flow_directive() {
447496
void function_step_then_flow_directive_enum_sets_end() {
448497
Workflow wf =
449498
FuncWorkflowBuilder.workflow("intelligent-newsletter")
450-
.tasks(function("myfunction", String::trim, String.class).then(FlowDirectiveEnum.END))
499+
.tasks(
500+
function("myfunction", String::trim, String.class)
501+
.then(io.serverlessworkflow.api.types.FlowDirectiveEnum.END))
451502
.build();
452503

453504
List<TaskItem> items = wf.getDo();
@@ -459,7 +510,7 @@ void function_step_then_flow_directive_enum_sets_end() {
459510
CallJava callJava = (CallJava) t.getCallTask().get();
460511
assertNotNull(callJava.getThen(), "then() should be set on the task");
461512
assertEquals(
462-
FlowDirectiveEnum.END,
513+
io.serverlessworkflow.api.types.FlowDirectiveEnum.END,
463514
callJava.getThen().getFlowDirectiveEnum(),
464515
"then() should be FlowDirectiveEnum.END");
465516
}
@@ -495,7 +546,7 @@ void consume_step_then_flow_directive_enum_sets_end() {
495546
FuncWorkflowBuilder.workflow("intelligent-newsletter")
496547
.tasks(
497548
consume("sendNewsletter", (String s) -> {}, String.class)
498-
.then(FlowDirectiveEnum.END))
549+
.then(io.serverlessworkflow.api.types.FlowDirectiveEnum.END))
499550
.build();
500551

501552
List<TaskItem> items = wf.getDo();
@@ -507,7 +558,7 @@ void consume_step_then_flow_directive_enum_sets_end() {
507558
CallJava callJava = (CallJava) t.getCallTask().get();
508559
assertNotNull(callJava.getThen(), "then() should be set on the consume task");
509560
assertEquals(
510-
FlowDirectiveEnum.END,
561+
io.serverlessworkflow.api.types.FlowDirectiveEnum.END,
511562
callJava.getThen().getFlowDirectiveEnum(),
512563
"then() should be FlowDirectiveEnum.END");
513564
}

0 commit comments

Comments
 (0)