Add then to the Java DSL

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/Step.java

@@ -15,6 +15,7 @@
  */
 package io.serverlessworkflow.fluent.func.dsl;
 
+import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.api.types.func.JavaContextFunction;
 import io.serverlessworkflow.api.types.func.JavaFilterFunction;
 import io.serverlessworkflow.fluent.func.FuncTaskItemListBuilder;
@@ -62,6 +63,36 @@ public SELF when(String jqExpr) {
     return self();
   }
 
+  /**
+   * Queue a {@code then(taskName)} to be applied on the concrete builder. Directs the workflow
+   * engine to jump to the named task after this one completes.
+   *
+   * @param taskName the name of the next task to execute
+   * @return this step for further chaining
+   * @see <a
+   *     href="https://github.com/serverlessworkflow/specification/blob/main/dsl-reference.md#task">DSL
+   *     Reference - Task</a>
+   */
+  public SELF then(String taskName) {
+    postConfigurers.add(b -> ((TaskBaseBuilder<?>) b).then(taskName));
+    return self();
+  }
+
+  /**
+   * Queue a {@code then(directive)} to be applied on the concrete builder. Directs the workflow
+   * engine to apply the given flow directive after this task completes.
+   *
+   * @param directive the flow directive (e.g., {@link FlowDirectiveEnum#END})
+   * @return this step for further chaining
+   * @see <a
+   *     href="https://github.com/serverlessworkflow/specification/blob/main/dsl-reference.md#task">DSL
+   *     Reference - Task</a>
+   */
+  public SELF then(FlowDirectiveEnum directive) {
+    postConfigurers.add(b -> ((TaskBaseBuilder<?>) b).then(directive));
+    return self();
+  }
+
   // ---------------------------------------------------------------------------
   // FuncTaskTransformations passthroughs: EXPORT (fn/context/filter + JQ)
   // ---------------------------------------------------------------------------

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

@@ -16,14 +16,14 @@
 package io.serverlessworkflow.fluent.func;
 
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.call;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.consume;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.emit;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.event;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.function;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.get;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.http;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.listen;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.toOne;
-import static io.serverlessworkflow.fluent.spec.dsl.DSL.auth;
 import static io.serverlessworkflow.fluent.spec.dsl.DSL.use;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -432,4 +432,95 @@ void call_with_preconfigured_http_spec() {
             .getUse());
     assertEquals(Map.of("foo", "bar"), http.getWith().getBody());
   }
+
+  @Test
+  @DisplayName("function(name, fn).then(taskName) sets FlowDirective string on the task")
+  void function_step_then_task_name_sets_flow_directive() {
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("intelligent-newsletter")
+            .tasks(
+                function("myfunction", String::trim, String.class).then("otherTask"),
+                function("otherTask", String::strip, String.class))
+            .build();
+
+    List<TaskItem> items = wf.getDo();
+    assertEquals(2, items.size());
+
+    Task t = items.get(0).getTask();
+    assertNotNull(t.getCallTask(), "CallTask expected");
+
+    CallJava callJava = (CallJava) t.getCallTask().get();
+    assertNotNull(callJava.getThen(), "then() should be set on the task");
+    assertEquals("otherTask", callJava.getThen().getString(), "then() should point to 'otherTask'");
+  }
+
+  @Test
+  @DisplayName("function(name, fn).then(FlowDirectiveEnum.END) sets END directive on the task")
+  void function_step_then_flow_directive_enum_sets_end() {
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("intelligent-newsletter")
+            .tasks(function("myfunction", String::trim, String.class).then(FlowDirectiveEnum.END))
+            .build();
+
+    List<TaskItem> items = wf.getDo();
+    assertEquals(1, items.size());
+
+    Task t = items.get(0).getTask();
+    assertNotNull(t.getCallTask(), "CallTask expected");
+
+    CallJava callJava = (CallJava) t.getCallTask().get();
+    assertNotNull(callJava.getThen(), "then() should be set on the task");
+    assertEquals(
+        FlowDirectiveEnum.END,
+        callJava.getThen().getFlowDirectiveEnum(),
+        "then() should be FlowDirectiveEnum.END");
+  }
+
+  @Test
+  @DisplayName(
+      "consume(name, Consumer, Class).then(taskName) sets FlowDirective string on the task")
+  void consume_step_then_task_name_sets_flow_directive() {
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("intelligent-newsletter")
+            .tasks(
+                consume("sendNewsletter", (String s) -> {}, String.class).then("otherTask"),
+                function("nextTask", String::strip, String.class),
+                function("otherTask", String::strip, String.class))
+            .build();
+
+    List<TaskItem> items = wf.getDo();
+    assertEquals(3, items.size());
+
+    Task t = items.get(0).getTask();
+    assertNotNull(t.getCallTask(), "CallTask expected for consume step");
+
+    CallJava callJava = (CallJava) t.getCallTask().get();
+    assertNotNull(callJava.getThen(), "then() should be set on the consume task");
+    assertEquals("otherTask", callJava.getThen().getString(), "then() should point to 'otherTask'");
+  }
+
+  @Test
+  @DisplayName(
+      "consume(name, Consumer, Class).then(FlowDirectiveEnum.END) sets END directive on the task")
+  void consume_step_then_flow_directive_enum_sets_end() {
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("intelligent-newsletter")
+            .tasks(
+                consume("sendNewsletter", (String s) -> {}, String.class)
+                    .then(FlowDirectiveEnum.END))
+            .build();
+
+    List<TaskItem> items = wf.getDo();
+    assertEquals(1, items.size());
+
+    Task t = items.get(0).getTask();
+    assertNotNull(t.getCallTask(), "CallTask expected for consume step");
+
+    CallJava callJava = (CallJava) t.getCallTask().get();
+    assertNotNull(callJava.getThen(), "then() should be set on the consume task");
+    assertEquals(
+        FlowDirectiveEnum.END,
+        callJava.getThen().getFlowDirectiveEnum(),
+        "then() should be FlowDirectiveEnum.END");
+  }
 }

