[fel] adapter flow case

Author: songyongtan

framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiWhenHappen.java

@@ -53,7 +53,7 @@ public AiWhenHappen<O, D, I, RF, F> when(Operators.Whether<I> whether, Operators
      *
      * @param processor The handler to process unmatched inputs.
      * @return An {@link AiState} representing the terminal node of the conditional flow.
-     * @throws IllegalArgumentException if processor is null
+     * @throws IllegalArgumentException if processor is null.
      */
     public AiState<O, D, O, RF, F> others(Operators.Then<I, O> processor) {
         Validation.notNull(processor, "Ai branch processor cannot be null.");

framework/fel/java/fel-flow/src/test/java/modelengine/fel/engine/AiFlowCaseTest.java

@@ -34,18 +34,22 @@
  * @since 2025-06-11
  */
 public class AiFlowCaseTest {
+    private static final int SPEED = 1;
     @Nested
     class DesensitizeCase {
         private final ChatFlowModel model = new ChatFlowModel((prompt, chatOption) -> Choir.create(emitter -> {
             emitter.emit(new AiMessage("<think>"));
-            for (int i = 0; i < 10; i++) {
+            int takeTime = 10 * SPEED;
+            SleepUtil.sleep(takeTime);
+            for (int i = 0; i < 48; i++) {
                 emitter.emit(new AiMessage(String.valueOf(i)));
-                SleepUtil.sleep(100);
+                SleepUtil.sleep(takeTime);
             }
             emitter.emit(new AiMessage("</think>"));
-            for (int i = 100; i < 110; i++) {
+            SleepUtil.sleep(takeTime);
+            for (int i = 100; i < 150; i++) {
                 emitter.emit(new AiMessage(String.valueOf(i)));
-                SleepUtil.sleep(100);
+                SleepUtil.sleep(takeTime);
             }
             emitter.complete();
         }), ChatOption.custom().model("modelName").stream(true).build());
@@ -60,7 +64,8 @@ class DesensitizeCase {
                     this.log(input);
                     return input;
                 })
-                .map(this::mockDesensitize)
+                .map(this::mockDesensitize1)
+                .map(this::mockDesensitize2)
                 .close();
 
         @Test
@@ -74,7 +79,7 @@ void run() {
             }).offer(Tip.fromArray("hi"));
             result.await();
             System.out.printf("time:%s, cost=%s\n", System.currentTimeMillis(), System.currentTimeMillis() - startTime);
-            Assertions.assertEquals(22, counter.get());
+            Assertions.assertEquals(100, counter.get());
         }
 
         private Chunk classic(ChatMessage message, StateContext ctx) {
@@ -92,10 +97,16 @@ private Chunk classic(ChatMessage message, StateContext ctx) {
             return new Chunk(false, message.text());
         }
 
-        private String mockDesensitize(Chunk chunk) {
+        private String mockDesensitize1(Chunk chunk) {
+            SleepUtil.sleep(10 * SPEED);
             return chunk.content.replace("3", "*");
         }
 
+        private String mockDesensitize2(String chunk) {
+            SleepUtil.sleep(10 * SPEED);
+            return chunk.replace("4", "*");
+        }
+
         private void log(Chunk chunk) {
             System.out.println("log content:" + chunk.content);
         }
@@ -113,17 +124,17 @@ private static class Chunk {
     /**
      * Simulates a backpressure scenario where:
      * <ol>
-     *     <li>The LLM generates data (50ms per item) faster than the TTS can process it.</li>
+     *     <li>The LLM generates data faster than the TTS can process it.</li>
      *     <li>TTS processing is constrained to a single thread.</li>
-     *     <li>TTS processing speed is artificially slowed (100ms per item).</li>
+     *     <li>TTS processing speed is artificially slowed.</li>
      * </ol>
      */
     @Nested
     class BackPressureCase {
         private final ChatFlowModel model = new ChatFlowModel((prompt, chatOption) -> Choir.create(emitter -> {
-            for (int i = 0; i < 10; i++) {
+            for (int i = 0; i < 100; i++) {
                 emitter.emit(new AiMessage(String.valueOf(i)));
-                SleepUtil.sleep(50);
+                SleepUtil.sleep(5 * SPEED);
             }
             emitter.complete();
             System.out.printf("time:%s, generate completed.\n", System.currentTimeMillis());
@@ -132,6 +143,7 @@ class BackPressureCase {
         private final AiProcessFlow<Tip, String> flow = AiFlows.<Tip>create()
                 .prompt(Prompts.human("{{0}}"))
                 .generate(model)
+                .map(this::mockDesensitize).concurrency(1) // Limit processing to 1 concurrent thread
                 .map(this::mockTTS).concurrency(1) // Limit processing to 1 concurrent thread
                 .close();
 
@@ -146,30 +158,36 @@ void run() {
             }).offer(Tip.fromArray("hi"));
             result.await();
             System.out.printf("time:%s, cost=%s\n", System.currentTimeMillis(), System.currentTimeMillis() - startTime);
-            Assertions.assertEquals(10, counter.get());
+            Assertions.assertEquals(100, counter.get());
+        }
+
+        private String mockDesensitize(ChatMessage chunk) {
+            // Simulate time-consuming operation with a delay.
+            SleepUtil.sleep(10 * SPEED);
+            return chunk.text().replace("3", "*");
         }
 
-        private String mockTTS(ChatMessage chunk) {
+        private String mockTTS(String chunk) {
             // Simulate time-consuming operation with a delay.
-            SleepUtil.sleep(100);
-            return chunk.text();
+            SleepUtil.sleep(10 * SPEED);
+            return chunk;
         }
     }
 
     /**
      * Demonstrates concurrent processing with balanced throughput where:
      * <ol>
-     *     <li>LLM generates data at moderate pace (50ms per item)</li>
-     *     <li>Downstream processing runs with 3 concurrent threads</li>
-     *     <li>Processing speed is slightly slower than generation (150ms vs 50ms)</li>
+     *     <li>LLM generates data at moderate pace.</li>
+     *     <li>Downstream processing runs with 3 concurrent threads.</li>
+     *     <li>Processing speed is slightly slower than generation (3 : 1).</li>
      * </ol>
      */
     @Nested
     class ConcurrencyCase {
         private final ChatFlowModel model = new ChatFlowModel((prompt, chatOption) -> Choir.create(emitter -> {
-            for (int i = 0; i < 10; i++) {
+            for (int i = 0; i < 100; i++) {
                 emitter.emit(new AiMessage(String.valueOf(i)));
-                SleepUtil.sleep(50);
+                SleepUtil.sleep(10 * SPEED);
             }
             emitter.complete();
         }), ChatOption.custom().model("modelName").stream(true).build());
@@ -191,12 +209,12 @@ void run() {
             }).offer(Tip.fromArray("hi"));
             result.await();
             System.out.printf("time:%s, cost=%s\n", System.currentTimeMillis(), System.currentTimeMillis() - startTime);
-            Assertions.assertEquals(10, counter.get());
+            Assertions.assertEquals(100, counter.get());
         }
 
         private String mockDesensitize(ChatMessage chunk) {
             // Simulate slower processing at 1/3 speed of LLM generation.
-            SleepUtil.sleep(150);
+            SleepUtil.sleep(30 * SPEED);
             return chunk.text().replace("3", "*");
         }
     }

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

@@ -52,7 +52,7 @@ public WhenHappen<O, D, I, F> when(Operators.Whether<I> whether, Operators.Then<
      *
      * @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
+     * @throws IllegalArgumentException if processor is null.
      */
     public State<O, D, O, F> others(Operators.Then<I, O> processor) {
         this.when(null, processor);

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

@@ -100,8 +100,6 @@ public class To<I, O> extends IdGenerator implements Subscriber<I, O> {
     @Getter
     private final FlowLocks locks;
 
-    private int maxConcurrency = MAX_CONCURRENCY;
-
     // 默认自动流转过滤器是按batchID批次过滤contexts
     private final Operators.Filter<I> defaultAutoFilter = (contexts) -> {
         if (CollectionUtils.isEmpty(contexts)) {
@@ -140,7 +138,7 @@ public class To<I, O> extends IdGenerator implements Subscriber<I, O> {
     @Getter
     private ProcessMode processMode;
 
-    private final Map<String, Integer> processingSessions = new ConcurrentHashMap<>();
+    private 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()),
@@ -162,6 +160,7 @@ public class To<I, O> extends IdGenerator implements Subscriber<I, O> {
      */
     private Operators.Produce<FlowContext<I>, O> produce;
 
+    private volatile int maxConcurrency = MAX_CONCURRENCY;
     /**
      * 当前并发度，已经提交的批次
      */

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

@@ -298,8 +298,8 @@ public interface Validator<T> {
      * 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
+     * @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> {