Merge pull request #600 from swfnotswift/swf

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

Author: loveTsong

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 ? 10 : 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/nodes/FlowConditionNode.java

@@ -80,6 +80,7 @@ public FitStream.Processor<FlowData, FlowData> getProcessor(String streamId, Flo
     protected void subscribe(
             FitStream.Publisher<FlowData> from, FitStream.Subscriber<FlowData, FlowData> to, FlowEvent event) {
         from.subscribe(event.getMetaId(), to, null, this.getWhether(from.getStreamId(), event));
+            to.setAllMode();
     }
 
     private Processors.Whether<FlowData> getWhether(String streamId, FlowEvent event) {

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

@@ -14,6 +14,7 @@
 import modelengine.fit.waterflow.flowsengine.domain.flows.context.repo.flowcontext.FlowContextRepo;
 import modelengine.fit.waterflow.flowsengine.domain.flows.context.repo.flowlock.FlowLocks;
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.FitStream;
+import modelengine.fit.waterflow.flowsengine.domain.flows.streams.FlowDataMerger;
 import modelengine.fit.waterflow.flowsengine.domain.flows.streams.nodes.Node;
 import modelengine.fitframework.log.Logger;
 
@@ -46,7 +47,8 @@ public class FlowForkNode extends FlowNode {
     public FitStream.Processor<FlowData, FlowData> getProcessor(String streamId, FlowContextRepo<FlowData> repo,
             FlowContextMessenger messenger, FlowLocks locks) {
         if (!Optional.ofNullable(processor).isPresent()) {
-            this.processor = new Node<>(streamId, this.metaId, this::forkJuster, repo, messenger, locks, this.type);
+            this.processor = new Node<>(streamId, this.metaId, this::forkJuster, repo, messenger, locks, this.type,
+                    new FlowDataMerger());
             this.processor.onError(errorHandler(streamId));
         }
         return this.processor;