- * 该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);
- }
-
-}