impl/test/pom.xml

@@ -94,6 +94,18 @@
             <artifactId>grpc-netty</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-experimental-fluent-func</artifactId>
+            <scope>test</scope>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-experimental-lambda</artifactId>
+            <scope>test</scope>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
     <build>
         <resources>

impl/test/src/test/java/io/serverlessworkflow/impl/test/WorkflowThenTest.java

@@ -0,0 +1,122 @@
+/*
+ * 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.test;
+
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.consume;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.function;
+
+import io.serverlessworkflow.api.types.FlowDirectiveEnum;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class WorkflowThenTest {
+
+  @Test
+  void consume_then_skips_next_task_and_jumps_to_target() {
+    AtomicInteger a = new AtomicInteger();
+    AtomicInteger b = new AtomicInteger();
+
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("intelligent-newsletter")
+            .tasks(
+                consume("sendNewsletter", (String s) -> {}, String.class).then("otherTask"),
+                function(
+                    "nextTask",
+                    (v) -> {
+                      a.incrementAndGet();
+                      return v.strip();
+                    },
+                    String.class),
+                function(
+                    "otherTask",
+                    (v) -> {
+                      b.incrementAndGet();
+                      return v.strip();
+                    },
+                    String.class))
+            .build();
+
+    WorkflowApplication app = WorkflowApplication.builder().build();
+    WorkflowDefinition def = app.workflowDefinition(wf);
+    def.instance(Map.of()).start().join();
+
+    Assertions.assertEquals(1, b.get(), "otherTask should execute");
+    Assertions.assertEquals(0, a.get(), "nextTask should not execute");
+  }
+
+  @Test
+  void function_then_skips_next_task_and_jumps_to_target() {
+    AtomicInteger a = new AtomicInteger();
+    AtomicInteger b = new AtomicInteger();
+
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("intelligent-newsletter")
+            .tasks(
+                function("myfunction", String::trim, String.class).then("otherTask"),
+                function(
+                    "nextTask",
+                    (v) -> {
+                      a.incrementAndGet();
+                      return v.strip();
+                    },
+                    String.class),
+                function(
+                    "otherTask",
+                    (v) -> {
+                      b.incrementAndGet();
+                      return v.strip();
+                    },
+                    String.class))
+            .build();
+
+    WorkflowApplication app = WorkflowApplication.builder().build();
+    WorkflowDefinition def = app.workflowDefinition(wf);
+    def.instance(Map.of()).start().join();
+
+    Assertions.assertEquals(1, b.get(), "otherTask should execute");
+    Assertions.assertEquals(0, a.get(), "nextTask should not execute");
+  }
+
+  @Test
+  void function_then_end_directive_stops_workflow_execution() {
+    AtomicInteger a = new AtomicInteger();
+
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("intelligent-newsletter")
+            .tasks(
+                function("myfunction", String::trim, String.class).then(FlowDirectiveEnum.END),
+                function(
+                    "nextTask",
+                    (v) -> {
+                      a.incrementAndGet();
+                      return v.strip();
+                    },
+                    String.class))
+            .build();
+
+    WorkflowApplication app = WorkflowApplication.builder().build();
+    WorkflowDefinition def = app.workflowDefinition(wf);
+    def.instance(Map.of()).start().join();
+
+    Assertions.assertEquals(0, a.get(), "nextTask should not execute when then(END) is set");
+  }
+}