[waterflow] support branch syntax for 'when' and 'then' clauses

Author: loveTsong

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/context/repo/flowcontext/FlowContextMessenger.java

@@ -54,4 +54,18 @@ default <I> void send(ProcessType type, Subscriber<I, ?> subscriber, List<FlowCo
      * @param <I> 流程实例执行时的入参数据类型，用于泛型推倒
      */
     <I> void sendCallback(List<FlowContext<I>> contexts);
+
+    /**
+     * Directly processes a list of flow contexts through the specified subscriber.
+     * This method serves as a default implementation for immediate processing without
+     * any intermediate transformations or routing.
+     *
+     * @param <I> The type of input data contained in the flow contexts.
+     * @param type The type of processing to be performed.
+     * @param subscriber The subscriber that will handle the processing.
+     * @param context List of flow contexts to be processed.
+     */
+    default <I> void directProcess(ProcessType type, Subscriber<I, ?> subscriber, List<FlowContext<I>> context) {
+        subscriber.process(type, context);
+    }
 }

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/states/Conditions.java

@@ -49,6 +49,21 @@ public <O> MatchHappen<O, D, I, F> match(Operators.Whether<I> whether,
         return new MatchHappen<>(whether, processor, this);
     }
 
+    /**
+     * Creates a conditional branch that executes when the specified condition is met.
+     * This establishes a processing path that will only be followed if the predicate
+     * evaluates to true for the input data.
+     *
+     * @param <O> The output type of the branch processor.
+     * @param whether The condition predicate that determines branch activation.
+     * @param processor The transformation to apply when the condition is met.
+     * @return A {@link WhenHappen} instance representing the conditional relationship,
+     *         allowing for further chaining of operations.
+     */
+    public <O> WhenHappen<O, D, I, F> when(Operators.Whether<I> whether, Operators.Then<I, O> processor) {
+        return new WhenHappen<>(whether, processor, this);
+    }
+
     /**
      * 在满足条件时跳转到指定节点。
      *

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/states/WhenHappen.java

@@ -0,0 +1,64 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.
+ *  This file is a part of the ModelEngine Project.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+package modelengine.fit.waterflow.domain.states;
+
+import modelengine.fit.waterflow.domain.enums.SpecialDisplayNode;
+import modelengine.fit.waterflow.domain.flow.Flow;
+import modelengine.fit.waterflow.domain.stream.operators.Operators;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Represent conditional branches (when clause) in a processing flow.
+ *
+ * @param <O> The output data type of this node.
+ * @param <D> The initial data type of the containing flow.
+ * @param <F> The flow type used for generic type inference.
+ *
+ * @author 宋永坦
+ * @since 2025-06-12
+ */
+public class WhenHappen<O, D, I, F extends Flow<D>> {
+    private final State<I, D, I, F> node;
+    private final List<State<O, D, ?, F>> branches = new ArrayList<>();
+
+    WhenHappen(Operators.Whether<I> whether, Operators.Then<I, O> processor, Conditions<D, I, F> conditions) {
+        this.node = conditions.node;
+        this.when(whether, processor);
+    }
+
+    /**
+     * Creates a conditional branch that executes when the specified condition is met.
+     *
+     * @param whether The condition predicate.
+     * @param processor The transformation logic.
+     * @return The current WhenHappen instance for method chaining.
+     */
+    public WhenHappen<O, D, I, F> when(Operators.Whether<I> whether, Operators.Then<I, O> processor) {
+        State<O, D, I, F> branch = new State<>(this.node.publisher()
+                .map(input -> processor.process(input.getData()), whether)
+                .displayAs(SpecialDisplayNode.BRANCH.name()), this.node.getFlow());
+        this.branches.add(branch);
+        return this;
+    }
+
+    /**
+     * Provides a default processing logic and terminates the conditional node.
+     *
+     * @param processor The handler to process unmatched inputs.
+     * @return An {@link State} representing the join node of the conditional flow.
+     * @throws IllegalArgumentException if processor is null
+     */
+    public State<O, D, O, F> others(Operators.Then<I, O> processor) {
+        this.when(null, processor);
+        State<O, D, O, F> joinState = this.branches.get(0).just(any -> {});
+        joinState.processor.displayAs(SpecialDisplayNode.OTHERS.name());
+        this.branches.stream().skip(1).forEach(branch -> branch.publisher().subscribe(joinState.subscriber()));
+        return joinState;
+    }
+}

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/nodes/ConditionsNode.java

