✨ 新功能 实现普通节点的一拖多/多对一的分支并行处理

agent-flow/src/components/base/jadeNode.jsx

@@ -469,7 +469,7 @@ export const jadeNode = (id, x, y, width, height, parent, drawer) => {
    * @returns {number} 连接数。
    */
   self.maxNumToLink = () => {
-    return 1;
+    return self.graph?.connectionLimitDisabled ? 100 : 1;
   };
 
   /**

agent-flow/src/components/base/validator.js

@@ -35,7 +35,9 @@ export class NormalNodeConnectorValidator extends Validator {
   validate() {
     const nextEvents = this.node.getNextRunnableEvents();
     const i18n = this.node.graph.i18n;
-    if (nextEvents.length !== 1) {
+    const isConnectionLimitDisabled = Boolean(this.node.graph?.connectionLimitDisabled);
+    const isValid = isConnectionLimitDisabled ? nextEvents.length >= 1 : nextEvents.length === 1;
+    if (!isValid) {
       return Promise.reject({
         errorFields: [{
           errors: [`${i18n?.t('node') ?? 'node'} ${this.node.text} ${i18n?.t('problemWithConnection') ?? 'problemWithConnection'}`],

agent-flow/src/components/code/codeNodeState.jsx

@@ -93,7 +93,7 @@ export const codeNodeState = (id, x, y, width, height, parent, drawer) => {
      * @override
      */
     self.maxNumToLink = () => {
-        return 10;
+        return self.graph?.connectionLimitDisabled ? 100 : 10;
     };
 
     return self;

agent-flow/src/flow/jadeFlowEntry.jsx

@@ -378,6 +378,10 @@ const jadeFlowAgent = (graph) => {
     graph.destroy();
   };
 
+  self.setConnectionLimitDisabled = (disabled) => {
+    graph.connectionLimitDisabled = Boolean(disabled);
+  };
+
   return self;
 };
 
