From cf917b5988ec3cd6e42759a920ed22a4cad52aab Mon Sep 17 00:00:00 2001 From: PIGCLOUD Date: Fri, 16 Oct 2026 12:32:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(flow):=20=E6=96=B0=E5=A2=9E=E6=B5=81?= =?UTF-8?q?=E7=A8=8B=E5=9B=9E=E8=B0=83=E6=8A=95=E9=80=92=E3=80=81=E8=8D=89?= =?UTF-8?q?=E7=A8=BF=E5=AD=98=E5=82=A8=E5=8F=8A=E6=B5=81=E7=A8=8B=E5=8F=AF?= =?UTF-8?q?=E9=9D=A0=E6=80=A7=E6=9C=8D=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 FlowCallbackDelivery 类，实现 HTTP 回调的至少一次投递机制 - 新增 FlowDraftService 类，支持个人流程草稿的保存、读取和删除 - 新增 FlowReliabilityService 类，处理流程通知、催办与回调的持久化 - 删除 FlowWorkspaceTests 测试类，整合测试逻辑至其他模块 (cherry picked from commit 07c8ccc0e845d30dcfaeeb6d01c999460093c8ac) --- .../pigx/flow/constant/IBaseEnum.java | 113 --- .../pigx/flow/dto/NodeHttpResultVO.java | 40 - .../flow/config/FlowDurableConfiguration.java | 4 +- .../controller/FlowWorkspaceController.java | 4 +- .../listeners/FlowDurableEventListener.java | 2 +- .../flow/support/utils/FlowableUtils.java | 752 ------------------ .../workspace}/FlowCallbackDelivery.java | 2 +- .../workspace}/FlowDraftService.java | 2 +- .../workspace}/FlowReliabilityService.java | 4 +- .../pigx/flow/service/FlowWorkspaceTests.java | 297 ------- 10 files changed, 10 insertions(+), 1210 deletions(-) delete mode 100644 pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/IBaseEnum.java delete mode 100644 pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/NodeHttpResultVO.java delete mode 100644 pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/FlowableUtils.java rename pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/{service => support/workspace}/FlowCallbackDelivery.java (98%) rename pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/{service => support/workspace}/FlowDraftService.java (99%) rename pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/{service => support/workspace}/FlowReliabilityService.java (99%) delete mode 100644 pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/IBaseEnum.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/IBaseEnum.java deleted file mode 100644 index 777e122c2..000000000 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/IBaseEnum.java +++ /dev/null @@ -1,113 +0,0 @@ -package com.pig4cloud.pigx.flow.constant; - -import cn.hutool.core.util.ObjectUtil; - -import java.util.EnumSet; -import java.util.Objects; - -/** - * 枚举通用接口 - *

- * 为所有业务枚举提供统一的接口规范，实现枚举值和标签的标准化管理。 - * 提供了根据值获取枚举、根据值获取标签、根据标签获取值等通用方法。 - * 所有业务枚举类都应实现此接口，以保证枚举使用的一致性。 - *

- * - * @param 枚举值的类型，通常为Integer或String - * @author haoxr - * @since 2022/3/27 12:06 - */ -public interface IBaseEnum { - - /** - * 获取枚举值 - *

- * 返回枚举对应的实际值，通常用于数据库存储或接口传输 - *

- * - * @return 枚举对应的值 - */ - T getValue(); - - /** - * 获取枚举标签 - *

- * 返回枚举对应的显示标签，通常用于界面展示 - *

- * - * @return 枚举对应的中文标签 - */ - String getLabel(); - - /** - * 根据值获取对应的枚举实例 - *

- * 通过枚举值查找对应的枚举实例，常用于数据库查询结果的转换。 - * 如果找不到匹配的枚举，返回null。 - *

- * - * @param value 枚举值 - * @param clazz 枚举类的Class对象 - * @param 枚举类型，必须实现IBaseEnum接口 - * @return 匹配的枚举实例，如果没有找到返回null - */ - static & IBaseEnum> E getEnumByValue(Object value, Class clazz) { - Objects.requireNonNull(value); - EnumSet allEnums = EnumSet.allOf(clazz); // 获取类型下的所有枚举 - E matchEnum = allEnums.stream().filter(e -> ObjectUtil.equal(e.getValue(), value)).findFirst().orElse(null); - return matchEnum; - } - - /** - * 根据值获取对应的标签 - *

- * 通过枚举值直接获取对应的显示标签，常用于列表展示时的值转换。 - * 如果找不到匹配的枚举，返回null。 - *

- * - * @param value 枚举值 - * @param clazz 枚举类的Class对象 - * @param 枚举类型，必须实现IBaseEnum接口 - * @return 枚举对应的标签，如果没有找到返回null - */ - static & IBaseEnum> String getLabelByValue(Object value, Class clazz) { - Objects.requireNonNull(value); - EnumSet allEnums = EnumSet.allOf(clazz); // 获取类型下的所有枚举 - E matchEnum = allEnums.stream().filter(e -> ObjectUtil.equal(e.getValue(), value)).findFirst().orElse(null); - - String label = null; - if (matchEnum != null) { - label = matchEnum.getLabel(); - } - return label; - } - - /** - * 根据标签获取对应的值 - *

- * 通过枚举标签反向查找对应的枚举值，常用于表单提交时的数据转换。 - * 如果找不到匹配的枚举，返回null。 - *

- * - * @param label 枚举标签 - * @param clazz 枚举类的Class对象 - * @param 枚举类型，必须实现IBaseEnum接口 - * @return 枚举对应的值，如果没有找到返回null - */ - static & IBaseEnum> Object getValueByLabel(String label, Class clazz) { - Objects.requireNonNull(label); - EnumSet allEnums = EnumSet.allOf(clazz); // 获取类型下的所有枚举 - String finalLabel = label; - E matchEnum = allEnums.stream() - .filter(e -> ObjectUtil.equal(e.getLabel(), finalLabel)) - .findFirst() - .orElse(null); - - Object value = null; - if (matchEnum != null) { - value = matchEnum.getValue(); - } - return value; - } - -} diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/NodeHttpResultVO.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/NodeHttpResultVO.java deleted file mode 100644 index cd770ec56..000000000 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/NodeHttpResultVO.java +++ /dev/null @@ -1,40 +0,0 @@ -package com.pig4cloud.pigx.flow.dto; - -import lombok.Data; - -import java.util.Map; - -/** - * 节点HTTP请求结果数据传输对象 - *

- * 该DTO用于封装流程节点执行HTTP请求后的返回结果。 - * 在流程执行过程中，某些节点可能需要调用外部系统的接口（如获取审批人、 - * 验证业务数据等），此对象用于统一封装这些HTTP请求的响应结果。 - *

- * - * @author pigx - */ -@Data -public class NodeHttpResultVO { - - /** - * 请求是否成功 - *

- * 标识HTTP请求是否执行成功： - * - true: 请求成功，data字段包含返回的业务数据 - * - false: 请求失败，data字段可能包含错误信息 - *

- */ - private Boolean ok; - - /** - * 返回的数据内容 - *

- * HTTP请求返回的业务数据，以Map形式存储，便于灵活处理各种数据结构。 - * 成功时包含业务数据，失败时可能包含错误码、错误信息等。 - * 具体内容取决于调用的接口定义。 - *

- */ - private Map data; - -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java index 458365a67..af3b56c3c 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java @@ -1,8 +1,8 @@ package com.pig4cloud.pigx.flow.config; -import com.pig4cloud.pigx.flow.service.FlowCallbackDelivery; +import com.pig4cloud.pigx.flow.support.workspace.FlowCallbackDelivery; import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum; -import com.pig4cloud.pigx.flow.service.FlowReliabilityService; +import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService; import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener; import com.pig4cloud.pigx.flow.constant.FlowCallbackStatusEnum; import lombok.RequiredArgsConstructor; diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java index 1e81143f6..8eca99fe1 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java @@ -3,8 +3,8 @@ package com.pig4cloud.pigx.flow.controller; import com.pig4cloud.pigx.common.core.util.R; import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder; import com.pig4cloud.pigx.common.security.util.SecurityUtils; -import com.pig4cloud.pigx.flow.service.FlowDraftService; -import com.pig4cloud.pigx.flow.service.FlowReliabilityService; +import com.pig4cloud.pigx.flow.support.workspace.FlowDraftService; +import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService; import com.pig4cloud.pigx.common.log.annotation.SysLog; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java index a65503b4f..275e705c4 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java @@ -1,6 +1,6 @@ package com.pig4cloud.pigx.flow.support.listeners; -import com.pig4cloud.pigx.flow.service.FlowReliabilityService; +import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService; import lombok.RequiredArgsConstructor; import org.flowable.common.engine.api.delegate.event.FlowableEntityEvent; import org.flowable.common.engine.api.delegate.event.FlowableEvent; diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/FlowableUtils.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/FlowableUtils.java deleted file mode 100644 index 1240151c9..000000000 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/FlowableUtils.java +++ /dev/null @@ -1,752 +0,0 @@ -package com.pig4cloud.pigx.flow.support.utils; - -import cn.hutool.core.collection.CollUtil; -import cn.hutool.core.lang.Dict; -import com.pig4cloud.pigx.common.security.util.SecurityUtils; -import com.pig4cloud.pigx.flow.constant.NodeUserTypeEnum; -import com.pig4cloud.pigx.flow.dto.BizFlowBaseDto; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; -import org.flowable.bpmn.model.*; -import org.flowable.engine.history.HistoricActivityInstance; -import org.flowable.engine.impl.bpmn.behavior.ParallelMultiInstanceBehavior; -import org.flowable.engine.impl.bpmn.behavior.SequentialMultiInstanceBehavior; - -import java.util.*; - -/** - * Flowable流程引擎工具类 - *

- * 提供流程模型遍历、节点查找、路径分析等功能的工具类。 - * 该类是流程引擎操作的核心工具，支持复杂的流程拓扑结构分析。 - *

- * 主要功能： - * 1. 流程元素的入口/出口连线获取 - * 2. 递归遍历流程定义获取所有元素 - * 3. 向前/向后查找用户任务节点 - * 4. 脏数据线路分析（用于回退场景） - * 5. 串行/并行关系判断 - * 6. 历史数据清洗 - *

- * 支持的流程结构： - * - 顺序流程 - * - 并行网关 - * - 排他网关 - * - 子流程 - * - 多实例任务（会签） - * - * @author linjinp - * @create 2019-12-24 13:51 - */ -public class FlowableUtils { - - public static final Logger logger = LogManager.getLogger(FlowableUtils.class); - - /** - * 根据节点获取入口连线集合 - *

- * 支持多种流程元素类型，包括任务、网关、子流程、开始/结束事件等。 - * 入口连线是指向该节点的所有连线。 - * - * @param source 流程元素节点 - * @return 入口连线集合，如果节点类型不支持则返回null - */ - public static List getElementIncomingFlows(FlowElement source) { - List sequenceFlows = null; - if (source instanceof Task) { - sequenceFlows = ((Task) source).getIncomingFlows(); - } else if (source instanceof Gateway) { - sequenceFlows = ((Gateway) source).getIncomingFlows(); - } else if (source instanceof SubProcess) { - sequenceFlows = ((SubProcess) source).getIncomingFlows(); - } else if (source instanceof StartEvent) { - sequenceFlows = ((StartEvent) source).getIncomingFlows(); - } else if (source instanceof EndEvent) { - sequenceFlows = ((EndEvent) source).getIncomingFlows(); - } - return sequenceFlows; - } - - /** - * 根据节点获取出口连线集合 - *

- * 支持多种流程元素类型，包括任务、网关、子流程、开始/结束事件等。 - * 出口连线是从该节点出发的所有连线。 - * - * @param source 流程元素节点 - * @return 出口连线集合，如果节点类型不支持则返回null - */ - public static List getElementOutgoingFlows(FlowElement source) { - List sequenceFlows = null; - if (source instanceof Task) { - sequenceFlows = ((Task) source).getOutgoingFlows(); - } else if (source instanceof Gateway) { - sequenceFlows = ((Gateway) source).getOutgoingFlows(); - } else if (source instanceof SubProcess) { - sequenceFlows = ((SubProcess) source).getOutgoingFlows(); - } else if (source instanceof StartEvent) { - sequenceFlows = ((StartEvent) source).getOutgoingFlows(); - } else if (source instanceof EndEvent) { - sequenceFlows = ((EndEvent) source).getOutgoingFlows(); - } - return sequenceFlows; - } - - /** - * 递归获取流程定义所有的流程元素 - *

- * 深度优先遍历流程定义，包括子流程中的所有元素。 - * 该方法常用于分析整个流程结构或查找特定元素。 - * - * @param flowElements 当前级别流程元素集合 - * @param allElements 用于存储的全部流程元素集合，可为null - * @return 全部流程元素集合，包含所有层级的元素 - */ - public static Collection getAllElements(Collection flowElements, - Collection allElements) { - allElements = allElements == null ? new ArrayList<>() : allElements; - - for (FlowElement flowElement : flowElements) { - allElements.add(flowElement); - if (flowElement instanceof SubProcess) { - // 继续深入子流程，进一步获取子流程 - allElements = FlowableUtils.getAllElements(((SubProcess) flowElement).getFlowElements(), allElements); - } - } - return allElements; - } - - /** - * 从后向前递归获取父级用户任务节点 - *

- * 从当前节点开始，沿着入口连线向前查找所有的用户任务节点。 - * 主要用于流程回退、驳回等场景，找到可以回退的目标节点。 - * 支持子流程的遍历。 - * - * @param source 当前节点 - * @param hasSequenceFlow 已处理过的顺序流集合，避免循环，可为null - * @param userTaskList 递归获取的用户任务节点集合，可为null - * @return 用户任务节点集合 - */ - public static List iteratorFindParentUserTasks(FlowElement source, Set hasSequenceFlow, - List userTaskList) { - userTaskList = userTaskList == null ? new ArrayList<>() : userTaskList; - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - - // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代 - if (source instanceof StartEvent && source.getSubProcess() != null) { - userTaskList = iteratorFindParentUserTasks(source.getSubProcess(), hasSequenceFlow, userTaskList); - } - - // 根据类型，获取入口连线 - List sequenceFlows = getElementIncomingFlows(source); - - if (sequenceFlows != null) { - // 循环找到目标元素 - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 类型为用户节点，则新增父级节点 - if (sequenceFlow.getSourceFlowElement() instanceof UserTask) { - userTaskList.add((UserTask) sequenceFlow.getSourceFlowElement()); - continue; - } - // 类型为子流程，则添加子流程开始节点出口处相连的节点 - if (sequenceFlow.getSourceFlowElement() instanceof SubProcess) { - // 获取子流程用户任务节点 - List childUserTaskList = findChildProcessUserTasks( - (StartEvent) ((SubProcess) sequenceFlow.getSourceFlowElement()).getFlowElements() - .toArray()[0], - null, null); - // 如果找到节点，则说明该线路找到节点，不继续向下找，反之继续 - if (childUserTaskList != null && childUserTaskList.size() > 0) { - userTaskList.addAll(childUserTaskList); - continue; - } - } - // 继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - userTaskList = iteratorFindParentUserTasks(sequenceFlow.getSourceFlowElement(), - new HashSet<>(hasSequenceFlow), userTaskList); - } - } - return userTaskList; - } - - /** - * 根据正在运行的任务节点，迭代获取子级任务节点列表 - *

- * 从当前节点开始，沿着出口连线向后查找正在运行的任务节点。 - * 主要用于找到需要撤回或影响的下游节点。 - * - * @param source 起始节点 - * @param runActiveIdList 正在运行的任务Key，用于校验任务节点是否是正在运行的节点 - * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复 - * @param flowElementList 需要撤回的用户任务列表 - * @return 找到的流程元素列表 - */ - public static List iteratorFindChildUserTasks(FlowElement source, List runActiveIdList, - Set hasSequenceFlow, List flowElementList) { - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - flowElementList = flowElementList == null ? new ArrayList<>() : flowElementList; - - // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代 - if (source instanceof EndEvent && source.getSubProcess() != null) { - flowElementList = iteratorFindChildUserTasks(source.getSubProcess(), runActiveIdList, hasSequenceFlow, - flowElementList); - } - - // 根据类型，获取出口连线 - List sequenceFlows = getElementOutgoingFlows(source); - - if (sequenceFlows != null) { - // 循环找到目标元素 - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 如果为用户任务类型，或者为网关 - // 活动节点ID 在运行的任务中存在，添加 - if ((sequenceFlow.getTargetFlowElement() instanceof UserTask - || sequenceFlow.getTargetFlowElement() instanceof Gateway) - && runActiveIdList.contains((sequenceFlow.getTargetFlowElement()).getId())) { - flowElementList.add(sequenceFlow.getTargetFlowElement()); - continue; - } - // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取 - if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) { - List childUserTaskList = iteratorFindChildUserTasks( - (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements() - .toArray()[0]), - runActiveIdList, hasSequenceFlow, null); - // 如果找到节点，则说明该线路找到节点，不继续向下找，反之继续 - if (childUserTaskList != null && childUserTaskList.size() > 0) { - flowElementList.addAll(childUserTaskList); - continue; - } - } - // 继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - flowElementList = iteratorFindChildUserTasks(sequenceFlow.getTargetFlowElement(), runActiveIdList, - new HashSet<>(hasSequenceFlow), flowElementList); - } - } - return flowElementList; - } - - /** - * 迭代获取子流程用户任务节点 - *

- * 专门用于处理子流程内部的用户任务节点查找。 - * 从子流程的起始节点开始，找到所有的用户任务节点。 - * - * @param source 起始节点 - * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复 - * @param userTaskList 需要撤回的用户任务列表 - * @return 用户任务节点列表 - */ - public static List findChildProcessUserTasks(FlowElement source, Set hasSequenceFlow, - List userTaskList) { - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - userTaskList = userTaskList == null ? new ArrayList<>() : userTaskList; - - // 根据类型，获取出口连线 - List sequenceFlows = getElementOutgoingFlows(source); - - if (sequenceFlows != null) { - // 循环找到目标元素 - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 如果为用户任务类型，且任务节点的 Key 正在运行的任务中存在，添加 - if (sequenceFlow.getTargetFlowElement() instanceof UserTask) { - userTaskList.add((UserTask) sequenceFlow.getTargetFlowElement()); - continue; - } - // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取 - if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) { - List childUserTaskList = findChildProcessUserTasks( - (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements() - .toArray()[0]), - hasSequenceFlow, null); - // 如果找到节点，则说明该线路找到节点，不继续向下找，反之继续 - if (childUserTaskList != null && childUserTaskList.size() > 0) { - userTaskList.addAll(childUserTaskList); - continue; - } - } - // 继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - userTaskList = findChildProcessUserTasks(sequenceFlow.getTargetFlowElement(), - new HashSet<>(hasSequenceFlow), userTaskList); - } - } - return userTaskList; - } - - /** - * 从后向前寻路，获取所有脏线路上的点 - *

- * 脏线路是指由于流程回退、驳回等操作产生的废弃路径。 - * 该方法用于识别这些废弃路径上的所有节点，以便进行清理或过滤。 - * - * @param source 起始节点 - * @param passRoads 已经经过的点集合 - * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复 - * @param targets 目标脏线路终点 - * @param dirtyRoads 确定为脏数据的点，因为不需要重复，因此使用set存储 - * @return 脏线路上的所有节点ID集合 - */ - public static Set iteratorFindDirtyRoads(FlowElement source, List passRoads, - Set hasSequenceFlow, List targets, Set dirtyRoads) { - passRoads = passRoads == null ? new ArrayList<>() : passRoads; - dirtyRoads = dirtyRoads == null ? new HashSet<>() : dirtyRoads; - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - - // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代 - if (source instanceof StartEvent && source.getSubProcess() != null) { - dirtyRoads = iteratorFindDirtyRoads(source.getSubProcess(), passRoads, hasSequenceFlow, targets, - dirtyRoads); - } - - // 根据类型，获取入口连线 - List sequenceFlows = getElementIncomingFlows(source); - - if (sequenceFlows != null) { - // 循环找到目标元素 - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 新增经过的路线 - passRoads.add(sequenceFlow.getSourceFlowElement().getId()); - // 如果此点为目标点，确定经过的路线为脏线路，添加点到脏线路中，然后找下个连线 - if (targets.contains(sequenceFlow.getSourceFlowElement().getId())) { - dirtyRoads.addAll(passRoads); - continue; - } - // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代 - if (sequenceFlow.getSourceFlowElement() instanceof SubProcess) { - dirtyRoads = findChildProcessAllDirtyRoad( - (StartEvent) ((SubProcess) sequenceFlow.getSourceFlowElement()).getFlowElements() - .toArray()[0], - null, dirtyRoads); - // 是否存在子流程上，true 是，false 否 - Boolean isInChildProcess = dirtyTargetInChildProcess( - (StartEvent) ((SubProcess) sequenceFlow.getSourceFlowElement()).getFlowElements() - .toArray()[0], - null, targets, null); - if (isInChildProcess) { - // 已在子流程上找到，该路线结束 - continue; - } - } - // 继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - dirtyRoads = iteratorFindDirtyRoads(sequenceFlow.getSourceFlowElement(), new ArrayList<>(passRoads), - new HashSet<>(hasSequenceFlow), targets, dirtyRoads); - } - } - return dirtyRoads; - } - - /** - * 迭代获取子流程脏路线 - *

- * 当回退点是子流程时，子流程中的所有节点都被认为是脏路线。 - * 该方法递归遍历子流程，获取其中的所有节点。 - * - * @param source 起始节点 - * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复 - * @param dirtyRoads 确定为脏数据的点，因为不需要重复，因此使用set存储 - * @return 子流程中所有节点的ID集合 - */ - public static Set findChildProcessAllDirtyRoad(FlowElement source, Set hasSequenceFlow, - Set dirtyRoads) { - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - dirtyRoads = dirtyRoads == null ? new HashSet<>() : dirtyRoads; - - // 根据类型，获取出口连线 - List sequenceFlows = getElementOutgoingFlows(source); - - if (sequenceFlows != null) { - // 循环找到目标元素 - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 添加脏路线 - dirtyRoads.add(sequenceFlow.getTargetFlowElement().getId()); - // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取 - if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) { - dirtyRoads = findChildProcessAllDirtyRoad( - (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements() - .toArray()[0]), - hasSequenceFlow, dirtyRoads); - } - // 继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - dirtyRoads = findChildProcessAllDirtyRoad(sequenceFlow.getTargetFlowElement(), - new HashSet<>(hasSequenceFlow), dirtyRoads); - } - } - return dirtyRoads; - } - - /** - * 判断脏路线结束节点是否在子流程上 - *

- * 用于判断脏路线的范围，如果脏路线终点在子流程中， - * 说明脏路线只到子流程为止，不会影响父流程。 - * - * @param source 起始节点 - * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复 - * @param targets 判断脏路线节点是否存在子流程上，只要存在一个，说明脏路线只到子流程为止 - * @param inChildProcess 是否存在子流程上，true是，false否 - * @return 是否在子流程中 - */ - public static Boolean dirtyTargetInChildProcess(FlowElement source, Set hasSequenceFlow, - List targets, Boolean inChildProcess) { - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - inChildProcess = inChildProcess == null ? false : inChildProcess; - - // 根据类型，获取出口连线 - List sequenceFlows = getElementOutgoingFlows(source); - - if (sequenceFlows != null && !inChildProcess) { - // 循环找到目标元素 - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 如果发现目标点在子流程上存在，说明只到子流程为止 - if (targets.contains(sequenceFlow.getTargetFlowElement().getId())) { - inChildProcess = true; - break; - } - // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取 - if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) { - inChildProcess = dirtyTargetInChildProcess( - (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements() - .toArray()[0]), - hasSequenceFlow, targets, inChildProcess); - } - // 继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - inChildProcess = dirtyTargetInChildProcess(sequenceFlow.getTargetFlowElement(), - new HashSet<>(hasSequenceFlow), targets, inChildProcess); - } - } - return inChildProcess; - } - - /** - * 迭代从后向前扫描，判断目标节点相对于当前节点是否是串行 - *

- * 用于判断两个节点之间是否存在并行关系。 - * 如果两个节点之间只有唯一路径，则为串行；如果存在多条路径，则为并行。 - * 不存在直接回退到子流程中的情况，但存在从子流程出去到父流程情况。 - * - * @param source 起始节点 - * @param targetKsy 目标节点ID - * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复 - * @param isSequential 是否串行，默认为true - * @return true表示串行，false表示并行 - */ - public static Boolean iteratorCheckSequentialReferTarget(FlowElement source, String targetKsy, - Set hasSequenceFlow, Boolean isSequential) { - isSequential = isSequential == null ? true : isSequential; - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - - // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代 - if (source instanceof StartEvent && source.getSubProcess() != null) { - isSequential = iteratorCheckSequentialReferTarget(source.getSubProcess(), targetKsy, hasSequenceFlow, - isSequential); - } - - // 根据类型，获取入口连线 - List sequenceFlows = getElementIncomingFlows(source); - - if (sequenceFlows != null) { - // 循环找到目标元素 - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 如果目标节点已被判断为并行，后面都不需要执行，直接返回 - if (isSequential == false) { - break; - } - // 这条线路存在目标节点，这条线路完成，进入下个线路 - if (targetKsy.equals(sequenceFlow.getSourceFlowElement().getId())) { - continue; - } - if (sequenceFlow.getSourceFlowElement() instanceof StartEvent) { - isSequential = false; - break; - } - // 否则就继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - isSequential = iteratorCheckSequentialReferTarget(sequenceFlow.getSourceFlowElement(), targetKsy, - new HashSet<>(hasSequenceFlow), isSequential); - } - } - return isSequential; - } - - /** - * 从后向前寻路，获取到达节点的所有路线 - *

- * 找到从当前节点到开始节点的所有可能路径。 - * 每条路径是一个用户任务节点的有序列表。 - * 不存在直接回退到子流程，但是存在回退到父级流程的情况。 - * - * @param source 起始节点 - * @param passRoads 已经经过的点集合 - * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复 - * @param roads 路线集合 - * @return 所有可能的路径列表 - */ - public static List> findRoad(FlowElement source, List passRoads, - Set hasSequenceFlow, List> roads) { - passRoads = passRoads == null ? new ArrayList<>() : passRoads; - roads = roads == null ? new ArrayList<>() : roads; - hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow; - - // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代 - if (source instanceof StartEvent && source.getSubProcess() != null) { - roads = findRoad(source.getSubProcess(), passRoads, hasSequenceFlow, roads); - } - - // 根据类型，获取入口连线 - List sequenceFlows = getElementIncomingFlows(source); - - if (sequenceFlows != null && sequenceFlows.size() != 0) { - for (SequenceFlow sequenceFlow : sequenceFlows) { - // 如果发现连线重复，说明循环了，跳过这个循环 - if (hasSequenceFlow.contains(sequenceFlow.getId())) { - continue; - } - // 添加已经走过的连线 - hasSequenceFlow.add(sequenceFlow.getId()); - // 添加经过路线 - if (sequenceFlow.getSourceFlowElement() instanceof UserTask) { - passRoads.add((UserTask) sequenceFlow.getSourceFlowElement()); - } - // 继续迭代 - // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象 - // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据 - roads = findRoad(sequenceFlow.getSourceFlowElement(), new ArrayList<>(passRoads), - new HashSet<>(hasSequenceFlow), roads); - } - } else { - // 添加路线 - roads.add(passRoads); - } - return roads; - } - - /** - * 历史节点数据清洗，清洗掉由于回滚导致的脏数据 - *

- * 在流程回退、驳回等操作后，历史数据中会存在废弃的节点记录。 - * 该方法通过分析删除原因和节点关系，识别并过滤掉这些脏数据， - * 返回真实有效的历史节点记录。 - *

- * 特别处理： - * - 会签节点的多实例情况 - * - 并行网关的多分支情况 - * - 连续驳回的情况 - * - * @param allElements 全部节点信息 - * @param historicActivityIdList 历史任务实例信息，数据采用开始时间升序 - * @return 清洗后的有效历史节点ID列表 - */ - public static List historicTaskInstanceClean(Collection allElements, - List historicActivityIdList) { - // 会签节点收集 - List multiTask = new ArrayList<>(); - allElements.forEach(flowElement -> { - if (flowElement instanceof UserTask) { - // 如果该节点的行为为会签行为，说明该节点为会签节点 - if (((UserTask) flowElement).getBehavior() instanceof ParallelMultiInstanceBehavior - || ((UserTask) flowElement).getBehavior() instanceof SequentialMultiInstanceBehavior) { - multiTask.add(flowElement.getId()); - } - } - }); - // 循环放入栈，栈 LIFO：后进先出 - Stack stack = new Stack<>(); - historicActivityIdList.forEach(item -> stack.push(item)); - // 清洗后的历史任务实例 - List lastHistoricTaskInstanceList = new ArrayList<>(); - // 网关存在可能只走了部分分支情况，且还存在跳转废弃数据以及其他分支数据的干扰，因此需要对历史节点数据进行清洗 - // 临时用户任务 key - StringBuilder userTaskKey = null; - // 临时被删掉的任务 key，存在并行情况 - List deleteKeyList = new ArrayList<>(); - // 临时脏数据线路 - List> dirtyDataLineList = new ArrayList<>(); - // 由某个点跳到会签点,此时出现多个会签实例对应 1 个跳转情况，需要把这些连续脏数据都找到 - // 会签特殊处理下标 - int multiIndex = -1; - // 会签特殊处理 key - StringBuilder multiKey = null; - // 会签特殊处理操作标识 - boolean multiOpera = false; - while (!stack.empty()) { - // 从这里开始 userTaskKey 都还是上个栈的 key - // 是否是脏数据线路上的点 - final boolean[] isDirtyData = {false}; - for (Set oldDirtyDataLine : dirtyDataLineList) { - if (oldDirtyDataLine.contains(stack.peek().getActivityId())) { - isDirtyData[0] = true; - } - } - // 删除原因不为空，说明从这条数据开始回跳或者回退的 - // MI_END：会签完成后，其他未签到节点的删除原因，不在处理范围内 - if (stack.peek().getDeleteReason() != null && !stack.peek().getDeleteReason().equals("MI_END")) { - // 可以理解为脏线路起点 - String dirtyPoint = ""; - if (stack.peek().getDeleteReason().indexOf("Change activity to ") >= 0) { - dirtyPoint = stack.peek().getDeleteReason().replace("Change activity to ", ""); - } - // 会签回退删除原因有点不同 - if (stack.peek().getDeleteReason().indexOf("Change parent activity to ") >= 0) { - dirtyPoint = stack.peek().getDeleteReason().replace("Change parent activity to ", ""); - } - FlowElement dirtyTask = null; - // 获取变更节点的对应的入口处连线 - // 如果是网关并行回退情况，会变成两条脏数据路线，效果一样 - for (FlowElement flowElement : allElements) { - if (flowElement.getId().equals(stack.peek().getActivityId())) { - dirtyTask = flowElement; - } - } - // 获取脏数据线路 - Set dirtyDataLine = FlowableUtils.iteratorFindDirtyRoads(dirtyTask, null, null, - Arrays.asList(dirtyPoint.split(",")), null); - // 自己本身也是脏线路上的点，加进去 - dirtyDataLine.add(stack.peek().getActivityId()); - logger.info(stack.peek().getActivityId() + "点脏路线集合：" + dirtyDataLine); - // 是全新的需要添加的脏线路 - boolean isNewDirtyData = true; - for (int i = 0; i < dirtyDataLineList.size(); i++) { - // 如果发现他的上个节点在脏线路内，说明这个点可能是并行的节点，或者连续驳回 - // 这时，都以之前的脏线路节点为标准，只需合并脏线路即可，也就是路线补全 - if (dirtyDataLineList.get(i).contains(userTaskKey.toString())) { - isNewDirtyData = false; - dirtyDataLineList.get(i).addAll(dirtyDataLine); - } - } - // 已确定时全新的脏线路 - if (isNewDirtyData) { - // deleteKey 单一路线驳回到并行，这种同时生成多个新实例记录情况，这时 deleteKey 其实是由多个值组成 - // 按照逻辑，回退后立刻生成的实例记录就是回退的记录 - // 至于驳回所生成的 Key，直接从删除原因中获取，因为存在驳回到并行的情况 - deleteKeyList.add(dirtyPoint + ","); - dirtyDataLineList.add(dirtyDataLine); - } - // 添加后，现在这个点变成脏线路上的点了 - isDirtyData[0] = true; - } - // 如果不是脏线路上的点，说明是有效数据，添加历史实例 Key - if (!isDirtyData[0]) { - lastHistoricTaskInstanceList.add(stack.peek().getActivityId()); - } - // 校验脏线路是否结束 - for (int i = 0; i < deleteKeyList.size(); i++) { - // 如果发现脏数据属于会签，记录下下标与对应 Key，以备后续比对，会签脏数据范畴开始 - if (multiKey == null && multiTask.contains(stack.peek().getActivityId()) - && deleteKeyList.get(i).contains(stack.peek().getActivityId())) { - multiIndex = i; - multiKey = new StringBuilder(stack.peek().getActivityId()); - } - // 会签脏数据处理，节点退回会签清空 - // 如果在会签脏数据范畴中发现 Key改变，说明会签脏数据在上个节点就结束了，可以把会签脏数据删掉 - if (multiKey != null && !multiKey.toString().equals(stack.peek().getActivityId())) { - deleteKeyList.set(multiIndex, - deleteKeyList.get(multiIndex).replace(stack.peek().getActivityId() + ",", "")); - multiKey = null; - // 结束进行下校验删除 - multiOpera = true; - } - // 其他脏数据处理 - // 发现该路线最后一条脏数据，说明这条脏数据线路处理完了，删除脏数据信息 - // 脏数据产生的新实例中是否包含这条数据 - if (multiKey == null && deleteKeyList.get(i).contains(stack.peek().getActivityId())) { - // 删除匹配到的部分 - deleteKeyList.set(i, deleteKeyList.get(i).replace(stack.peek().getActivityId() + ",", "")); - } - // 如果每组中的元素都以匹配过，说明脏数据结束 - if ("".equals(deleteKeyList.get(i))) { - // 同时删除脏数据 - deleteKeyList.remove(i); - dirtyDataLineList.remove(i); - break; - } - } - // 会签数据处理需要在循环外处理，否则可能导致溢出 - // 会签的数据肯定是之前放进去的所以理论上不会溢出，但还是校验下 - if (multiOpera && deleteKeyList.size() > multiIndex && "".equals(deleteKeyList.get(multiIndex))) { - // 同时删除脏数据 - deleteKeyList.remove(multiIndex); - dirtyDataLineList.remove(multiIndex); - multiIndex = -1; - multiOpera = false; - } - // pop() 方法与 peek() 方法不同，在返回值的同时，会把值从栈中移除 - // 保存新的 userTaskKey 在下个循环中使用 - userTaskKey = new StringBuilder(stack.pop().getActivityId()); - } - logger.info("清洗后的历史节点数据：" + lastHistoricTaskInstanceList); - return lastHistoricTaskInstanceList; - } - - /** - * 获取流程参数Map，并设置发起人信息 - * - * @param flowBaseDto 流程基础数据传输对象 - * @return 包含发起人信息的流程参数Map - */ - public static Map getFlowParamMap(BizFlowBaseDto flowBaseDto) { - Map paramMap = flowBaseDto.getFlowParamMap(); - // 设置发起人 (当前登录用户) - Dict rootUser = Dict.create() - .set("id", SecurityUtils.getUser().getId()) - .set("name", SecurityUtils.getUser().getUsername()) - .set("type", NodeUserTypeEnum.USER.getKey()); - paramMap.put("root", CollUtil.newArrayList(rootUser)); - return paramMap; - } -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowCallbackDelivery.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowCallbackDelivery.java similarity index 98% rename from pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowCallbackDelivery.java rename to pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowCallbackDelivery.java index 6d38c4727..a095f45e0 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowCallbackDelivery.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowCallbackDelivery.java @@ -1,4 +1,4 @@ -package com.pig4cloud.pigx.flow.service; +package com.pig4cloud.pigx.flow.support.workspace; import cn.hutool.core.util.StrUtil; import cn.hutool.http.HttpRequest; diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowDraftService.java similarity index 99% rename from pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java rename to pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowDraftService.java index c65e8a643..4bb610b66 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowDraftService.java @@ -1,4 +1,4 @@ -package com.pig4cloud.pigx.flow.service; +package com.pig4cloud.pigx.flow.support.workspace; import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONUtil; diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowReliabilityService.java similarity index 99% rename from pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java rename to pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowReliabilityService.java index de939d41d..2a3064517 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowReliabilityService.java @@ -1,4 +1,6 @@ -package com.pig4cloud.pigx.flow.service; +package com.pig4cloud.pigx.flow.support.workspace; + +import com.pig4cloud.pigx.flow.service.IProcessNodeDataService; import cn.hutool.core.convert.Convert; import cn.hutool.core.util.StrUtil; diff --git a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java deleted file mode 100644 index 6c6fc8b3d..000000000 --- a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java +++ /dev/null @@ -1,297 +0,0 @@ -package com.pig4cloud.pigx.flow.service; - -import com.pig4cloud.pigx.common.core.util.R; -import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO; -import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService; -import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder; -import cn.hutool.json.JSONUtil; -import com.pig4cloud.pigx.flow.dto.Node; -import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum; -import com.pig4cloud.pigx.flow.config.FlowDurableConfiguration; -import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener; -import com.sun.net.httpserver.HttpServer; -import org.flowable.engine.ProcessEngine; -import org.flowable.engine.ProcessEngineConfiguration; -import org.flowable.spring.SpringProcessEngineConfiguration; -import org.h2.jdbcx.JdbcDataSource; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Test; -import org.springframework.aop.framework.ProxyFactory; -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.datasource.DataSourceTransactionManager; -import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; -import org.springframework.transaction.interceptor.TransactionInterceptor; -import org.springframework.transaction.support.TransactionTemplate; -import java.net.InetSocketAddress; -import java.nio.file.Files; -import java.nio.file.Path; -import java.util.ArrayList; -import java.util.List; -import java.util.UUID; -import java.util.concurrent.atomic.AtomicInteger; -import static org.assertj.core.api.Assertions.*; -import static org.mockito.Mockito.*; - -class FlowWorkspaceTests { - - private ProcessEngine engine; - - private JdbcTemplate jdbc; - - private TransactionTemplate transaction; - - private FlowReliabilityService reliability; - - private FlowDraftService drafts; - - private RemoteMessageService messages; - - private final List sentNotices = new ArrayList<>(); - - private HttpServer server; - - private final List eventIds = new ArrayList<>(); - - private final AtomicInteger requests = new AtomicInteger(); - - @BeforeEach - void setUp() throws Exception { - JdbcDataSource source = new JdbcDataSource(); - source.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE"); - source.setUser("sa"); - jdbc = new JdbcTemplate(source); - jdbc.execute("SET MODE MySQL"); - String schema = Files.readString(Path.of("../../db/migration/20261014_flow_first_batch.sql")); - for (String sql : schema.split(";")) - if (!sql.isBlank()) - jdbc.execute(sql); - jdbc.execute( - "CREATE TABLE process_instance_record(id bigint primary key,process_instance_id varchar(64),tenant_id bigint,user_id bigint,status int,del_flag char(1),name varchar(200))"); - var manager = new DataSourceTransactionManager(source); - transaction = new TransactionTemplate(manager); - var configuration = new SpringProcessEngineConfiguration(); - configuration.setDataSource(source); - configuration.setTransactionManager(manager); - configuration.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE); - configuration.setDisableIdmEngine(true); - configuration.setDisableEventRegistry(true); - configuration.setAsyncExecutorActivate(false); - ObjectProvider provider = mock(ObjectProvider.class); - when(provider.getObject()).thenAnswer(invocation -> reliability); - configuration.setEventListeners(List.of(new FlowDurableEventListener(provider))); - jdbc.execute("SET MODE LEGACY"); - engine = configuration.buildProcessEngine(); - jdbc.execute("SET MODE MySQL"); - server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); - server.createContext("/callback", exchange -> { - synchronized (eventIds) { - eventIds.add(exchange.getRequestHeaders().getFirst("X-Flow-Event-Id")); - } - exchange.getRequestBody().readAllBytes(); - exchange.sendResponseHeaders(requests.incrementAndGet() == 1 ? 503 : 204, -1); - exchange.close(); - }); - server.start(); - Node node = new Node(); - node.setReminderMinutes(1); - node.setEventConfig("http://127.0.0.1:" + server.getAddress().getPort() + "/callback"); - var nodes = mock(IProcessNodeDataService.class); - when(nodes.getPublishedNode(anyString(), anyString())).thenReturn(R.ok(node)); - messages = mock(RemoteMessageService.class); - when(messages.sendNotice(any())).thenAnswer(invocation -> { - assertThat(TenantContextHolder.getTenantId()).isEqualTo(1L); - sentNotices.add(invocation.getArgument(0)); - return R.ok(123L); - }); - var target = new FlowReliabilityService(jdbc, nodes, engine.getTaskService(), messages); - var proxy = new ProxyFactory(target); - proxy.addAdvice(new TransactionInterceptor(manager, new AnnotationTransactionAttributeSource())); - reliability = (FlowReliabilityService) proxy.getProxy(); - drafts = new FlowDraftService(jdbc); - engine.getRepositoryService() - .createDeployment() - .tenantId("1") - .addString("test.bpmn20.xml", - """ - - - - - - - - """) - .deploy(); - } - - @AfterEach - void tearDown() { - if (server != null) - server.stop(0); - if (engine != null) - engine.close(); - if (jdbc != null) - jdbc.execute("SHUTDOWN"); - } - - @Test - void draftsAndTaskEventsSurviveNewServiceInstancesAndHttpRetries() throws Exception { - // 请求仍能读取个人表单，但操作日志使用的 Jackson 不得输出标题和载荷。 - var logMapper = com.pig4cloud.pigx.common.log.util.JacksonSensitiveFieldUtil.getObjectMapper().copy(); - logMapper.setFilterProvider(new com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider() - .setDefaultFilter(com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter.serializeAll())); - var input = logMapper.readValue( - "{\"kind\":\"APPLICATION\",\"flowId\":\"approval\",\"title\":\"private-title\",\"payload\":\"{\\\"secret\\\":1}\",\"revision\":0}", - FlowDraftService.Draft.class); - assertThat(input.title()).isEqualTo("private-title"); - assertThat(input.payload()).contains("secret"); - assertThat(logMapper.writeValueAsString(new Object[] { input })).doesNotContain("private-title", "secret", - "payload", "title"); - assertThat(FlowNoticeStatusEnum.values()).extracting(FlowNoticeStatusEnum::getCode) - .containsExactly("PENDING", "READY", "SENT", "CANCELLED", "DISABLED"); - var design = drafts.save(1, 7, - new FlowDraftService.Draft(null, "DESIGN", "approval", "设计", "{\"step1\":{}}", 0)); - var application = drafts.save(1, 7, - new FlowDraftService.Draft(null, "APPLICATION", "approval", "申请", "{\"formData\":{\"amount\":8}}", 0)); - assertThat(new FlowDraftService(jdbc).get(1, 7, design.get("id").toString()).get("payload")) - .isEqualTo("{\"step1\":{}}"); - assertThat(drafts.list(1, 7, "APPLICATION", 1, null)).hasSize(1); - assertThat(drafts.list(1, 7, "APPLICATION", 1, "approval")).hasSize(1); - assertThat(drafts.list(1, 7, "APPLICATION", 1, "other-flow")).isEmpty(); - assertThat(drafts.list(1, 8, "APPLICATION", 1, "approval")).isEmpty(); - assertThat(drafts.get(1, 7, application.get("id").toString()).get("payload")) - .isEqualTo("{\"formData\":{\"amount\":8}}"); - var instance = engine.getRuntimeService() - .startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); - jdbc.update("INSERT INTO process_instance_record VALUES(1,?,1,7,1,'0','差旅申请')", instance.getId()); - var task = engine.getTaskService().createTaskQuery().singleResult(); - verifyNoInteractions(messages); - String notice = jdbc.queryForObject("SELECT id FROM flow_notice WHERE status='READY'", String.class); - reliability.deliverNotice(notice); - reliability.deliverNotice(notice); - assertThat(sentNotices).hasSize(1); - assertThat(sentNotices.get(0).getUserIds()).containsExactly(1001L); - assertThat(sentNotices.get(0).getBusinessType()).isEqualTo("FLOW_TASK"); - assertThat(sentNotices.get(0).getBusinessId()).isEqualTo(task.getId()); - assertThat(sentNotices.get(0).getRequestKey()).isEqualTo("flow:" + notice); - assertThat(sentNotices.get(0).getCategory()).isEqualTo("1"); - assertThat(sentNotices.get(0).getAllFlag()).isEqualTo("0"); - assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class)) - .isEqualTo(1); - engine.getRuntimeService().suspendProcessInstanceById(instance.getId()); - jdbc.update( - "UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE status='PENDING'"); - reliability.remind(1, task.getId()); - assertThat( - jdbc.queryForObject("SELECT status FROM flow_notice WHERE reminder_key=?", String.class, task.getId())) - .isEqualTo("PENDING"); - assertThat(jdbc.queryForObject("SELECT scheduled_at>CURRENT_TIMESTAMP(3) FROM flow_notice WHERE reminder_key=?", - Boolean.class, task.getId())) - .isTrue(); - engine.getRuntimeService().activateProcessInstanceById(instance.getId()); - engine.getTaskService().setAssignee(task.getId(), "2002"); - jdbc.update( - "UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE status='PENDING'"); - ObjectProvider provider = mock(ObjectProvider.class); - when(provider.getObject()).thenReturn(reliability); - var scheduler = new FlowDurableConfiguration(jdbc, provider, mock(ObjectProvider.class)); - scheduler.scanReminders(); - scheduler.scanReminders(); - assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE tenant_id=1 AND user_id='2002'", Integer.class)).isEqualTo(2); - engine.getTaskService().setAssignee(task.getId(), "1001"); - assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE tenant_id=1 AND user_id='1001'", Integer.class)).isEqualTo(2); - engine.getTaskService().setAssignee(task.getId(), "2002"); - assertThat(reliability.urge(1, 7, instance.getId())).isEqualTo(1); - assertThatThrownBy(() -> reliability.urge(1, 7, instance.getId())).hasMessageContaining("十分钟"); - scheduler.scanNotices(); - assertThat(sentNotices).extracting(MessageNoticeDTO::getTitle) - .anyMatch(title -> title.startsWith("待办:")) - .anyMatch(title -> title.equals("催办:差旅申请 · 审批")) - .anyMatch(title -> title.startsWith("超时提醒:")); - assertThat(sentNotices.stream().filter(message -> message.getTitle().startsWith("超时提醒:")) - .map(MessageNoticeDTO::getUserIds).toList()).containsExactly(List.of(2002L)); - engine.getTaskService().complete(task.getId()); - assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class)) - .isZero(); - var callback = reliability.callbacks(1, 7, instance.getId()).get(0); - String id = callback.get("id").toString(); - new FlowCallbackDelivery(jdbc, transaction).deliver(id); - assertThat(jdbc.queryForObject("SELECT status FROM flow_callback", String.class)).isEqualTo("PENDING"); - jdbc.update("UPDATE flow_callback SET next_at=CURRENT_TIMESTAMP"); - new FlowCallbackDelivery(jdbc, transaction).deliver(id); - assertThat(jdbc.queryForObject("SELECT status FROM flow_callback", String.class)).isEqualTo("DELIVERED"); - var history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class)); - assertThat(history).hasSize(2); - assertThat(history.getJSONObject(0).getStr("result")).isEqualTo("ERROR"); - assertThat(history.getJSONObject(1).getStr("result")).isEqualTo("DELIVERED"); - jdbc.update("UPDATE flow_callback SET status='FAILED'"); - reliability.retry(1, 7, id); - history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class)); - assertThat(history).hasSize(3); - assertThat(history.getJSONObject(2).getStr("result")).isEqualTo("REQUEUED"); - new FlowCallbackDelivery(jdbc, transaction).deliver(id); - history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class)); - assertThat(history).hasSize(4); - assertThat(history.getJSONObject(0).getStr("result")).isEqualTo("ERROR"); - assertThat(history.getJSONObject(3).getStr("result")).isEqualTo("DELIVERED"); - assertThat(eventIds).containsExactly(id, id, id); - engine.getRuntimeService().startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); - String completed = engine.getTaskService().createTaskQuery().singleResult().getId(); - engine.getTaskService().complete(completed); - jdbc.update("UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE reminder_key=?", - completed); - reliability.remind(1, completed); - assertThat(jdbc.queryForObject("SELECT status FROM flow_notice WHERE reminder_key=?", String.class, completed)) - .isEqualTo("CANCELLED"); - assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE reminder_key=? AND status='SENT'", - Integer.class, completed)) - .isZero(); - } - - @Test - void ownershipConflictsAndStorageFailuresCannotLoseCommittedWork() { - var saved = drafts.save(1, 7, new FlowDraftService.Draft(null, "APPLICATION", "approval", "申请", "{}", 0)); - String id = saved.get("id").toString(); - assertThatThrownBy(() -> drafts.get(2, 7, id)).hasMessageContaining("无权"); - assertThatThrownBy(() -> drafts.get(1, 8, id)).hasMessageContaining("无权"); - assertThatThrownBy( - () -> drafts.save(1, 7, new FlowDraftService.Draft(id, "APPLICATION", "approval", "申请", "{}", 0))) - .hasMessageContaining("重新打开"); - assertThat(drafts.delete(1, 8, id, 1)).isFalse(); - assertThatThrownBy(() -> transaction.executeWithoutResult(tx -> { - engine.getRuntimeService().startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); - throw new IllegalStateException("rollback"); - })).hasMessage("rollback"); - assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice", Integer.class)).isZero(); - assertThat(engine.getTaskService().createTaskQuery().count()).isZero(); - verifyNoInteractions(messages); - var instance = engine.getRuntimeService() - .startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); - jdbc.update("INSERT INTO process_instance_record VALUES(1,?,1,7,1,'0','差旅申请')", instance.getId()); - assertThatThrownBy(() -> reliability.urge(2, 7, instance.getId())).hasMessageContaining("无权"); - assertThatThrownBy(() -> reliability.callbacks(1, 8, instance.getId())).hasMessageContaining("无权"); - String notice = jdbc.queryForObject("SELECT id FROM flow_notice WHERE status='READY'", String.class); - doReturn(R.failed("发送暂不可用"), R.ok(123L)).when(messages).sendNotice(any()); - ObjectProvider provider = mock(ObjectProvider.class); - when(provider.getObject()).thenReturn(reliability); - var scheduler = new FlowDurableConfiguration(jdbc, provider, mock(ObjectProvider.class)); - scheduler.scanNotices(); - assertThat(jdbc.queryForObject("SELECT status FROM flow_notice WHERE id=?", String.class, notice)).isEqualTo("READY"); - assertThat(jdbc.queryForObject("SELECT scheduled_at>CURRENT_TIMESTAMP(3) FROM flow_notice WHERE id=?", Boolean.class, notice)).isTrue(); - jdbc.update("UPDATE flow_notice SET scheduled_at=CURRENT_TIMESTAMP(3) WHERE id=?", notice); - scheduler.scanNotices(); - assertThat(jdbc.queryForObject("SELECT status FROM flow_notice WHERE id=?", String.class, notice)).isEqualTo("SENT"); - var captor = org.mockito.ArgumentCaptor.forClass(MessageNoticeDTO.class); - verify(messages, times(2)).sendNotice(captor.capture()); - assertThat(captor.getAllValues()).extracting(MessageNoticeDTO::getRequestKey).containsExactly("flow:" + notice, "flow:" + notice); - String task = engine.getTaskService().createTaskQuery().singleResult().getId(); - jdbc.execute("DROP TABLE flow_callback"); - assertThatThrownBy(() -> engine.getTaskService().complete(task)).isInstanceOf(RuntimeException.class); - assertThat(engine.getTaskService().createTaskQuery().taskId(task).count()).isEqualTo(1); - assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class)) - .isEqualTo(1); - } - -}