@@ -12,6 +12,7 @@
 import modelengine.fit.waterflow.domain.context.repo.flowlock.FlowLocks;
 import modelengine.fit.waterflow.domain.enums.FlowNodeType;
 import modelengine.fit.waterflow.domain.utils.UUIDUtil;
+import modelengine.fitframework.util.CollectionUtils;
 
 import java.util.List;
 import java.util.stream.Collectors;
@@ -82,7 +83,12 @@ public void offer(List<FlowContext<I>> contexts) {
                         .filter(context -> subscription.getWhether().is(context.getData()))
                         .collect(Collectors.toList());
                 matched.forEach(contexts::remove);
-                subscription.cache(matched);
+                // For order-sensitive data, directly synchronously executes the next conditional branch node.
+                if (CollectionUtils.isNotEmpty(matched) && matched.get(0).getSession().preserved()) {
+                    subscription.process(matched);
+                } else {
+                    subscription.cache(matched);
+                }
             });
         }
 

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/nodes/To.java

@@ -140,7 +140,7 @@ public class To<I, O> extends IdGenerator implements Subscriber<I, O> {
     @Getter
     private ProcessMode processMode;
 
-    private Map<String, Integer> processingSessions = new ConcurrentHashMap<>();
+    private final Map<String, Integer> processingSessions = new ConcurrentHashMap<>();
 
     private Operators.Validator<I> validator = (repo, to) -> repo.requestMappingContext(to.streamId,
             to.froms.stream().map(Identity::getId).collect(Collectors.toList()),
@@ -291,6 +291,38 @@ public synchronized void accept(ProcessType type, List<FlowContext<I>> contexts)
         this.triggerNodeProcessor(type);
     }
 
+    @Override
+    public void process(ProcessType type, List<FlowContext<I>> contexts) {
+        Validation.isTrue(ProcessType.PROCESS.equals(type),
+                "Direct processing requires PROCESS type, but received: " + type);
+        this.directProcess(contexts);
+    }
+
+    private void directProcess(List<FlowContext<I>> preList) {
+        try {
+            if (CollectionUtils.isEmpty(preList)) {
+                return;
+            }
+            if (preList.size() == 1 && preList.get(0).getData() == null) {
+                this.afterProcess(preList, new ArrayList<>());
+                return;
+            }
+            List<FlowContext<O>> afterList = this.getProcessMode().process(this, preList);
+            this.afterProcess(preList, afterList);
+            if (CollectionUtils.isNotEmpty(afterList)) {
+                feedback(afterList);
+                this.onNext(afterList.get(0).getBatchId());
+            }
+            afterList.forEach(context -> this.emit(context.getData(), context.getSession()));
+        } catch (Exception ex) {
+            LOG.error("Node direct process exception. [streamId={}, nodeId={}, positionId={}, traceId={}, causedBy={}]",
+                    this.streamId, this.id, preList.get(0).getPosition(), preList.get(0).getTraceId(),
+                    ex.getClass().getName());
+            LOG.debug("Node process exception details: ", ex);
+            this.fail(ex, preList);
+        }
+    }
+
     private synchronized void triggerNodeProcessor(ProcessType type) {
         if (type == ProcessType.PRE_PROCESS && (preProcessT == null || !preProcessRunning)) {
             preProcessRunning = true;

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/operators/Operators.java

@@ -292,5 +292,24 @@ public interface Validator<T> {
          */
         List<FlowContext<T>> validate(FlowContextRepo repo, To<T, ?> to);
     }
+
+    /**
+     * Represents a conditional node's transformation from raw material to product.
+     * This functional interface defines the processing operation that converts an
+     * input of one type to an output of potentially different type.
+     *
+     * @param <T> The type of raw material (input) to be processed
+     * @param <R> The type of product (output) to be produced
+     */
+    @FunctionalInterface
+    public interface Then<T, R> {
+        /**
+         * Transforms the input raw material into a processed product.
+         *
+         * @param input The raw material to be processed.
+         * @return The transformed product result.
+         */
+        R process(T input);
+    }
 }
 

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/reactive/Subscriber.java

@@ -32,6 +32,16 @@ public interface Subscriber<I, O> extends StreamIdentity, Emitter<O, FlowSession
      */
     void accept(ProcessType type, List<FlowContext<I>> contexts);
 
+    /**
+     * Processes a batch of flow contexts according to the specified processing type.
+     * This method handles the core execution logic for the workflow engine, applying
+     * the appropriate operations to each context in the batch.
+     *
+     * @param type The type of processing to perform.
+     * @param contexts The list of flow contexts to process.
+     */
+    void process(ProcessType type, List<FlowContext<I>> contexts);
+
     /**
      * 设置节点block
      *

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/reactive/Subscription.java

@@ -27,6 +27,14 @@ public interface Subscription<I> extends StreamIdentity {
      */
     void cache(List<FlowContext<I>> contexts);
 
+    /**
+     * Immediately sends data to subscriber for processing the given flow contexts.
+     * This method executes synchronously and blocks until all contexts have been processed.
+     *
+     * @param contexts The list of flow contexts to process and send to subscribers。
+     */
+    void process(List<FlowContext<I>> contexts);
+
     /**
      * getWhether
      *

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/reactive/When.java

@@ -105,4 +105,18 @@ public void cache(List<FlowContext<I>> contexts) {
     public String getStreamId() {
         return this.streamId;
     }
+
+    @Override
+    public void process(List<FlowContext<I>> contexts) {
+        if (CollectionUtils.isEmpty(contexts)) {
+            return;
+        }
+        List<FlowContext<I>> converted = contexts.stream()
+                .map(context -> context.convertData(context.getData(), context.getId())
+                        .setPosition(this.to.getId())
+                        .setStatus(FlowNodeStatus.READY))
+                .collect(Collectors.toList());
+        repo.updateStatus(converted, FlowNodeStatus.READY.toString(), this.to.getId());
+        messenger.directProcess(this.to.isAuto() ? PROCESS : PRE_PROCESS, this.to, converted);
+    }
 }

framework/waterflow/java/waterflow-core/src/test/java/modelengine/fit/waterflow/domain/WaterFlowsTest.java

@@ -1137,5 +1137,45 @@ void try_a_complicated_map_reduce() throws InterruptedException {
             assertEquals("150unit", result.get(0));
             assertEquals("240unit", result.get(1));
         }
+
+        @Test
+        void shouldExecuteRightBranchWhenUseWhenThenBranchGivenPreserved() {
+            List<Integer> result = new ArrayList<>();
+            ProcessFlow<Integer> flow = Flows.<Integer>create()
+                    .conditions()
+                    .when(i -> i >= 5, i -> i * 10)
+                    .others(i -> i)
+                    .reduce(Integer::sum)
+                    .just(i -> result.add(i))
+                    .close();
+            FlowSession session = new FlowSession(true);
+            Window window = session.begin();
+            flow.offer(new Integer[] {1, 2, 3, 4, 5, 6}, session);
+            window.complete();
+
+            FlowsTestUtil.waitUntil(() -> !result.isEmpty(), 1000);
+            assertEquals(1, result.size());
+            assertEquals(120, result.get(0));
+        }
+
+        @Test
+        void shouldExecuteRightBranchWhenUseWhenThenBranchGivenNotPreserved() {
+            List<Integer> result = new ArrayList<>();
+            ProcessFlow<Integer> flow = Flows.<Integer>create()
+                    .conditions()
+                    .when(i -> i >= 5, i -> i * 10)
+                    .others(i -> i)
+                    .reduce(Integer::sum)
+                    .just(i -> result.add(i))
+                    .close();
+            FlowSession session = new FlowSession(false);
+            Window window = session.begin();
+            flow.offer(new Integer[] {1, 2, 3, 4, 5, 6}, session);
+            window.complete();
+
+            FlowsTestUtil.waitUntil(() -> !result.isEmpty(), 1000);
+            assertEquals(1, result.size());
+            assertEquals(120, result.get(0));
+        }
     }
 }