@@ -432,6 +436,7 @@ export const JadeFlow = (() => {
                        div,
                        tenant,
                        appId,
+                       connectionLimitDisabled = false,
                        flowConfigData,
                        configs,
                        i18n,
@@ -440,7 +445,7 @@ export const JadeFlow = (() => {
                      }) => {
     const graphDom = getGraphDom(div);
     const g = jadeFlowGraph(div, 'jadeFlow');
-    await configGraph(g, tenant, appId, flowConfigData, configs, i18n, importStatements);
+    await configGraph(g, tenant, appId, flowConfigData, configs, i18n, importStatements, connectionLimitDisabled);
     g.flowType = flowType;
     const pageData = g.getPageData(0);
     await g.editFlow(0, graphDom, pageData.id);
@@ -470,8 +475,9 @@ export const JadeFlow = (() => {
     return jadeFlowAgent(g);
   };
 
-  const configGraph = async (g, tenant, appId, flowConfigData, configs, i18n, importStatements) => {
+  const configGraph = async (g, tenant, appId, flowConfigData, configs, i18n, importStatements, connectionLimitDisabled = false) => {
     g.collaboration.mute = true;
+    g.connectionLimitDisabled = Boolean(connectionLimitDisabled);
     g.configs = configs;
     g.i18n = i18n;
     for (let i = 0; i < importStatements.length; i++) {

agent-flow/src/flow/jadeFlowGraph.js

@@ -76,6 +76,7 @@ export const jadeFlowGraph = (div, title) => {
   const self = defaultGraph(div, title);
   self.type = 'jadeFlowGraph';
   self.pageType = 'jadeFlowPage';
+  self.connectionLimitDisabled = false;
   self.enableText = false;
   self.flowMeta = {
     exceptionFitables: ['modelengine.fit.jober.aipp.fitable.AippFlowExceptionHandler'],

agent-flow/src/flow/jadeFlowPage.js

@@ -40,6 +40,8 @@ export const jadeFlowPage = (div, graph, name, id) => {
   self.addEventListener('COPY_SHAPE', shapeChangeListener);
   self.addEventListener('DELETE_SHAPE', shapeChangeListener);
 
+  const isConnectionLimitDisabled = () => Boolean(self.graph?.connectionLimitDisabled);
+
   /**
    * @override
    */
@@ -305,7 +307,7 @@ export const jadeFlowPage = (div, graph, name, id) => {
    */
   self.canDragOut = (node, connector) => {
     const lines = self.getEvents().filter(s => s.fromShape === node.id && s.getDefinedFromConnector() === connector);
-    return lines && lines.length < 1;
+    return lines.length < (isConnectionLimitDisabled() ? 10 : 1);
   };
 
   /**
@@ -330,7 +332,9 @@ export const jadeFlowPage = (div, graph, name, id) => {
       }
     };
 
-    return jadeEvent.fromShape !== jadeEvent.toShape && isConnectorAllowToLink() && isConnectorWithinLimit();
+    return jadeEvent.fromShape !== jadeEvent.toShape
+      && isConnectorAllowToLink()
+      && (isConnectionLimitDisabled() || isConnectorWithinLimit());
   };
 
   /**

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

@@ -265,6 +265,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.copyContextWithoutID(data);
+        context.previous = this.previous;
+        context.nextPositionId = this.nextPositionId;
+        return context;
+    }
+
     /**
      * 用于when.convert数据时候的转换context，除了包裹的数据类型不一样，所有其他信息都一样
      *
@@ -274,12 +297,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.copyContextWithoutID(data);
+        context.previous = this.previous;
+        context.id = id;
+        return context;
+    }
+
+    private <R> FlowContext<R> copyContextWithoutID(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/context/repo/flowcontext/FlowContextRepo.java

@@ -92,16 +92,16 @@ default void update(List<FlowContext<T>> contexts) {
     }
 
     /**
-     * updateToSent
+     * 更新context状态为SENT
      *
-     * @param contexts contexts
+     * @param contexts 上下文列表
      */
     void updateToSent(List<FlowContext<T>> contexts);
 
     /**
-     * updateToReady
+     * 更新context状态为READY
      *
-     * @param contexts contexts
+     * @param contexts 上下文列表
      */
     void updateToReady(List<FlowContext<T>> contexts);
 

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

@@ -179,6 +179,7 @@ public FitStream.Publisher<FlowData> convertToFlow(FlowContextRepo<FlowData> rep
                     //  startNode不能出现在event的to属性, endNode不能出现在event的from属性
                     FlowNode toNode = nodeMap.get(event.getTo());
                     fromNode.subscribe(streamId, flowEnv, toNode, event);
+
                 });
             });
             return getFlowNode(FlowNodeType.START).getPublisher(streamId, repo, messenger, locks);

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

@@ -20,8 +20,13 @@
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.FitStream;
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.nodes.Blocks;
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.nodes.Node;
+import modelengine.fit.waterflow.flowsengine.utils.FlowUtil;
 import modelengine.fitframework.log.Logger;
+import modelengine.fitframework.util.CollectionUtils;
+import modelengine.fitframework.util.ObjectUtils;
+import modelengine.fitframework.util.StringUtils;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -54,10 +59,11 @@ 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);
+
             if (!Objects.isNull(this.jober)) {
                 node.setIsAsyncJob(this.jober.isAsync());
             }
@@ -74,6 +80,50 @@ public FitStream.Processor<FlowData, FlowData> getProcessor(String streamId, Flo
         return this.processor;
     }
 
+    private List<FlowContext<FlowData>> mergeProcessInputs(List<FlowContext<FlowData>> pre) {
+        if (CollectionUtils.isEmpty(pre) || pre.size() <= 1) {
+            return pre;
+        }
+        if (pre.stream().anyMatch(context -> !(context.getData() instanceof FlowData))) {
+            return pre;
+        }
+        if (pre.stream().map(FlowContext::getPosition).filter(StringUtils::isNotEmpty).distinct().count() <= 1) {
+            return pre;
+        }
+        FlowContext<FlowData> baseContext = pre.get(0);
+        FlowData mergedFlowData = mergeFlowData(pre, baseContext.getId());
+        return Collections.singletonList(
+                baseContext.convertData(ObjectUtils.cast(mergedFlowData), baseContext.getId()));
+    }
+
+    private FlowData mergeFlowData(List<FlowContext<FlowData>> pre, String baseContextId) {
+        FlowData first = pre.get(0).getData();
+        Map<String, Object> businessData = new HashMap<>(
+                Optional.ofNullable(first.getBusinessData()).orElseGet(HashMap::new));
+        Map<String, Object> contextData = new HashMap<>(
+                Optional.ofNullable(first.getContextData()).orElseGet(HashMap::new));
+        Map<String, Object> passData = new HashMap<>(Optional.ofNullable(first.getPassData()).orElseGet(HashMap::new));
+
+        pre.stream().skip(1).map(FlowContext::getData).forEach(flowData -> {
+            businessData.putAll(FlowUtil.mergeMaps(businessData,
+                    Optional.ofNullable(flowData.getBusinessData()).orElseGet(HashMap::new)));
+            contextData.putAll(FlowUtil.mergeMaps(contextData,
+                    Optional.ofNullable(flowData.getContextData()).orElseGet(HashMap::new)));
+            passData.putAll(FlowUtil.mergeMaps(passData,
+                    Optional.ofNullable(flowData.getPassData()).orElseGet(HashMap::new)));
+        });
+            contextData.put(Constant.CONTEXT_ID, baseContextId);
+        return FlowData.builder()
+                .operator(first.getOperator())
+                .startTime(first.getStartTime())
+                .businessData(businessData)
+                .contextData(contextData)
+                .passData(passData)
+                .errorMessage(first.getErrorMessage())
+                .errorInfo(first.getErrorInfo())
+                .build();
+    }
+
     private List<FlowData> stateProduce(List<FlowContext<FlowData>> inputs) {
         addContextData(inputs);
         List<FlowData> flowDataList = inputs.stream().map(FlowContext::getData).collect(Collectors.toList());

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,55 @@ 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.forEach(FitStream.Subscription::cache);
     }
 
+    // 依赖约束：preSendCallback 仅做只读回调，不改写 matchedContexts，因此 forkedContexts 可直接持久化。
+    private void persistForkedContexts(List<FlowContext<I>> forkedContexts) {
+        if (CollectionUtils.isEmpty(forkedContexts)) {
+            return;
+        }
+        Set<String> traces = forkedContexts.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(forkedContexts, traces);
+            this.repo.save(forkedContexts);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     /**
      * 是否有publisher目标
      * 用于stream闭环时将没有subscribed的publisher关闭到close subscriber