修改后端代码，实现条件流程校验阶段对多节点的支持

Author: qq_62395577

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java

@@ -26,6 +26,7 @@ public final class FlowContext<T> extends IdGenerator {
     /**
      * 通过from.offer(data)而不是.offer(context)发起的数据会新增一个trace，这个trace会延续到flow end
      */
+
     @Getter
     private final Set<String> traceId;
 
@@ -265,6 +266,29 @@ public <R> List<FlowContext<R>> generate(List<R> data, String position) {
         return data.stream().map(d -> this.generate(d, position, LocalDateTime.now())).collect(Collectors.toList());
     }
 
+    /**
+     * fork一个新的context用于一拖多分支，继承当前context的运行元数据，但生成新的contextId。
+     *
+     * @return 新的分支context
+     */
+    public FlowContext<T> fork() {
+        return this.convertData(this.data);
+    }
+
+    /**
+     * convertData
+     *
+     * @param <R> 转换后的数据类型
+     * @param data 转换后的数据
+     * @return 转换后的context
+     */
+    public <R> FlowContext<R> convertData(R data) {
+        FlowContext<R> context = this.copyContext(data);
+        context.previous = this.id;
+        context.nextPositionId = this.nextPositionId;
+        return context;
+    }
+
     /**
      * 用于when.convert数据时候的转换context，除了包裹的数据类型不一样，所有其他信息都一样
      *
@@ -274,12 +298,17 @@ public <R> List<FlowContext<R>> generate(List<R> data, String position) {
      * @return 转换后的context
      */
     public <R> FlowContext<R> convertData(R data, String id) {
+        FlowContext<R> context = this.copyContext(data);
+        context.previous = this.previous;
+        context.id = id;
+        return context;
+    }
+
+    private <R> FlowContext<R> copyContext(R data) {
         FlowContext<R> context = new FlowContext<>(this.streamId, this.rootId, data, this.traceId, this.position,
                 this.parallel, this.parallelMode, LocalDateTime.now());
-        context.previous = this.previous;
         context.status = this.status;
         context.trans = this.trans;
-        context.id = id;
         context.batchId = this.batchId;
         context.toBatch = this.toBatch;
         context.createAt = this.createAt;

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowStateNode.java

@@ -54,7 +54,7 @@ public class FlowStateNode extends FlowNode {
      */
     @Override
     public FitStream.Processor<FlowData, FlowData> getProcessor(String streamId, FlowContextRepo<FlowData> repo,
-            FlowContextMessenger messenger, FlowLocks locks) {
+                                                                FlowContextMessenger messenger, FlowLocks locks) {
         if (!Optional.ofNullable(this.processor).isPresent()) {
             Node<FlowData, FlowData> node = new Node<>(streamId, this.metaId, this::stateProduce, repo, messenger,
                     locks, this.type);

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/From.java

@@ -42,6 +42,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.locks.Lock;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -322,26 +323,64 @@ public void offer(List<FlowContext<I>> contexts, Consumer<PreSendCallbackInfo<I>
         // qualifiedWhens表示的与from节点连接的所有事件，条件节点符合条件的事件在这里筛选，在事件上处理需要下发的context
         java.util.Map<FitStream.Subscription<I, ?>, List<FlowContext<I>>> matchedContexts = new LinkedHashMap<>();
         Set<FlowContext<I>> matchedContextSet = new HashSet<>();
-        qualifiedWhens.forEach(
-                w -> {
-                    List<FlowContext<I>> afterContexts = contexts
-                            .stream()
-                            .filter(c -> w.getWhether().is(c))
-                            .peek(c -> c.setNextPositionId(w.getId()))
-                            .collect(Collectors.toList());
-                    matchedContexts.put(w, afterContexts);
-                    matchedContextSet.addAll(afterContexts);
+        List<FlowContext<I>> forkedContexts = new ArrayList<>();
+        for (FlowContext<I> contextItem : contexts) {
+            List<FitStream.Subscription<I, ?>> matchedSubscriptions = qualifiedWhens.stream()
+                    .filter(w -> w.getWhether().is(contextItem))
+                    .collect(Collectors.toList());
+            if (CollectionUtils.isEmpty(matchedSubscriptions)) {
+                continue;
+            }
+            matchedContextSet.add(contextItem);
+            for (int index = 0; index < matchedSubscriptions.size(); index++) {
+                FitStream.Subscription<I, ?> matchedSubscription = matchedSubscriptions.get(index);
+                FlowContext<I> branchContext = index == 0 ? contextItem : contextItem.fork();
+                branchContext.setNextPositionId(matchedSubscription.getId());
+                matchedContexts.computeIfAbsent(matchedSubscription, key -> new ArrayList<>()).add(branchContext);
+                if (index > 0) {
+                    forkedContexts.add(branchContext);
                 }
-        );
+            }
+        }
+        qualifiedWhens.forEach(w -> matchedContexts.computeIfAbsent(w, key -> new ArrayList<>()));
         List<FlowContext<I>> unMatchedContexts = contexts
                 .stream()
                 .filter(c -> !matchedContextSet.contains(c))
                 .collect(Collectors.toList());
         PreSendCallbackInfo<I> callbackInfo = new PreSendCallbackInfo<>(matchedContexts, unMatchedContexts);
         preSendCallback.accept(callbackInfo);
+        persistForkedContexts(forkedContexts, matchedContexts);
         matchedContexts.forEach(FitStream.Subscription::cache);
     }
 
+    private void persistForkedContexts(List<FlowContext<I>> forkedContexts,
+            java.util.Map<FitStream.Subscription<I, ?>, List<FlowContext<I>>> matchedContexts) {
+        if (CollectionUtils.isEmpty(forkedContexts)) {
+            return;
+        }
+        Set<String> forkedIds = forkedContexts.stream().map(FlowContext::getId).collect(Collectors.toSet());
+        List<FlowContext<I>> effectiveForkedContexts = matchedContexts.values()
+                .stream()
+                .flatMap(List::stream)
+                .filter(context -> forkedIds.contains(context.getId()))
+                .collect(Collectors.toList());
+        if (CollectionUtils.isEmpty(effectiveForkedContexts)) {
+            return;
+        }
+        Set<String> traces = effectiveForkedContexts.stream()
+                .flatMap(context -> context.getTraceId().stream())
+                .collect(Collectors.toSet());
+        Lock lock = this.locks.getDistributedLock(this.locks.streamNodeLockKey(this.streamId, this.id,
+                "ForkContextPool"));
+        lock.lock();
+        try {
+            this.repo.updateContextPool(effectiveForkedContexts, traces);
+            this.repo.save(effectiveForkedContexts);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     /**
      * 是否有publisher目标
      * 用于stream闭环时将没有subscribed的publisher关闭到close subscriber