fix(flow): 按发布版本隔离流程节点配置

(cherry picked from commit 86ae98ad5b29a6d20b7e44283549d7337bb961df)
This commit is contained in:
PIGCLOUD committed 2026-11-03 14:55:00 +08:00
1 parent a177f3073a
commit 749931b53e
28 files changed
+3328 -2767

No files matched your search

+2
View File
@@ -1701,11 +1701,13 @@ CREATE TABLE `process_node_data` (
`del_flag` char(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT '0' COMMENT '删除标志',
`create_time` datetime COMMENT '创建时间',
`update_time` datetime COMMENT '更新时间',
`process_definition_id` varchar(64) DEFAULT NULL COMMENT '引擎流程定义ID,发布版本快照',
`flow_id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '流程id',
`data` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '表单数据',
`node_id` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`tenant_id` bigint(20) NOT NULL DEFAULT '0' COMMENT '所属租户id',
PRIMARY KEY (`id`) USING BTREE,
UNIQUE KEY `uk_node_definition` (`tenant_id`,`process_definition_id`,`node_id`),
KEY `data_idx_id` (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1195 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC COMMENT='流程节点数据';
+2
View File
@@ -138,11 +138,13 @@ CREATE TABLE `process_node_data` (
`del_flag` char(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT '0' COMMENT '删除标志',
`create_time` datetime COMMENT '创建时间',
`update_time` datetime COMMENT '更新时间',
`process_definition_id` varchar(64) DEFAULT NULL COMMENT '引擎流程定义ID,发布版本快照',
`flow_id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '流程id',
`data` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '表单数据',
`node_id` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`tenant_id` bigint(20) NOT NULL DEFAULT '0' COMMENT '所属租户id',
PRIMARY KEY (`id`) USING BTREE,
UNIQUE KEY `uk_node_definition` (`tenant_id`,`process_definition_id`,`node_id`),
KEY `data_idx_id` (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1195 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC COMMENT='流程节点数据';
@@ -0,0 +1,69 @@
# 节点配置版本隔离升级
本次只隔离节点配置。业务 flowId 不变;流程图、表单定义的完整历史快照不在本次范围。任务、抄送和实例详情使用对应版本的字段权限,当前表单中未被该快照声明的字段默认隐藏;这不替代服务端的提交字段鉴权。已结束实例的详情依赖 Flowable 历史数据来确定版本,历史已清理时会明确报错,不能回退最新版。
## 上线顺序
1. 暂停流程发布、发起、任务处理和异步执行器,备份流程业务表及 Flowable 数据。不得新旧版本混跑:旧代码会按 flowId + nodeId 删除节点记录。
2. 对已有库执行同目录 20261013_flow_node_version.sql;新库直接使用初始化脚本。升级 SQL 只执行一次,不删除旧配置。
3. 核对存量定义与节点配置,建立明确的版本映射。至少覆盖所有运行中定义;如需要查询历史任务、抄送或使用旧定义,也必须覆盖对应历史定义。
4. 将已核验配置写入对应 process_definition_id,包含 root 和所有业务节点。每个租户、定义、节点只能有一条快照。按该版本的已核验表单字段集合补齐各节点 formPerms:原来未配置/空值的字段按原有前端默认语义补 E,已有 H/R/E 保留;包含 children 中的嵌套字段及带 field 的容器。不能用当前表单的新增字段补旧版权限。无法恢复原字段集合时,此版本迁移尚未完成。
5. 复查覆盖情况,部署全部新代码后恢复业务。无法确认映射时不要恢复受影响流程;运行期会明确报“缺少节点版本快照”,不会回退读取最新版。
## 只读盘点
在工作流所在数据库执行以下查询;若自定义了 Flowable 表名前缀,请先调整。
SELECT DISTINCT d.TENANT_ID_, d.KEY_, d.ID_, d.VERSION_, d.DEPLOYMENT_ID_
FROM ACT_RU_EXECUTION e
JOIN ACT_RE_PROCDEF d ON d.ID_ = e.PROC_DEF_ID_
ORDER BY d.TENANT_ID_, d.KEY_, d.VERSION_;
SELECT p.tenant_id, p.flow_id, p.unique_id, d.ID_, d.VERSION_, p.process
FROM process p
JOIN ACT_RE_PROCDEF d
ON d.DEPLOYMENT_ID_ = p.unique_id
AND d.KEY_ = p.flow_id
AND d.TENANT_ID_ = CAST(p.tenant_id AS CHAR)
WHERE p.del_flag = '0';
SELECT id, tenant_id, flow_id, node_id, del_flag, process_definition_id, data
FROM process_node_data
ORDER BY tenant_id, flow_id, node_id, id;
process.unique_id 存储部署 ID,可以定位当前模板的引擎定义。它只能帮助定位当前发布版本,不代表所有旧节点记录都属于该版本。旧节点表保留了逻辑删除数据,但没有发布版本标记;不能按行 ID 或时间顺序自动配对。
核对节点 JSON 与相应发布资料/备份,包括审批人、多实例模式、拒绝策略、字段权限、触发器和通知地址。配置中的敏感业务内容不要写入审查日志。
## 存量写入模板
下面只是单行映射示例,所有占位值都必须替换为已核验结果。先 SELECT 确认目标行和定义的租户、业务 key、版本一致,再在受控事务中执行。不要按 flowId 批量绑定最新版。
UPDATE process_node_data n
JOIN ACT_RE_PROCDEF d
ON d.ID_ = '<已核验的流程定义ID>'
AND d.KEY_ = n.flow_id
AND d.TENANT_ID_ = CAST(n.tenant_id AS CHAR)
SET n.process_definition_id = d.ID_, n.del_flag = '0'
WHERE n.id = <已核验的节点记录ID>
AND n.tenant_id = <目标租户ID>
AND n.process_definition_id IS NULL;
一条旧记录如果需属于多个已验证版本,应分别插入不同 ID 的快照副本;不要反复改写同一行的版本 ID。若需从原始设计 JSON 重建节点,注意前端的 nodeName/childNode 与数据库 JSON 的 name/children 映射,并保存所有节点。无法恢复的历史版本应明确列为阻塞,不能用最新版替代。
迁移后检查所有运行实例是否有 root 快照:
SELECT DISTINCT d.ID_, d.KEY_, d.TENANT_ID_
FROM ACT_RU_EXECUTION e
JOIN ACT_RE_PROCDEF d ON d.ID_ = e.PROC_DEF_ID_
LEFT JOIN process_node_data n
ON n.process_definition_id = d.ID_
AND CAST(n.tenant_id AS CHAR) = d.TENANT_ID_
AND n.node_id = 'root' AND n.del_flag = '0'
WHERE n.id IS NULL;
该查询必须无遗漏;还需对照设计检查全部业务节点,不能仅用 root 存在代替完整性核验。新发布时程序先按当时的表单字段集合补齐默认 E 权限,再在部署事务中写快照,后续不得人工修改已发布数据。
## 回退
不能在产生新快照后直接恢复旧代码继续写流程。需要先冻结流程写入,再根据备份和期间新增实例单独制定回退方案。不要直接删除快照或批量清空版本列。
@@ -0,0 +1,6 @@
-- 升级已有数据库;全新数据库使用 0pigxx_boot.sql 或 3pigxx_flow.sql。
-- 执行前停止流程写入并备份;本脚本不猜测存量节点与流程版本的对应关系。
-- 必须完成同目录说明中的存量核对后再上线应用。
ALTER TABLE process_node_data
ADD COLUMN process_definition_id varchar(64) DEFAULT NULL COMMENT '引擎流程定义ID,发布版本快照',
ADD UNIQUE KEY uk_node_definition (tenant_id, process_definition_id, node_id);
@@ -5,11 +5,9 @@ import lombok.Data;
/**
* 流程节点记录执行人参数数据传输对象
* <p>
* 该DTO用于记录流程节点的执行人信息。当任务分配给用户或用户处理任务时,
* 创建相应的执行人记录,用于追踪任务的处理过程和审批历史。
* 包含任务执行的完整上下文信息。
* 该DTO用于记录流程节点的执行人信息。当任务分配给用户或用户处理任务时, 创建相应的执行人记录,用于追踪任务的处理过程和审批历史。 包含任务执行的完整上下文信息。
* </p>
*
*
* @author pigx
*/
@Data
@@ -18,17 +16,18 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 流程定义ID
* <p>
* 流程模板的唯一标识。
* 标识该记录所属的流程定义。
* 流程模板的唯一标识。 标识该记录所属的流程定义。
* </p>
*/
private String flowId;
/** 事件产生时绑定的引擎流程定义 ID。 */
private String processDefinitionId;
/**
* 流程实例ID
* <p>
* 运行中的流程实例唯一标识。
* 用于关联到具体的流程实例。
* 运行中的流程实例唯一标识。 用于关联到具体的流程实例。
* </p>
*/
private String processInstanceId;
@@ -36,8 +35,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 表单数据
* <p>
* JSON格式的表单数据,包含流程实例的业务数据。
* 记录任务执行时的表单内容,便于历史查看。
* JSON格式的表单数据,包含流程实例的业务数据。 记录任务执行时的表单内容,便于历史查看。
* </p>
*/
private String data;
@@ -45,8 +43,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 节点本地数据
* <p>
* 节点级别的局部数据,仅在当前节点范围内有效。
* 可能包含节点特定的配置或临时数据。
* 节点级别的局部数据,仅在当前节点范围内有效。 可能包含节点特定的配置或临时数据。
* </p>
*/
private String localData;
@@ -54,8 +51,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 节点ID
* <p>
* 流程节点的唯一标识。
* 标识任务所在的节点。
* 流程节点的唯一标识。 标识任务所在的节点。
* </p>
*/
private String nodeId;
@@ -63,8 +59,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 执行人用户ID
* <p>
* 被分配执行该任务的用户ID。
* 记录谁负责处理这个任务。
* 被分配执行该任务的用户ID。 记录谁负责处理这个任务。
* </p>
*/
private Long userId;
@@ -72,8 +67,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 执行ID
* <p>
* Flowable引擎的执行ID。
* 用于关联流程引擎中的执行实例。
* Flowable引擎的执行ID。 用于关联流程引擎中的执行实例。
* </p>
*/
private String executionId;
@@ -81,8 +75,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 任务ID
* <p>
* Flowable引擎生成的任务ID。
* 用于唯一标识一个用户任务实例。
* Flowable引擎生成的任务ID。 用于唯一标识一个用户任务实例。
* </p>
*/
private String taskId;
@@ -90,8 +83,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 审批意见
* <p>
* 用户处理任务时填写的审批意见或备注。
* 记录审批理由、建议等信息。
* 用户处理任务时填写的审批意见或备注。 记录审批理由、建议等信息。
* </p>
*/
private String approveDesc;
@@ -99,8 +91,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 节点名称
* <p>
* 流程节点的显示名称,如"部门经理审批"。
* 用于在界面上展示,提高可读性。
* 流程节点的显示名称,如"部门经理审批"。 用于在界面上展示,提高可读性。
* </p>
*/
private String nodeName;
@@ -108,11 +99,7 @@ public class ProcessNodeRecordAssignUserParamDto {
/**
* 任务类型
* <p>
* 标识任务的类型:
* - APPROVAL: 审批任务
* - COPY: 抄送任务
* - NOTIFY: 通知任务
* - CUSTOM: 自定义任务
* 标识任务的类型: - APPROVAL: 审批任务 - COPY: 抄送任务 - NOTIFY: 通知任务 - CUSTOM: 自定义任务
* </p>
*/
private String taskType;
@@ -7,9 +7,7 @@ import java.util.Map;
/**
* 任务结果数据传输对象
* <p>
* 该DTO用于返回任务的详细信息和执行结果。
* 包含任务的当前状态、节点配置、流程变量、委派状态等信息。
* 主要用于任务详情展示、任务处理页面的数据渲染。
* 该DTO用于返回任务的详细信息和执行结果。 包含任务的当前状态、节点配置、流程变量、委派状态等信息。 主要用于任务详情展示、任务处理页面的数据渲染。
* </p>
*
* @author pigx
@@ -20,9 +18,7 @@ public class TaskResultDto {
/**
* 是否为当前任务
* <p>
* 标识该任务是否为流程的当前活动任务。
* true: 任务正在等待处理
* false: 任务已完成或不是当前节点
* 标识该任务是否为流程的当前活动任务。 true: 任务正在等待处理 false: 任务已完成或不是当前节点
* </p>
*/
private Boolean currentTask;
@@ -30,18 +26,18 @@ public class TaskResultDto {
/**
* 流程定义ID
* <p>
* 任务所属的流程定义唯一标识。
* 用于获取流程模板信息。
* 任务所属的流程定义唯一标识。 用于获取流程模板信息。
* </p>
*/
private String flowId;
/** 任务实际绑定的引擎流程定义 ID。 */
private String processDefinitionId;
/**
* 任务节点配置
* <p>
* 包含任务节点的完整配置信息。
* 如节点类型、审批人设置、表单权限、条件配置等。
* 用于渲染任务处理界面。
* 包含任务节点的完整配置信息。 如节点类型、审批人设置、表单权限、条件配置等。 用于渲染任务处理界面。
* </p>
*/
private Node taskNode;
@@ -49,8 +45,7 @@ public class TaskResultDto {
/**
* 节点ID
* <p>
* 任务对应的流程节点ID。
* 用于定位任务在流程中的位置。
* 任务对应的流程节点ID。 用于定位任务在流程中的位置。
* </p>
*/
private String nodeId;
@@ -58,8 +53,7 @@ public class TaskResultDto {
/**
* 流程实例ID
* <p>
* 任务所属的流程实例唯一标识。
* 用于关联到具体的流程实例。
* 任务所属的流程实例唯一标识。 用于关联到具体的流程实例。
* </p>
*/
private String processInstanceId;
@@ -67,10 +61,7 @@ public class TaskResultDto {
/**
* 委派状态
* <p>
* 标识任务的委派状态:
* - PENDING: 待处理,委派任务等待被委派人处理
* - RESOLVED: 已解决,委派任务已完成并返回给委派人
* - null: 非委派任务
* 标识任务的委派状态: - PENDING: 待处理,委派任务等待被委派人处理 - RESOLVED: 已解决,委派任务已完成并返回给委派人 - null: 非委派任务
* </p>
*/
private String delegationState;
@@ -78,9 +69,7 @@ public class TaskResultDto {
/**
* 是否允许继续委派
* <p>
* 控制任务是否可以被继续委派给其他人。
* true: 允许当前处理人将任务委派给其他人
* false: 不允许继续委派
* 控制任务是否可以被继续委派给其他人。 true: 允许当前处理人将任务委派给其他人 false: 不允许继续委派
* </p>
*/
private Boolean delegate;
@@ -88,10 +77,7 @@ public class TaskResultDto {
/**
* 所有流程变量
* <p>
* 包含流程实例的所有变量,包括:
* - 表单数据:用户填写的业务数据
* - 系统变量:流程引擎设置的变量
* - 控制变量:用于流程控制的变量
* 包含流程实例的所有变量,包括: - 表单数据:用户填写的业务数据 - 系统变量:流程引擎设置的变量 - 控制变量:用于流程控制的变量
* 这些变量可用于任务处理和页面展示。
* </p>
*/
@@ -25,11 +25,16 @@ import java.time.LocalDateTime;
@TableName("process_node_data")
public class ProcessNodeData {
/** 已发布的引擎流程定义 ID;旧记录为空,不作为运行期快照。 */
private String processDefinitionId;
/** 所属租户。 */
private Long tenantId;
/**
* 流程实例ID
* <p>
* 关联到具体的流程实例,标识这些节点数据属于哪个流程实例。
* 对应ProcessInstanceRecord表的flowId字段。
* 关联到具体的流程实例,标识这些节点数据属于哪个流程实例。 对应ProcessInstanceRecord表的flowId字段。
* </p>
*/
@TableField("flow_id")
@@ -38,8 +43,7 @@ public class ProcessNodeData {
/**
* 节点表单数据
* <p>
* 以JSON格式存储的节点表单数据,包含用户在该节点填写的所有表单字段值。
* 数据结构根据节点配置的表单项动态变化,支持文本、数字、日期、选择等多种表单类型。
* 以JSON格式存储的节点表单数据,包含用户在该节点填写的所有表单字段值。 数据结构根据节点配置的表单项动态变化,支持文本、数字、日期、选择等多种表单类型。
* 例如:{"amount": 5000, "reason": "采购办公用品", "approver": "user123"}
* </p>
*/
@@ -49,9 +53,7 @@ public class ProcessNodeData {
/**
* 节点ID
* <p>
* 流程定义中的节点标识符,用于关联到具体的流程节点。
* 通过此ID可以确定数据属于流程的哪个环节,如发起节点、审批节点、抄送节点等。
* 与Node对象中的id字段对应。
* 流程定义中的节点标识符,用于关联到具体的流程节点。 通过此ID可以确定数据属于流程的哪个环节,如发起节点、审批节点、抄送节点等。 与Node对象中的id字段对应。
* </p>
*/
@TableField("node_id")
@@ -69,8 +71,7 @@ public class ProcessNodeData {
/**
* 逻辑删除标记
* <p>
* 用于软删除功能,删除时不会真正从数据库中移除记录,而是标记为已删除。
* 0-正常,1-已删除。保留历史数据便于审计和数据恢复。
* 用于软删除功能,删除时不会真正从数据库中移除记录,而是标记为已删除。 0-正常,1-已删除。保留历史数据便于审计和数据恢复。
* </p>
*/
@TableLogic
@@ -80,8 +81,7 @@ public class ProcessNodeData {
/**
* 创建时间
* <p>
* 记录节点数据的创建时间,通常是用户提交表单的时间。
* 自动填充,用于数据审计和流程追踪。
* 记录节点数据的创建时间,通常是用户提交表单的时间。 自动填充,用于数据审计和流程追踪。
* </p>
*/
@TableField(fill = FieldFill.INSERT)
@@ -90,8 +90,7 @@ public class ProcessNodeData {
/**
* 更新时间
* <p>
* 记录节点数据的最后更新时间,在创建和更新时自动填充。
* 用于跟踪数据的修改历史,如表单数据的补充或修正。
* 记录节点数据的最后更新时间,在创建和更新时自动填充。 用于跟踪数据的修改历史,如表单数据的补充或修正。
* </p>
*/
@TableField(fill = FieldFill.INSERT_UPDATE)
+5
View File
@@ -117,6 +117,11 @@
<groupId>com.pig4cloud</groupId>
<artifactId>pigx-common-swagger</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>test</scope>
</dependency>
<!--测试: spring boot test-->
<dependency>
<groupId>org.springframework.boot</groupId>
@@ -9,9 +9,8 @@ import com.pig4cloud.pigx.flow.entity.ProcessNodeData;
/**
* 流程节点数据管理服务接口
* <p>
* 该服务负责管理流程定义中每个节点的配置数据,包括节点的属性设置、审批人配置、
* 表单权限、条件表达式等。这些数据是流程引擎执行时的重要依据,决定了流程节点
* 的行为方式。服务提供了节点数据的保存和查询功能,支持流程的动态配置。
* 该服务负责管理流程定义中每个节点的配置数据,包括节点的属性设置、审批人配置、 表单权限、条件表达式等。这些数据是流程引擎执行时的重要依据,决定了流程节点
* 的行为方式。节点配置随发布保存为不可变快照;设计态可查询最新版,运行期必须指定引擎定义 ID。
* </p>
*
* @author Vincent
@@ -20,43 +19,41 @@ import com.pig4cloud.pigx.flow.entity.ProcessNodeData;
public interface IProcessNodeDataService extends IService<ProcessNodeData> {
/**
* 保存或更新流程节点数据
* <p>
* 保存流程节点的完整配置信息,包括节点基本属性、审批人设置、表单字段权限、
* 条件配置等。如果节点数据已存在则更新,不存在则新增。保存时会进行数据
* 验证,确保配置的合法性和完整性。该接口通常在流程设计时调用。
* </p>
*
* @param processNodeDataDto 节点数据传输对象,包含:
* - flowId: 流程定义ID
* - nodeId: 节点ID
* - nodeName: 节点名称
* - nodeType: 节点类型(审批/抄送/条件等)
* - nodeUser: 节点处理人配置
* - formPermission: 表单字段权限配置
* - conditions: 条件表达式(条件节点使用)
* - properties: 其他扩展属性
* @return R 响应结果,成功返回保存后的节点数据ID,失败返回错误信息
* 保留旧接口契约但拒绝直接修改节点配置;修改须随流程重新发布。
* @param processNodeDataDto 旧接口的节点数据,允许为 null;不读取也不持久化
* @return 说明需重新发布的失败结果,不为 null
*/
R saveNodeData(ProcessNodeDataDto processNodeDataDto);
/**
* 获取指定节点的配置数据
* <p>
* 根据流程ID和节点ID查询节点的完整配置信息。返回的数据包括节点的所有
* 配置项,可直接用于流程引擎的执行或流程设计器的回显。查询时会进行
* 权限校验,确保调用者有权限访问该流程的节点数据。
* </p>
*
* @param flowId 流程定义ID,标识节点所属的流程
* @param nodeId 节点ID,流程中节点的唯一标识
* @return R<Node> 响应结果,包含节点的完整配置信息:
* - 基本信息(ID、名称、类型等)
* - 处理人配置(固定人员、角色、部门负责人等)
* - 表单权限(各字段的可见、可编辑权限)
* - 条件配置(分支条件、跳转条件等)
* - 其他属性(超时设置、多实例配置等)
* 按当前租户查询业务流程的最新发布配置,仅供设计态使用。 运行期必须使用 getPublishedNode,避免读取其他发布版本。
* @param flowId 业务流程 key,不允许为 null
* @param nodeId 节点 ID,不允许为空
* @return 最新发布节点;没有发布定义或请求的是生成节点时,结果 data 为 null
* @throws IllegalArgumentException 已找到定义但 nodeId 为空时抛出
* @throws IllegalStateException 已发布定义缺少版本快照或业务节点快照时抛出
*/
R<Node> getNodeData(String flowId, String nodeId);
/**
* 读取指定引擎发布版本的节点快照,不回退到最新版本。
* @param processDefinitionId 引擎发布定义 ID,包含版本信息,非业务流程 key,不允许为空
* @param nodeId 节点 ID,不允许为空
* @return 节点配置;根快照存在且其中不包含该节点(如编译器生成节点)时,data 为 null
* @throws IllegalArgumentException 任一参数为空时抛出
* @throws IllegalStateException 根快照缺失,或根树中存在该业务节点但独立快照缺失时抛出
* @throws cn.hutool.json.JSONException 已保存的节点 JSON 无法解析时抛出
*/
R<Node> getPublishedNode(String processDefinitionId, String nodeId);
/**
* 递归插入发布节点快照,使用引擎定义的 key、ID 和租户,不覆盖既有数据。 调用方必须在同一事务中部署引擎定义;此方法参与该事务,失败时整体回滚。
* @param definition 刚部署的引擎定义,不允许为 null,租户 ID 须为 Long 格式
* @param root 发起节点树,不允许为 null,ID 必须为 root;空节点和空 ID 子树跳过
* @throws IllegalArgumentException 定义或根节点无效、节点 ID 重复或租户 ID 非数字时抛出
* @throws IllegalStateException 任一节点插入数量不为一时抛出
* @throws org.springframework.dao.DataAccessException 插入失败或违反版本唯一约束时抛出
*/
void savePublishedNodes(org.flowable.engine.repository.ProcessDefinition definition, Node root);
}
@@ -14,17 +14,18 @@ import com.pig4cloud.pigx.flow.service.IProcessService;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.springframework.stereotype.Service;
import org.flowable.engine.HistoryService;
import org.flowable.engine.RuntimeService;
import com.pig4cloud.pigx.flow.support.utils.NodeUtil;
import java.util.Map;
import java.util.Objects;
/**
* 流程抄送数据服务实现类
* <p>
* 该类负责处理流程抄送相关的业务逻辑,主要功能包括:
* 1. 管理流程抄送记录的增删改查
* 2. 查询抄送详情,包括流程表单、表单数据和权限信息
* 3. 为被抄送人提供查看流程执行情况的能力
*
* 该类负责处理流程抄送相关的业务逻辑,主要功能包括: 1. 管理流程抄送记录的增删改查 2. 查询抄送详情,包括流程表单、表单数据和权限信息 3.
* 为被抄送人提供查看流程执行情况的能力
*
* 抄送功能允许流程在执行过程中通知相关人员,被抄送人可以查看但不参与流程审批
* </p>
*
@@ -35,56 +36,71 @@ import java.util.Map;
@RequiredArgsConstructor
public class ProcessCopyServiceImpl extends ServiceImpl<ProcessCopyMapper, ProcessCopy> implements IProcessCopyService {
private final ObjectMapper objectMapper;
private final ObjectMapper objectMapper;
private final IProcessService processService;
private final IProcessService processService;
private final IProcessNodeDataService nodeDataService;
private final IProcessNodeDataService nodeDataService;
/**
* 查询单个抄送详细信息
* <p>
* 根据抄送记录ID查询完整的抄送详情,包括:
* 1. 流程定义信息(process)
* 2. 表单项定义(formItems)
* 3. 表单数据(formData)- 流程实例的实际填写数据
* 4. 表单权限(formPerms)- 基于当前节点的表单字段权限设置
*
* 该方法主要用于抄送人查看被抄送的流程详情
* </p>
*
* @param id 抄送记录ID
* @return R 包含流程详情的响应对象,包括process、formItems、formData、formPerms
* @throws Exception 当解析JSON数据失败时抛出异常
*/
@SneakyThrows
@Override
public R querySingleDetail(long id) {
// 获取抄送记录
ProcessCopy processCopy = this.getById(id);
String flowId = processCopy.getFlowId();
// 查询流程定义
Process oaForms = processService.getByFlowId(flowId);
if (oaForms == null) {
return R.failed("流程不存在");
}
// 获取表单数据和节点信息
String formData = processCopy.getFormData();
String nodeId = processCopy.getNodeId();
private final HistoryService historyService;
// 获取节点的表单权限配置
Node node = nodeDataService.getNodeData(flowId, nodeId).getData();
Map<String, String> formPerms = node.getFormPerms();
private final RuntimeService runtimeService;
// 构建返回数据
Dict set = Dict.create()
.set("process", oaForms.getProcess()) // 流程定义
.set("formItems", oaForms.getFormItems()) // 表单项定义
.set("formData", formData) // 表单数据
.set("formPerms", formPerms); // 表单权限
return R.ok(set);
}
/**
* 查询单个抄送详细信息
* <p>
* 根据抄送记录ID查询完整的抄送详情,包括: 1. 流程定义信息(process) 2. 表单项定义(formItems) 3. 表单数据(formData)-
* 流程实例的实际填写数据 4. 表单权限(formPerms)- 基于当前节点的表单字段权限设置
*
* 该方法主要用于抄送人查看被抄送的流程详情
* </p>
* @param id 抄送记录ID
* @return R 包含流程详情的响应对象,包括process、formItems、formData、formPerms
* @throws Exception 当解析JSON数据失败时抛出异常
*/
@SneakyThrows
@Override
public R querySingleDetail(long id) {
// 获取抄送记录
ProcessCopy processCopy = this.getById(id);
String flowId = processCopy.getFlowId();
// 查询流程定义
Process oaForms = processService.getByFlowId(flowId);
if (oaForms == null) {
return R.failed("流程不存在");
}
// 获取表单数据和节点信息
String formData = processCopy.getFormData();
String nodeId = processCopy.getNodeId();
// 获取节点的表单权限配置
var running = runtimeService.createProcessInstanceQuery()
.processInstanceId(processCopy.getProcessInstanceId())
.singleResult();
String definitionId;
if (Objects.nonNull(running)) {
definitionId = running.getProcessDefinitionId();
}
else {
var historic = historyService.createHistoricProcessInstanceQuery()
.processInstanceId(processCopy.getProcessInstanceId())
.singleResult();
if (Objects.isNull(historic)) {
return R.failed("流程实例历史不存在,无法确定节点配置版本");
}
definitionId = historic.getProcessDefinitionId();
}
Node node = nodeDataService.getPublishedNode(definitionId, nodeId).getData();
// 构建返回数据
Dict set = Dict.create()
.set("process", oaForms.getProcess()) // 流程定义
.set("formItems", oaForms.getFormItems()) // 表单项定义
.set("formData", formData) // 表单数据
.set("formPerms", NodeUtil.publishedFormPermissions(node, oaForms.getFormItems())); // 表单权限
return R.ok(set);
}
}
@@ -1,74 +1,144 @@
package com.pig4cloud.pigx.flow.service.impl;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.pig4cloud.pigx.common.core.util.R;
import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder;
import com.pig4cloud.pigx.flow.dto.Node;
import com.pig4cloud.pigx.flow.dto.ProcessNodeDataDto;
import com.pig4cloud.pigx.flow.entity.ProcessNodeData;
import com.pig4cloud.pigx.flow.mapper.ProcessNodeDataMapper;
import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
import lombok.RequiredArgsConstructor;
import org.flowable.engine.RepositoryService;
import org.flowable.engine.repository.ProcessDefinition;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.Optional;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
/**
* 流程节点数据服务实现类
* <p>
* 该类负责管理流程节点的详细数据,主要功能包括: 1. 保存流程节点的配置数据(审批人、表单权限、条件等) 2. 查询指定节点的详细配置信息 3.
* 为流程执行提供节点配置数据支持
*
* 节点数据以JSON格式存储,包含节点类型、审批人配置、表单权限等信息
* </p>
*
* @author Vincent
* @since 2023-05-07
*/
/** 已发布节点配置按引擎定义 ID 隔离,不允许原地修改。 */
@Service
@RequiredArgsConstructor
public class ProcessNodeDataServiceImpl extends ServiceImpl<ProcessNodeDataMapper, ProcessNodeData>
implements IProcessNodeDataService {
private final RepositoryService repositoryService;
/**
* 保存流程节点数据
* <p>
* 将流程节点的配置信息保存到数据库。节点数据包括: - 节点类型(审批、抄送、条件等) - 审批人配置 - 表单权限设置 - 条件表达式等
* </p>
* @param processNodeDataDto 流程节点数据DTO,包含flowId、nodeId和data(JSON格式的节点配置)
* @return R 保存结果
* {@inheritDoc}
*/
@Override
public R saveNodeData(ProcessNodeDataDto processNodeDataDto) {
ProcessNodeData processNodeData = BeanUtil.copyProperties(processNodeDataDto, ProcessNodeData.class);
baseMapper.delete(Wrappers.<ProcessNodeData>lambdaQuery()
.eq(ProcessNodeData::getNodeId, processNodeData.getNodeId())
.eq(ProcessNodeData::getFlowId, processNodeData.getFlowId()));
baseMapper.insert(processNodeData);
return R.ok();
return R.failed("不允许直接修改已发布节点配置,请编辑流程后重新发布");
}
/**
* 获取节点数据
* <p>
* 根据流程ID和节点ID查询节点的详细配置信息。 节点数据存储为JSON格式,查询后会自动转换为Node对象
* </p>
* @param flowId 流程ID
* @param nodeId 节点ID
* @return R<Node> 包含节点配置的Node对象,如果节点不存在则返回null
* {@inheritDoc}
*/
@Override
public R<Node> getNodeData(String flowId, String nodeId) {
Optional<ProcessNodeData> processNodeDataOptional = this.lambdaQuery()
.eq(ProcessNodeData::getFlowId, flowId)
ProcessDefinition definition = repositoryService.createProcessDefinitionQuery()
.processDefinitionKey(flowId)
.processDefinitionTenantId(String.valueOf(TenantContextHolder.getTenantId()))
.latestVersion()
.singleResult();
if (Objects.isNull(definition)) {
return R.ok(null);
}
return getPublishedNode(definition.getId(), nodeId);
}
/**
* {@inheritDoc}
*/
@Override
public R<Node> getPublishedNode(String processDefinitionId, String nodeId) {
if (StrUtil.hasBlank(processDefinitionId, nodeId)) {
throw new IllegalArgumentException("查询已发布节点必须指定流程定义 ID 和节点 ID");
}
ProcessNodeData data = this.lambdaQuery()
.eq(ProcessNodeData::getProcessDefinitionId, processDefinitionId)
.eq(ProcessNodeData::getNodeId, nodeId)
.oneOpt();
.one();
if (Objects.nonNull(data)) {
return R.ok(JSONUtil.toBean(data.getData(), Node.class));
}
// start、网关汇聚及审批后的服务任务由编译器生成,没有业务节点配置。
ProcessNodeData root = this.lambdaQuery()
.eq(ProcessNodeData::getProcessDefinitionId, processDefinitionId)
.eq(ProcessNodeData::getNodeId, "root")
.one();
if (Objects.isNull(root)) {
throw new IllegalStateException("流程定义 " + processDefinitionId + " 缺少节点版本快照,请先核对并迁移存量配置");
}
if (containsNode(JSONUtil.toBean(root.getData(), Node.class), nodeId)) {
throw new IllegalStateException("流程定义 " + processDefinitionId + " 缺少节点版本快照:" + nodeId);
}
return R.ok(null);
}
Node node = processNodeDataOptional
.map(processNodeData -> JSONUtil.toBean(processNodeData.getData(), Node.class))
.orElse(null);
private boolean containsNode(Node node, String nodeId) {
if (Objects.isNull(node)) {
return false;
}
if (nodeId.equals(node.getId())) {
return true;
}
if (Objects.nonNull(node.getConditionNodes())) {
for (Node branch : node.getConditionNodes()) {
if (containsNode(branch, nodeId)) {
return true;
}
}
}
return containsNode(node.getChildren(), nodeId);
}
return R.ok(node);
/**
* {@inheritDoc}
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void savePublishedNodes(ProcessDefinition definition, Node root) {
if (Objects.isNull(definition) || Objects.isNull(root) || !"root".equals(root.getId())) {
throw new IllegalArgumentException("发布流程缺少定义或发起节点");
}
List<ProcessNodeData> nodes = new ArrayList<>();
collectNodes(definition, root, new HashSet<>(), nodes);
// 数据库唯一约束防止同一版本被重复写入,禁止删除或覆盖已有快照。
for (ProcessNodeData node : nodes) {
if (baseMapper.insert(node) != 1) {
throw new IllegalStateException("保存节点版本快照失败");
}
}
}
private void collectNodes(ProcessDefinition definition, Node node, Set<String> ids, List<ProcessNodeData> nodes) {
if (Objects.isNull(node) || StrUtil.isBlank(node.getId())) {
return;
}
if (!ids.add(node.getId())) {
throw new IllegalArgumentException("流程节点 ID 重复:" + node.getId());
}
ProcessNodeData data = new ProcessNodeData();
data.setProcessDefinitionId(definition.getId());
data.setFlowId(definition.getKey());
data.setTenantId(Long.valueOf(definition.getTenantId()));
data.setNodeId(node.getId());
data.setData(JSONUtil.toJsonStr(node));
nodes.add(data);
if (Objects.nonNull(node.getConditionNodes())) {
for (Node branch : node.getConditionNodes()) {
collectNodes(definition, branch, ids, nodes);
}
}
collectNodes(definition, node.getChildren(), ids, nodes);
}
}
@@ -26,6 +26,7 @@ import com.pig4cloud.pigx.flow.entity.ProcessStarter;
import com.pig4cloud.pigx.flow.event.FlowStatusEventPublisher;
import com.pig4cloud.pigx.flow.mapper.ProcessMapper;
import com.pig4cloud.pigx.flow.service.IProcessService;
import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
import com.pig4cloud.pigx.flow.service.IProcessStarterService;
import com.pig4cloud.pigx.flow.support.utils.ModelUtil;
import com.pig4cloud.pigx.flow.support.utils.NodeUtil;
@@ -39,6 +40,7 @@ import org.flowable.bpmn.model.BpmnModel;
import org.flowable.engine.RepositoryService;
import org.flowable.engine.repository.Deployment;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.Date;
import java.util.List;
@@ -60,270 +62,272 @@ import java.util.Objects;
@RequiredArgsConstructor
public class ProcessServiceImpl extends ServiceImpl<ProcessMapper, Process> implements IProcessService {
private final IProcessStarterService processStarterService;
private final IProcessStarterService processStarterService;
private final RepositoryService repositoryService;
private final RepositoryService repositoryService;
private final ObjectMapper objectMapper;
private final IProcessNodeDataService processNodeDataService;
private final Sequence flowSequence;
private final ObjectMapper objectMapper;
private final FlowStatusEventPublisher flowStatusEventPublisher;
private final Sequence flowSequence;
/**
* 获取流程详细信息
* <p>
* 根据流程ID获取完整的流程配置信息,包括: - 流程基本信息 - 表单配置(自动生成序列号字段的值) - 需要发起人选择执行人的节点列表 - 表单权限配置
* </p>
*
* @param flowId 流程定义ID
* @return 包含流程详细信息的响应对象
*/
@Override
public R<ProcessVO> getDetail(String flowId) {
// 获取流程详情
ProcessVO processVO = this.getProcessVO(flowId);
return R.ok(processVO);
}
private final FlowStatusEventPublisher flowStatusEventPublisher;
/**
* 获取流程视图对象
* <p>
* 内部方法,用于构建流程的视图对象。主要处理: - 解析流程节点结构 - 查找需要发起人选择的节点 - 处理序列号类型的表单字段,自动生成下一个序号 -
* 提取起始节点的表单权限配置
* </p>
*
* @param flowId 流程定义ID
* @return 流程视图对象,包含所有展示所需的信息
*/
@SneakyThrows
private ProcessVO getProcessVO(String flowId) {
/**
* 获取流程详细信息
* <p>
* 根据流程ID获取完整的流程配置信息,包括: - 流程基本信息 - 表单配置(自动生成序列号字段的值) - 需要发起人选择执行人的节点列表 - 表单权限配置
* </p>
* @param flowId 流程定义ID
* @return 包含流程详细信息的响应对象
*/
@Override
public R<ProcessVO> getDetail(String flowId) {
// 获取流程详情
ProcessVO processVO = this.getProcessVO(flowId);
return R.ok(processVO);
}
Process oaForms = getByFlowId(flowId);
String process = oaForms.getProcess();
String formItems = oaForms.getFormItems();
Node startNode = objectMapper.readValue(process, new TypeReference<>() {
});
/**
* 获取流程视图对象
* <p>
* 内部方法,用于构建流程的视图对象。主要处理: - 解析流程节点结构 - 查找需要发起人选择的节点 - 处理序列号类型的表单字段,自动生成下一个序号 -
* 提取起始节点的表单权限配置
* </p>
* @param flowId 流程定义ID
* @return 流程视图对象,包含所有展示所需的信息
*/
@SneakyThrows
private ProcessVO getProcessVO(String flowId) {
List<String> selectUserNodeId = NodeUtil.selectUserNodeId(startNode);
Process oaForms = getByFlowId(flowId);
String process = oaForms.getProcess();
String formItems = oaForms.getFormItems();
Node startNode = objectMapper.readValue(process, new TypeReference<>() {
});
ProcessVO processVO = BeanUtil.copyProperties(oaForms, ProcessVO.class);
processVO.setSelectUserNodeId(selectUserNodeId);
List<String> selectUserNodeId = NodeUtil.selectUserNodeId(startNode);
JSONArray formItemJsonArray = JSONUtil.parseArray(formItems);
// 特殊字段复制
for (Object object : formItemJsonArray) {
if (object instanceof JSONObject) {
JSONObject formVO = (JSONObject) object;
JSONObject props = formVO.getJSONObject(FormItemVO.Fields.props);
if (Objects.nonNull(props)
&& FormTypeEnum.SEQUENCE.getType().equals(props.getStr(FormItemVO.Fields.type))) {
String nextNo = flowSequence.nextNo();
props.set(FormItemVO.Props.Fields.value, nextNo);
formVO.set(FormItemVO.Props.Fields.value, nextNo);
}
}
}
ProcessVO processVO = BeanUtil.copyProperties(oaForms, ProcessVO.class);
processVO.setSelectUserNodeId(selectUserNodeId);
processVO.setFormItems(formItemJsonArray.toStringPretty());
processVO.setFormPerms(startNode.getFormPerms());
return processVO;
}
JSONArray formItemJsonArray = JSONUtil.parseArray(formItems);
// 特殊字段复制
for (Object object : formItemJsonArray) {
if (object instanceof JSONObject) {
JSONObject formVO = (JSONObject) object;
JSONObject props = formVO.getJSONObject(FormItemVO.Fields.props);
if (Objects.nonNull(props)
&& FormTypeEnum.SEQUENCE.getType().equals(props.getStr(FormItemVO.Fields.type))) {
String nextNo = flowSequence.nextNo();
props.set(FormItemVO.Props.Fields.value, nextNo);
formVO.set(FormItemVO.Props.Fields.value, nextNo);
}
}
}
/**
* 根据流程ID查询流程定义
* <p>
* 通过流程的唯一标识符查询流程定义实体。 注意:一个流程可能有多个版本,此方法返回当前有效版本。
* </p>
*
* @param flowId 流程定义ID
* @return 流程定义实体,如果不存在返回null
*/
@Override
public Process getByFlowId(String flowId) {
return this.lambdaQuery().eq(Process::getFlowId, flowId).one();
}
processVO.setFormItems(formItemJsonArray.toStringPretty());
processVO.setFormPerms(startNode.getFormPerms());
return processVO;
}
/**
* 根据流程ID更新流程定义
* <p>
* 更新流程的配置信息,包括表单、节点、权限等。 更新不会影响已经运行的流程实例。
* </p>
*
* @param process 包含更新信息的流程实体
*/
@Override
public void updateByFlowId(Process process) {
this.lambdaUpdate().eq(Process::getFlowId, process.getFlowId()).update(process);
}
/**
* 根据流程ID查询流程定义
* <p>
* 通过流程的唯一标识符查询流程定义实体。 注意:一个流程可能有多个版本,此方法返回当前有效版本。
* </p>
* @param flowId 流程定义ID
* @return 流程定义实体,如果不存在返回null
*/
@Override
public Process getByFlowId(String flowId) {
return this.lambdaQuery().eq(Process::getFlowId, flowId).one();
}
/**
* 隐藏流程定义
* <p>
* 将流程标记为隐藏状态,隐藏的流程: - 不在流程列表中显示 - 不能创建新的流程实例 - 已有的流程实例可以继续执行
* </p>
*
* @param flowId 要隐藏的流程ID
*/
@Override
public void hide(String flowId) {
this.lambdaUpdate()
.set(Process::getHidden, YesNoEnum.YES.getCode())
.eq(Process::getFlowId, flowId)
.update(new Process());
}
/**
* 根据流程ID更新流程定义
* <p>
* 更新流程的配置信息,包括表单、节点、权限等。 更新不会影响已经运行的流程实例。
* </p>
* @param process 包含更新信息的流程实体
*/
@Override
public void updateByFlowId(Process process) {
this.lambdaUpdate().eq(Process::getFlowId, process.getFlowId()).update(process);
}
/**
* 创建新的流程定义
* <p>
* 创建流程的完整流程: 1. 调用流程引擎创建BPMN模型 2. 如果是更新操作,隐藏旧版本 3. 保存流程定义到数据库 4. 设置流程发起人权限 5.
* 生成流程使用范围描述
* </p>
*
* @param process 流程定义实体,包含流程的所有配置信息
* @return 创建结果,成功返回流程ID,失败返回错误信息
*/
@SneakyThrows
@Override
public R create(Process process) {
String flowId;
if (StrUtil.isNotBlank(process.getFlowId())) {
flowId = process.getFlowId();
} else {
flowId = "P" + DateUtil.format(new Date(), DatePattern.PURE_DATE_PATTERN)
+ RandomUtil.randomString(5).toUpperCase();
}
/**
* 隐藏流程定义
* <p>
* 将流程标记为隐藏状态,隐藏的流程: - 不在流程列表中显示 - 不能创建新的流程实例 - 已有的流程实例可以继续执行
* </p>
* @param flowId 要隐藏的流程ID
*/
@Override
public void hide(String flowId) {
this.lambdaUpdate()
.set(Process::getHidden, YesNoEnum.YES.getCode())
.eq(Process::getFlowId, flowId)
.update(new Process());
}
BpmnModel bpmnModel = ModelUtil.buildBpmnModel(objectMapper.readValue(process.getProcess(), Node.class),
process.getName(), flowId);
{
byte[] bpmnBytess = new BpmnXMLConverter().convertToXML(bpmnModel);
String filename = "/tmp/flowable-deployment/" + flowId + ".bpmn20.xml";
log.debug("部署时的模型文件:{}", filename);
FileUtil.writeBytes(bpmnBytess, filename);
}
/**
* 创建新的流程定义
* <p>
* 创建流程的完整流程: 1. 调用流程引擎创建BPMN模型 2. 如果是更新操作,隐藏旧版本 3. 保存流程定义到数据库 4. 设置流程发起人权限 5.
* 生成流程使用范围描述
* </p>
* @param process 流程定义实体,包含流程的所有配置信息
* @return 创建结果,成功返回流程ID,失败返回错误信息
*/
@SneakyThrows
@Override
@Transactional(rollbackFor = Exception.class)
public R create(Process process) {
String flowId;
if (StrUtil.isNotBlank(process.getFlowId())) {
flowId = process.getFlowId();
}
else {
flowId = "P" + DateUtil.format(new Date(), DatePattern.PURE_DATE_PATTERN)
+ RandomUtil.randomString(5).toUpperCase();
}
Process oldProcess = this.getByFlowId(flowId);
Node publishedRoot = objectMapper.readValue(process.getProcess(), Node.class);
NodeUtil.snapshotFormPermissions(publishedRoot, process.getFormItems());
BpmnModel bpmnModel = ModelUtil.buildBpmnModel(publishedRoot, process.getName(), flowId);
{
byte[] bpmnBytess = new BpmnXMLConverter().convertToXML(bpmnModel);
String filename = "/tmp/flowable-deployment/" + flowId + ".bpmn20.xml";
log.debug("部署时的模型文件:{}", filename);
FileUtil.writeBytes(bpmnBytess, filename);
}
Deployment deployment = repositoryService.createDeployment()
.addBpmnModel(StrUtil.format("{}.bpmn20.xml", "pig"), bpmnModel)
.tenantId(TenantContextHolder.getTenantId().toString())
.deploy();
Process oldProcess = this.getByFlowId(flowId);
NodeUser nodeUser = objectMapper.readValue(process.getAdminList(), new TypeReference<List<NodeUser>>() {
}).get(0);
Deployment deployment = repositoryService.createDeployment()
.addBpmnModel(StrUtil.format("{}.bpmn20.xml", "pig"), bpmnModel)
.tenantId(TenantContextHolder.getTenantId().toString())
.deploy();
// 更新流程
if (StrUtil.isNotBlank(process.getFlowId()) && Objects.nonNull(oldProcess)) {
this.hide(process.getFlowId());
// 修改所有的管理员
this.lambdaUpdate()
.set(Process::getAdminId, nodeUser.getId())
.eq(Process::getUniqueId, oldProcess.getUniqueId())
.update(new Process());
var definition = repositoryService.createProcessDefinitionQuery()
.deploymentId(deployment.getId())
.singleResult();
processNodeDataService.savePublishedNodes(definition, publishedRoot);
}
NodeUser nodeUser = objectMapper.readValue(process.getAdminList(), new TypeReference<List<NodeUser>>() {
}).get(0);
Node startNode = objectMapper.readValue(process.getProcess(), Node.class);
// 更新流程
if (StrUtil.isNotBlank(process.getFlowId()) && Objects.nonNull(oldProcess)) {
this.hide(process.getFlowId());
// 修改所有的管理员
this.lambdaUpdate()
.set(Process::getAdminId, nodeUser.getId())
.eq(Process::getUniqueId, oldProcess.getUniqueId())
.update(new Process());
List<NodeUser> nodeUserList = startNode.getNodeUserList();
}
StringBuilder stringBuilder = new StringBuilder();
if (CollUtil.isNotEmpty(nodeUserList)) {
int index = 0;
Node startNode = objectMapper.readValue(process.getProcess(), Node.class);
for (NodeUser user : nodeUserList) {
if (index > 0) {
stringBuilder.append(",");
}
stringBuilder.append(user.getName());
index++;
if (index > 5) {
break;
}
List<NodeUser> nodeUserList = startNode.getNodeUserList();
}
}
Process p = new Process();
StringBuilder stringBuilder = new StringBuilder();
if (CollUtil.isNotEmpty(nodeUserList)) {
int index = 0;
if (Objects.nonNull(oldProcess)) {
p = oldProcess;
}
for (NodeUser user : nodeUserList) {
if (index > 0) {
stringBuilder.append(",");
}
stringBuilder.append(user.getName());
index++;
if (index > 5) {
break;
}
p.setFlowId(flowId);
p.setName(process.getName());
p.setLogo(process.getLogo());
p.setSettings(process.getSettings());
p.setGroupId(process.getGroupId());
p.setFormConfig(process.getFormConfig());
p.setFormItems(process.getFormItems());
p.setProcess(process.getProcess());
p.setRemark(process.getRemark());
p.setSort(0);
p.setHidden("0");
p.setStop("0");
p.setAdminId(nodeUser.getId());
p.setUniqueId(deployment.getId());
p.setAdminList(process.getAdminList());
p.setRangeShow(stringBuilder.toString());
this.saveOrUpdate(p);
}
}
Process p = new Process();
// 删除原有发起范围
processStarterService.remove(Wrappers.<ProcessStarter>lambdaQuery().eq(ProcessStarter::getProcessId, p.getId()));
if (Objects.nonNull(oldProcess)) {
p = oldProcess;
}
if (Objects.isNull(nodeUserList)) {
return R.ok();
}
for (NodeUser nodeUserDto : nodeUserList) {
ProcessStarter processStarter = new ProcessStarter();
processStarter.setProcessId(p.getId());
processStarter.setTypeId(nodeUserDto.getId());
processStarter.setType(nodeUserDto.getType());
processStarterService.save(processStarter);
}
p.setFlowId(flowId);
p.setName(process.getName());
p.setLogo(process.getLogo());
p.setSettings(process.getSettings());
p.setGroupId(process.getGroupId());
p.setFormConfig(process.getFormConfig());
p.setFormItems(process.getFormItems());
p.setProcess(process.getProcess());
p.setRemark(process.getRemark());
p.setSort(0);
p.setHidden("0");
p.setStop("0");
p.setAdminId(nodeUser.getId());
p.setUniqueId(deployment.getId());
p.setAdminList(process.getAdminList());
p.setRangeShow(stringBuilder.toString());
this.saveOrUpdate(p);
return R.ok();
}
// 删除原有发起范围
processStarterService
.remove(Wrappers.<ProcessStarter>lambdaQuery().eq(ProcessStarter::getProcessId, p.getId()));
/**
* 更新流程状态或分组
* <p>
* 支持三种操作类型: - stop:停用流程(设置stop=1) - delete:删除流程(逻辑删除,设置hidden=1) - 其他:启用流程(设置stop=0)
* 同时支持修改流程所属分组
* </p>
*
* @param flowId 流程定义ID
* @param type 操作类型:stop(停用)、delete(删除)、其他(启用)
* @param groupId 流程分组ID,可选参数
* @return 操作结果
*/
@Override
public R update(String flowId, String type, Long groupId) {
Process process = new Process();
process.setFlowId(flowId);
process.setStop("stop".equals(type) ? "1" : "0");
process.setHidden("delete".equals(type) ? "1" : "0");
process.setGroupId(groupId);
this.updateByFlowId(process);
return R.ok();
}
if (Objects.isNull(nodeUserList)) {
return R.ok();
}
for (NodeUser nodeUserDto : nodeUserList) {
ProcessStarter processStarter = new ProcessStarter();
processStarter.setProcessId(p.getId());
processStarter.setTypeId(nodeUserDto.getId());
processStarter.setType(nodeUserDto.getType());
processStarterService.save(processStarter);
}
/**
* 验证流程ID是否已注册流程状态事件处理
* <p>
* 通过 Redis Set 查询 flowId 是否已被任意微服务模块注册,
* 支持跨服务校验。
* </p>
*
* @param flowId 流程定义ID
* @return 验证结果:true表示存在,false表示不存在
*/
@Override
public R<Boolean> validateFlowId(String flowId) {
if (StrUtil.isBlank(flowId)) {
return R.ok(false);
}
return R.ok(flowStatusEventPublisher.isFlowIdRegistered(flowId));
}
return R.ok();
}
/**
* 更新流程状态或分组
* <p>
* 支持三种操作类型: - stop:停用流程(设置stop=1) - delete:删除流程(逻辑删除,设置hidden=1) - 其他:启用流程(设置stop=0)
* 同时支持修改流程所属分组
* </p>
* @param flowId 流程定义ID
* @param type 操作类型:stop(停用)、delete(删除)、其他(启用)
* @param groupId 流程分组ID,可选参数
* @return 操作结果
*/
@Override
public R update(String flowId, String type, Long groupId) {
Process process = new Process();
process.setFlowId(flowId);
process.setStop("stop".equals(type) ? "1" : "0");
process.setHidden("delete".equals(type) ? "1" : "0");
process.setGroupId(groupId);
this.updateByFlowId(process);
return R.ok();
}
/**
* 验证流程ID是否已注册流程状态事件处理
* <p>
* 通过 Redis Set 查询 flowId 是否已被任意微服务模块注册, 支持跨服务校验。
* </p>
* @param flowId 流程定义ID
* @return 验证结果:true表示存在,false表示不存在
*/
@Override
public R<Boolean> validateFlowId(String flowId) {
if (StrUtil.isBlank(flowId)) {
return R.ok(false);
}
return R.ok(flowStatusEventPublisher.isFlowIdRegistered(flowId));
}
}
@@ -20,14 +20,10 @@ import org.flowable.task.service.impl.persistence.entity.TaskEntityImpl;
/**
* 审批任务创建监听器
* <p>
* Flowable任务监听器实现,用于处理审批任务创建时的特殊逻辑。
* 主要功能是处理没有指定执行人的任务,根据节点配置的"无人处理策略"自动执行相应操作。
* Flowable任务监听器实现,用于处理审批任务创建时的特殊逻辑。 主要功能是处理没有指定执行人的任务,根据节点配置的"无人处理策略"自动执行相应操作。
* <p>
* 支持的无人处理策略:
* 1. 自动通过:任务自动完成并通过
* 2. 指派给管理员:将任务分配给流程管理员
* 3. 指派给特定用户:将任务分配给配置中指定的用户
* 4. 自动拒绝:任务自动完成并拒绝
* 支持的无人处理策略: 1. 自动通过:任务自动完成并通过 2. 指派给管理员:将任务分配给流程管理员 3. 指派给特定用户:将任务分配给配置中指定的用户 4.
* 自动拒绝:任务自动完成并拒绝
* <p>
* 该监听器需要在流程定义的用户任务节点中配置,通过taskListener扩展属性指定。
*
@@ -39,9 +35,7 @@ public class ApprovalCreateListener implements TaskListener {
/**
* 任务创建事件处理方法
* <p>
* 当任务被创建时触发,检查任务是否有执行人,如果没有则根据配置的策略自动处理。
* 这个机制解决了流程执行过程中可能出现的无人处理任务的问题。
*
* 当任务被创建时触发,检查任务是否有执行人,如果没有则根据配置的策略自动处理。 这个机制解决了流程执行过程中可能出现的无人处理任务的问题。
* @param delegateTask 委派的任务对象,包含任务的所有信息和上下文
*/
@Override
@@ -72,7 +66,7 @@ public class ApprovalCreateListener implements TaskListener {
IRemoteService remoteService = SpringUtil.getBean(IRemoteService.class);
// 查询节点原始配置数据,获取无人处理策略
Node node = processNodeDataService.getNodeData(flowId, nodeId).getData();
Node node = processNodeDataService.getPublishedNode(processDefinitionId, nodeId).getData();
Nobody nobody = node.getNobody(); // 获取无人处理配置
String handler = nobody.getHandler(); // 获取无人处理的处理方式
@@ -75,7 +75,7 @@ public class FlowProcessEventListener implements FlowableEventListener {
String flowId = NodeUtil.getFlowId(processDefinitionId);
// 查询节点原始数据
Node node = processNodeDataService.getNodeData(flowId, activityId).getData();
Node node = processNodeDataService.getPublishedNode(processDefinitionId, activityId).getData();
// 构建节点开始记录参数并发送事件
ProcessNodeRecordParamDto processNodeRecordParamDto = new ProcessNodeRecordParamDto();
@@ -217,6 +217,7 @@ public class FlowProcessEventListener implements FlowableEventListener {
// 构建任务完成记录参数并发送事件
ProcessNodeRecordAssignUserParamDto processNodeRecordAssignUserParamDto = new ProcessNodeRecordAssignUserParamDto();
processNodeRecordAssignUserParamDto.setFlowId(flowId);
processNodeRecordAssignUserParamDto.setProcessDefinitionId(processDefinitionId);
processNodeRecordAssignUserParamDto.setProcessInstanceId(processInstanceId);
processNodeRecordAssignUserParamDto
.setData(objectMapper.writeValueAsString(taskService.getVariables(task.getId())));
@@ -258,28 +259,29 @@ public class FlowProcessEventListener implements FlowableEventListener {
// 构建任务分配记录参数并发送事件
ProcessNodeRecordAssignUserParamDto processNodeRecordAssignUserParamDto = new ProcessNodeRecordAssignUserParamDto();
processNodeRecordAssignUserParamDto.setFlowId(flowId);
processNodeRecordAssignUserParamDto.setProcessDefinitionId(processDefinitionId);
processNodeRecordAssignUserParamDto.setProcessInstanceId(processInstanceId);
processNodeRecordAssignUserParamDto.setNodeId(taskDefinitionKey);
processNodeRecordAssignUserParamDto.setUserId(Long.parseLong(task.getAssignee()));
processNodeRecordAssignUserParamDto.setTaskId(task.getId());
processNodeRecordAssignUserParamDto.setNodeName(task.getName());
processNodeRecordAssignUserParamDto.setNodeRecordId(
Convert.toLong(task.getVariable(ProcessInstanceConstant.VAR_NODE_RECORD_ID), null));
processNodeRecordAssignUserParamDto.setAssignmentOrigin(
Convert.toStr(task.getVariable(ProcessInstanceConstant.VAR_ASSIGNMENT_ORIGIN)));
processNodeRecordAssignUserParamDto.setInitialUserId(
Convert.toLong(task.getVariable(ProcessInstanceConstant.VAR_INITIAL_USER_ID), null));
processNodeRecordAssignUserParamDto
.setNodeRecordId(Convert.toLong(task.getVariable(ProcessInstanceConstant.VAR_NODE_RECORD_ID), null));
processNodeRecordAssignUserParamDto
.setAssignmentOrigin(Convert.toStr(task.getVariable(ProcessInstanceConstant.VAR_ASSIGNMENT_ORIGIN)));
processNodeRecordAssignUserParamDto
.setInitialUserId(Convert.toLong(task.getVariable(ProcessInstanceConstant.VAR_INITIAL_USER_ID), null));
processNodeRecordAssignUserParamDto.setAddSignRecordId(
Convert.toLong(task.getVariable(ProcessInstanceConstant.VAR_ADD_SIGN_RECORD_ID), null));
// 根据委派状态设置任务类型
String assignmentOperation = Convert
.toStr(task.getVariableLocal(ProcessInstanceConstant.VAR_ASSIGNMENT_OPERATION));
.toStr(task.getVariableLocal(ProcessInstanceConstant.VAR_ASSIGNMENT_OPERATION));
processNodeRecordAssignUserParamDto
.setTaskType(StrUtil.isNotBlank(assignmentOperation) ? assignmentOperation
: (StrUtil.equals(DelegationState.PENDING.toString(), delegationStateString) ? "DELEGATION"
: (StrUtil.equals(DelegationState.RESOLVED.toString(), delegationStateString) ? "RESOLVED"
: StrUtil.EMPTY)));
: (StrUtil.equals(DelegationState.RESOLVED.toString(), delegationStateString)
? "RESOLVED" : StrUtil.EMPTY)));
processNodeRecordAssignUserParamDto.setApproveDesc(Convert.toStr(task.getVariableLocal("approveDesc")));
processNodeRecordAssignUserParamDto.setExecutionId(task.getExecutionId());
@@ -57,7 +57,8 @@ public class NodeTaskCompleteNotify {
// 获取节点配置并发送通知
RetOps
.of(processNodeDataService.getNodeData(taskCompleteParamDto.getFlowId(), taskCompleteParamDto.getNodeId()))
.of(processNodeDataService.getPublishedNode(taskCompleteParamDto.getProcessDefinitionId(),
taskCompleteParamDto.getNodeId()))
.getData()
.ifPresent(node -> {
// 验证事件配置是否为有效的URL
@@ -75,8 +76,7 @@ public class NodeTaskCompleteNotify {
/**
* 异步发送流程实例状态变更通知
* <p>
* 当流程实例状态发生变更时,查询最新的流程实例记录状态,
* 并通过Redis发布流程状态变更事件,供订阅方消费处理。
* 当流程实例状态发生变更时,查询最新的流程实例记录状态, 并通过Redis发布流程状态变更事件,供订阅方消费处理。
* </p>
* @param token 安全令牌,用于在异步线程中传递认证信息
* @param instanceRecordParamDto 流程实例参数,包含流程实例ID等信息
@@ -71,7 +71,7 @@ public class MultiInstanceHandler {
}).get(0);
// 节点数据
Node node = processNodeDataService.getNodeData(flowId, nodeId).getData();
Node node = processNodeDataService.getPublishedNode(execution.getProcessDefinitionId(), nodeId).getData();
if (node != null) {
Map<String, Object> variables = execution.getVariables();
variables.put("execution", execution);
@@ -116,7 +116,7 @@ public class MultiInstanceHandler {
String processDefinitionKey = entity.getProcessDefinitionKey();
String nodeId = execution.getCurrentActivityId();
Node node = processNodeDataService.getNodeData(processDefinitionKey, nodeId).getData();
Node node = processNodeDataService.getPublishedNode(execution.getProcessDefinitionId(), nodeId).getData();
Integer multipleMode = node.getMultipleMode();
BigDecimal modePercentage = BigDecimal.valueOf(100);
@@ -14,17 +14,11 @@ import org.flowable.engine.impl.persistence.entity.ExecutionEntityImpl;
/**
* 审批服务任务
* <p>
* Flowable服务任务(Service Task)实现,用于处理审批节点的自动化逻辑。
* 该任务在审批节点执行后自动触发,根据审批结果(通过/拒绝)决定流程的走向。
* Flowable服务任务(Service Task)实现,用于处理审批节点的自动化逻辑。 该任务在审批节点执行后自动触发,根据审批结果(通过/拒绝)决定流程的走向。
* <p>
* 主要功能:
* 1. 读取审批节点的审批结果
* 2. 如果审批通过,流程继续正常执行
* 3. 如果审批拒绝,根据配置执行拒绝策略(跳转到指定节点或结束流程)
* 主要功能: 1. 读取审批节点的审批结果 2. 如果审批通过,流程继续正常执行 3. 如果审批拒绝,根据配置执行拒绝策略(跳转到指定节点或结束流程)
* <p>
* 使用方式:
* 在流程定义中,将此类配置为审批节点后的服务任务,
* 任务ID需要遵循命名规则:approve_service_task_{nodeId}
* 使用方式: 在流程定义中,将此类配置为审批节点后的服务任务, 任务ID需要遵循命名规则:approve_service_task_{nodeId}
*
* @author pigx
*/
@@ -33,9 +27,7 @@ public class ApproveServiceTask implements JavaDelegate {
/**
* 执行审批服务任务
* <p>
* 根据审批节点的审批结果决定流程的下一步走向。
* 如果审批通过,流程正常继续;如果审批拒绝,执行拒绝策略。
*
* 根据审批节点的审批结果决定流程的下一步走向。 如果审批通过,流程正常继续;如果审批拒绝,执行拒绝策略。
* @param execution 流程执行上下文,包含流程实例信息和变量
*/
@Override
@@ -45,7 +37,7 @@ public class ApproveServiceTask implements JavaDelegate {
// 从执行上下文中提取关键标识符
String currentActivityId = entity.getActivityId(); // 当前活动ID
String flowDefinitionKey = entity.getProcessDefinitionKey(); // 流程定义键
String processDefinitionId = entity.getProcessDefinitionId(); // 实例绑定的发布版本
String processInstanceId = entity.getProcessInstanceId(); // 流程实例ID
// 从活动ID中提取节点标识符
@@ -64,7 +56,7 @@ public class ApproveServiceTask implements JavaDelegate {
// 如果未批准,确定工作流重定向的位置
if (!isApproved) {
handleRejection(flowDefinitionKey, nodeId, currentActivityId, processInstanceId, runtimeService);
handleRejection(processDefinitionId, nodeId, currentActivityId, processInstanceId, runtimeService);
}
// 如果已批准,工作流将沿其正常路径继续(无需操作)
}
@@ -73,24 +65,20 @@ public class ApproveServiceTask implements JavaDelegate {
/**
* 处理任务被拒绝时的流程重定向逻辑
* <p>
* 根据节点配置的拒绝策略,将流程重定向到指定节点或结束节点。
* 支持的拒绝策略:
* 1. TO_NODE:跳转到指定节点
* 2. 其他:直接结束流程
*
* @param flowDefinitionKey 流程定义键
* @param nodeId 当前节点标识符
* 根据节点配置的拒绝策略,将流程重定向到指定节点或结束节点。 支持的拒绝策略: 1. TO_NODE:跳转到指定节点 2. 其他:直接结束流程
* @param processDefinitionId 流程定义 ID
* @param nodeId 当前节点标识符
* @param currentActivityId 当前活动标识符
* @param processInstanceId 流程实例标识符
* @param runtimeService 用于流程状态操作的运行时服务
* @param runtimeService 用于流程状态操作的运行时服务
*/
private void handleRejection(String flowDefinitionKey, String nodeId, String currentActivityId,
private void handleRejection(String processDefinitionId, String nodeId, String currentActivityId,
String processInstanceId, RuntimeService runtimeService) {
// 获取流程节点数据服务以获取节点配置数据
IProcessNodeDataService processNodeDataService = SpringUtil.getBean(IProcessNodeDataService.class);
// 查询节点的原始配置数据
Node node = processNodeDataService.getNodeData(flowDefinitionKey, nodeId).getData();
Node node = processNodeDataService.getPublishedNode(processDefinitionId, nodeId).getData();
// 获取拒绝处理配置
Refuse refuseConfig = node.getRefuse();
@@ -26,19 +26,12 @@ import java.util.stream.Collectors;
/**
* 抄送服务任务
* <p>
* Flowable服务任务(Service Task)实现,用于处理流程中的抄送功能。
* 抄送任务会将流程信息发送给指定的用户或部门成员,供其查看但不需要审批。
* Flowable服务任务(Service Task)实现,用于处理流程中的抄送功能。 抄送任务会将流程信息发送给指定的用户或部门成员,供其查看但不需要审批。
* <p>
* 主要功能:
* 1. 获取节点配置的抄送人员(支持用户、部门、角色、岗位)
* 2. 将部门、角色、岗位展开为具体的用户列表
* 3. 为每个抄送人创建抄送记录
* 4. 记录流程快照信息供抄送人查看
* 主要功能: 1. 获取节点配置的抄送人员(支持用户、部门、角色、岗位) 2. 将部门、角色、岗位展开为具体的用户列表 3. 为每个抄送人创建抄送记录 4.
* 记录流程快照信息供抄送人查看
* <p>
* 使用场景:
* - 需要通知相关人员但不需要审批的节点
* - 流程信息备案和知会
* - 跨部门协作中的信息同步
* 使用场景: - 需要通知相关人员但不需要审批的节点 - 流程信息备案和知会 - 跨部门协作中的信息同步
*
* @author pigx
*/
@@ -47,9 +40,7 @@ public class CopyServiceTask implements JavaDelegate {
/**
* 执行抄送任务
* <p>
* 从节点配置中获取抄送人员列表,创建抄送记录并通过远程服务保存。
* 支持直接指定用户或指定部门(会展开为部门下的所有用户)。
*
* 从节点配置中获取抄送人员列表,创建抄送记录并通过远程服务保存。 支持直接指定用户或指定部门(会展开为部门下的所有用户)。
* @param execution 流程执行上下文,包含流程实例信息和变量
* @throws Exception 当JSON解析或远程服务调用失败时抛出异常
*/
@@ -63,7 +54,7 @@ public class CopyServiceTask implements JavaDelegate {
IRemoteService remoteService = SpringUtil.getBean(IRemoteService.class);
IProcessNodeDataService processNodeDataService = SpringUtil.getBean(IProcessNodeDataService.class);
Node node = processNodeDataService.getNodeData(flowId, nodeId).getData();
Node node = processNodeDataService.getPublishedNode(execution.getProcessDefinitionId(), nodeId).getData();
// 获取指定人员
List<NodeUser> userDtoList = node.getNodeUserList();
@@ -80,29 +71,29 @@ public class CopyServiceTask implements JavaDelegate {
if (CollUtil.isNotEmpty(deptIdList)) {
R<List<Long>> r = remoteService.queryUserIdListByDepIdList(deptIdList);
addUserIds(userIdList, r.getData());
}
addUserIds(userIdList, r.getData());
}
// 角色ID列表
List<Long> roleIdList = userDtoList.stream()
.filter(w -> StrUtil.equals(w.getType(), NodeUserTypeEnum.ROLE.getKey()))
.map(NodeUser::getId)
.toList();
// 角色ID列表
List<Long> roleIdList = userDtoList.stream()
.filter(w -> StrUtil.equals(w.getType(), NodeUserTypeEnum.ROLE.getKey()))
.map(NodeUser::getId)
.toList();
if (CollUtil.isNotEmpty(roleIdList)) {
R<List<Long>> r = remoteService.queryUserIdListByRoleIdList(roleIdList);
addUserIds(userIdList, r.getData());
}
if (CollUtil.isNotEmpty(roleIdList)) {
R<List<Long>> r = remoteService.queryUserIdListByRoleIdList(roleIdList);
addUserIds(userIdList, r.getData());
}
// 岗位ID列表
List<Long> postIdList = userDtoList.stream()
.filter(w -> StrUtil.equals(w.getType(), NodeUserTypeEnum.POST.getKey()))
.map(NodeUser::getId)
.toList();
// 岗位ID列表
List<Long> postIdList = userDtoList.stream()
.filter(w -> StrUtil.equals(w.getType(), NodeUserTypeEnum.POST.getKey()))
.map(NodeUser::getId)
.toList();
if (CollUtil.isNotEmpty(postIdList)) {
R<List<Long>> r = remoteService.queryUserIdListByPostIdList(postIdList);
addUserIds(userIdList, r.getData());
if (CollUtil.isNotEmpty(postIdList)) {
R<List<Long>> r = remoteService.queryUserIdListByPostIdList(postIdList);
addUserIds(userIdList, r.getData());
}
ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);
@@ -114,7 +105,7 @@ public class CopyServiceTask implements JavaDelegate {
.get(0);
Map<String, Object> variables = execution.getVariables();
String formData = objectMapper.writeValueAsString(variables);
String formData = objectMapper.writeValueAsString(variables);
for (String userIds : userIdList) {
// 发送抄送任务
@@ -125,28 +116,27 @@ public class CopyServiceTask implements JavaDelegate {
processCopyDto.setProcessInstanceId(execution.getProcessInstanceId());
processCopyDto.setNodeId(nodeId);
processCopyDto.setNodeName(node.getName());
processCopyDto.setFormData(formData);
processCopyDto.setFormData(formData);
processCopyDto.setUserId(Long.parseLong(userIds));
remoteService.saveCC(processCopyDto);
}
}
/**
* 将用户ID列表合并到目标列表中(去重)
*
* @param targetList 目标用户ID列表
* @param sourceList 待合并的用户ID列表
*/
private void addUserIds(List<String> targetList, List<Long> sourceList) {
if (CollUtil.isNotEmpty(sourceList)) {
for (Long userId : sourceList) {
String userIdStr = Convert.toStr(userId);
if (!targetList.contains(userIdStr)) {
targetList.add(userIdStr);
}
}
}
}
/**
* 将用户ID列表合并到目标列表中(去重)
* @param targetList 目标用户ID列表
* @param sourceList 待合并的用户ID列表
*/
private void addUserIds(List<String> targetList, List<Long> sourceList) {
if (CollUtil.isNotEmpty(sourceList)) {
for (Long userId : sourceList) {
String userIdStr = Convert.toStr(userId);
if (!targetList.contains(userIdStr)) {
targetList.add(userIdStr);
}
}
}
}
}
@@ -29,11 +29,9 @@ import java.util.Map;
/**
* 触发器服务任务
* <p>
* 在流程执行过程中调用外部 HTTP POST 接口,
* 支持传递表单数据作为请求参数,并将返回值映射回流程变量。
* 在流程执行过程中调用外部 HTTP POST 接口, 支持传递表单数据作为请求参数,并将返回值映射回流程变量。
* <p>
* HTTP 状态码 200 即视为调用成功,响应体按 JSON 对象解析后进行返回值映射。
* 异常处理支持两种策略:THROW(抛出异常)和 TERMINATE(终止流程)。
* HTTP 状态码 200 即视为调用成功,响应体按 JSON 对象解析后进行返回值映射。 异常处理支持两种策略:THROW(抛出异常)和 TERMINATE(终止流程)。
*
* @author pigx
* @date 2026-03-03
@@ -41,162 +39,173 @@ import java.util.Map;
@Slf4j
public class TriggerServiceTask implements JavaDelegate {
private static final int READ_TIMEOUT = 30000;
private static final int READ_TIMEOUT = 30000;
@SneakyThrows
@Override
public void execute(DelegateExecution execution) {
ExecutionEntityImpl entity = (ExecutionEntityImpl) execution;
String nodeId = entity.getActivityId();
String flowId = entity.getProcessDefinitionKey();
/**
* 按当前实例的发布定义读取触发器配置,调用外部 POST 接口并将响应映射回流程变量。 请求或响应处理失败时遵循配置的 THROW / TERMINATE
* 策略;节点快照缺失时直接失败。
* @param execution 当前触发器节点的 Flowable 执行上下文,不允许为 null,须为 ExecutionEntityImpl
* @throws IllegalStateException 当前发布定义缺少业务节点快照时抛出
* @throws FlowableException 配置无效或调用失败时抛出;TERMINATE 策略使用特定异常前缀交由上层终止实例
*/
@SneakyThrows
@Override
public void execute(DelegateExecution execution) {
ExecutionEntityImpl entity = (ExecutionEntityImpl) execution;
String nodeId = entity.getActivityId();
String flowId = entity.getProcessDefinitionKey();
IProcessNodeDataService processNodeDataService = SpringUtil.getBean(IProcessNodeDataService.class);
Node node = processNodeDataService.getNodeData(flowId, nodeId).getData();
IProcessNodeDataService processNodeDataService = SpringUtil.getBean(IProcessNodeDataService.class);
Node node = processNodeDataService.getPublishedNode(execution.getProcessDefinitionId(), nodeId).getData();
TriggerConfig config = node.getTriggerConfig();
if (config == null || StrUtil.isBlank(config.getUrl())) {
throw new FlowableException("触发器配置无效: URL为空");
}
TriggerConfig config = node.getTriggerConfig();
if (config == null || StrUtil.isBlank(config.getUrl())) {
throw new FlowableException("触发器配置无效: URL为空");
}
// 构建请求头
Map<String, String> headerMap = new HashMap<>();
headerMap.put("Content-Type", "application/json");
if (CollUtil.isNotEmpty(config.getHeaders())) {
for (TriggerParam param : config.getHeaders()) {
String value = resolveParamValue(param, execution);
if (StrUtil.isNotBlank(param.getKey()) && value != null) {
headerMap.put(param.getKey(), value);
}
}
}
// 构建请求头
Map<String, String> headerMap = new HashMap<>();
headerMap.put("Content-Type", "application/json");
if (CollUtil.isNotEmpty(config.getHeaders())) {
for (TriggerParam param : config.getHeaders()) {
String value = resolveParamValue(param, execution);
if (StrUtil.isNotBlank(param.getKey()) && value != null) {
headerMap.put(param.getKey(), value);
}
}
}
// 构建请求体
Map<String, Object> body = new HashMap<>();
if (CollUtil.isNotEmpty(config.getBody())) {
for (TriggerParam param : config.getBody()) {
Object value = resolveParamObjectValue(param, execution);
if (StrUtil.isNotBlank(param.getKey())) {
body.put(param.getKey(), value);
}
}
}
// 构建请求体
Map<String, Object> body = new HashMap<>();
if (CollUtil.isNotEmpty(config.getBody())) {
for (TriggerParam param : config.getBody()) {
Object value = resolveParamObjectValue(param, execution);
if (StrUtil.isNotBlank(param.getKey())) {
body.put(param.getKey(), value);
}
}
}
// 发起 POST 请求
try {
HttpResponse response = HttpRequest.post(config.getUrl())
.headerMap(headerMap, true)
.body(JSONUtil.toJsonStr(body))
.timeout(READ_TIMEOUT)
.execute();
// 发起 POST 请求
try {
HttpResponse response = HttpRequest.post(config.getUrl())
.headerMap(headerMap, true)
.body(JSONUtil.toJsonStr(body))
.timeout(READ_TIMEOUT)
.execute();
// HTTP 状态码非 2xx 视为调用失败
if (!response.isOk()) {
String msg = StrUtil.format("HTTP请求失败: status={}, body={}", response.getStatus(), response.body());
log.warn("触发器返回异常: url={}, {}", config.getUrl(), msg);
handleError(config.getErrorHandler(), execution, msg);
return;
}
// HTTP 状态码非 2xx 视为调用失败
if (!response.isOk()) {
String msg = StrUtil.format("HTTP请求失败: status={}, body={}", response.getStatus(), response.body());
log.warn("触发器返回异常: url={}, {}", config.getUrl(), msg);
handleError(config.getErrorHandler(), execution, msg);
return;
}
// 响应体按 JSON 对象解析,进行返回值映射
String responseBody = response.body();
if (StrUtil.isNotBlank(responseBody)) {
ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);
Object data = objectMapper.readValue(responseBody, Object.class);
mapReturnValues(config.getReturnMapping(), data, execution);
}
} catch (Exception ex) {
log.error("触发器调用异常: url={}, error={}", config.getUrl(), ex.getMessage());
handleError(config.getErrorHandler(), execution, ex.getMessage());
}
}
// 响应体按 JSON 对象解析,进行返回值映射
String responseBody = response.body();
if (StrUtil.isNotBlank(responseBody)) {
ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);
Object data = objectMapper.readValue(responseBody, Object.class);
mapReturnValues(config.getReturnMapping(), data, execution);
}
}
catch (Exception ex) {
log.error("触发器调用异常: url={}, error={}", config.getUrl(), ex.getMessage());
handleError(config.getErrorHandler(), execution, ex.getMessage());
}
}
private String resolveParamValue(TriggerParam param, DelegateExecution execution) {
Object value = resolveParamObjectValue(param, execution);
return value != null ? value.toString() : null;
}
private String resolveParamValue(TriggerParam param, DelegateExecution execution) {
Object value = resolveParamObjectValue(param, execution);
return value != null ? value.toString() : null;
}
private Object resolveParamObjectValue(TriggerParam param, DelegateExecution execution) {
if (Boolean.TRUE.equals(param.getUseForm()) && StrUtil.isNotBlank(param.getFormFieldId())) {
return execution.getVariable(param.getFormFieldId());
}
return param.getValue();
}
private Object resolveParamObjectValue(TriggerParam param, DelegateExecution execution) {
if (Boolean.TRUE.equals(param.getUseForm()) && StrUtil.isNotBlank(param.getFormFieldId())) {
return execution.getVariable(param.getFormFieldId());
}
return param.getValue();
}
@SneakyThrows
@SuppressWarnings("unchecked")
private void mapReturnValues(List<TriggerReturnMapping> mappings, Object data, DelegateExecution execution) {
if (CollUtil.isEmpty(mappings) || data == null) {
return;
}
@SneakyThrows
@SuppressWarnings("unchecked")
private void mapReturnValues(List<TriggerReturnMapping> mappings, Object data, DelegateExecution execution) {
if (CollUtil.isEmpty(mappings) || data == null) {
return;
}
Map<String, Object> dataMap;
if (data instanceof Map) {
dataMap = (Map<String, Object>) data;
} else {
ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);
dataMap = objectMapper.convertValue(data, Map.class);
}
Map<String, Object> dataMap;
if (data instanceof Map) {
dataMap = (Map<String, Object>) data;
}
else {
ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);
dataMap = objectMapper.convertValue(data, Map.class);
}
// 收集需要更新的表单字段
Map<String, Object> formUpdates = new HashMap<>();
for (TriggerReturnMapping mapping : mappings) {
Object value = dataMap.get(mapping.getResponseKey());
if (value != null && StrUtil.isNotBlank(mapping.getFormFieldId())) {
execution.setVariable(mapping.getFormFieldId(), value);
formUpdates.put(mapping.getFormFieldId(), value);
}
}
// 收集需要更新的表单字段
Map<String, Object> formUpdates = new HashMap<>();
for (TriggerReturnMapping mapping : mappings) {
Object value = dataMap.get(mapping.getResponseKey());
if (value != null && StrUtil.isNotBlank(mapping.getFormFieldId())) {
execution.setVariable(mapping.getFormFieldId(), value);
formUpdates.put(mapping.getFormFieldId(), value);
}
}
// 同步更新 ProcessInstanceRecord.formData
if (CollUtil.isNotEmpty(formUpdates)) {
syncFormData(execution.getProcessInstanceId(), formUpdates);
}
}
// 同步更新 ProcessInstanceRecord.formData
if (CollUtil.isNotEmpty(formUpdates)) {
syncFormData(execution.getProcessInstanceId(), formUpdates);
}
}
/**
* 同步更新流程实例记录的表单数据
*/
@SneakyThrows
private void syncFormData(String processInstanceId, Map<String, Object> formUpdates) {
try {
ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);
IProcessInstanceRecordService recordService = SpringUtil.getBean(IProcessInstanceRecordService.class);
/**
* 同步更新流程实例记录的表单数据
*/
@SneakyThrows
private void syncFormData(String processInstanceId, Map<String, Object> formUpdates) {
try {
ObjectMapper objectMapper = SpringUtil.getBean(ObjectMapper.class);
IProcessInstanceRecordService recordService = SpringUtil.getBean(IProcessInstanceRecordService.class);
ProcessInstanceRecord record = recordService.lambdaQuery()
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
.one();
ProcessInstanceRecord record = recordService.lambdaQuery()
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
.one();
if (record == null) {
return;
}
if (record == null) {
return;
}
Map<String, Object> existingFormData = new HashMap<>();
if (StrUtil.isNotBlank(record.getFormData())) {
existingFormData = objectMapper.readValue(record.getFormData(), new TypeReference<Map<String, Object>>() {
});
}
Map<String, Object> existingFormData = new HashMap<>();
if (StrUtil.isNotBlank(record.getFormData())) {
existingFormData = objectMapper.readValue(record.getFormData(),
new TypeReference<Map<String, Object>>() {
});
}
existingFormData.putAll(formUpdates);
existingFormData.putAll(formUpdates);
String updatedFormData = objectMapper.writeValueAsString(existingFormData);
recordService.lambdaUpdate()
.set(ProcessInstanceRecord::getFormData, updatedFormData)
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
.update();
String updatedFormData = objectMapper.writeValueAsString(existingFormData);
recordService.lambdaUpdate()
.set(ProcessInstanceRecord::getFormData, updatedFormData)
.eq(ProcessInstanceRecord::getProcessInstanceId, processInstanceId)
.update();
log.debug("触发器同步更新流程实例 {} 的表单数据", processInstanceId);
} catch (Exception e) {
log.error("触发器同步更新表单数据失败: {}", e.getMessage(), e);
}
}
log.debug("触发器同步更新流程实例 {} 的表单数据", processInstanceId);
}
catch (Exception e) {
log.error("触发器同步更新表单数据失败: {}", e.getMessage(), e);
}
}
private void handleError(String handler, DelegateExecution execution, String msg) {
if ("TERMINATE".equals(handler)) {
// 不在 JavaDelegate 内调用 deleteProcessInstance,因为后续异常会导致 Flowable 事务回滚
// 抛出带特定前缀的异常,由上层 completeTask 捕获后在事务外执行终止操作
throw new FlowableException("TRIGGER_TERMINATE:" + msg);
}
throw new FlowableException("触发器调用失败: " + msg);
}
private void handleError(String handler, DelegateExecution execution, String msg) {
if ("TERMINATE".equals(handler)) {
// 不在 JavaDelegate 内调用 deleteProcessInstance,因为后续异常会导致 Flowable 事务回滚
// 抛出带特定前缀的异常,由上层 completeTask 捕获后在事务外执行终止操作
throw new FlowableException("TRIGGER_TERMINATE:" + msg);
}
throw new FlowableException("触发器调用失败: " + msg);
}
}
@@ -1,23 +1,86 @@
package com.pig4cloud.pigx.flow.support.utils;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.pig4cloud.pigx.flow.constant.NodeTypeEnum;
import com.pig4cloud.pigx.flow.constant.ProcessInstanceConstant;
import com.pig4cloud.pigx.flow.dto.Node;
import java.util.ArrayList;
import java.util.List;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
/**
* 流程节点工具类
* <p>
* 提供流程节点的常用操作方法,包括节点遍历、节点查找、节点构建等功能。 主要用于处理流程定义的树形结构,支持条件分支和并行分支的递归处理。
* </p>
*
*
* @author pigx
*/
public class NodeUtil {
/**
* 发布时递归修改节点及条件分支的权限映射,为当前表单中未配置权限的字段补齐 EDIT。 已配置的非空权限保持不变;支持表单 children 中的嵌套字段。
* @param node 待固化的节点树;为 null 时不执行任何操作
* @param formItems 表单字段 JSON 数组;为 null、空串或空白时只复制已有权限
* @throws cn.hutool.json.JSONException 表单内容不能解析为 JSON 数组时抛出;此前已处理的节点不会恢复
*/
public static void snapshotFormPermissions(Node node, String formItems) {
if (Objects.isNull(node)) {
return;
}
node.setFormPerms(formPermissions(node, formItems, ProcessInstanceConstant.FormPermClass.EDIT));
if (Objects.nonNull(node.getConditionNodes())) {
for (Node branch : node.getConditionNodes()) {
snapshotFormPermissions(branch, formItems);
}
}
snapshotFormPermissions(node.getChildren(), formItems);
}
/**
* 复制发布时的字段权限,为当前表单中快照未声明的字段补齐 HIDE,不修改节点原有映射。 同时处理表单 children 中的嵌套字段,并保留已有非空权限。
* @param node 已发布节点,不允许为 null;其权限映射允许为 null
* @param formItems 当前表单字段 JSON 数组;为 null、空串或空白时仅返回已有权限的副本
* @return 独立、可修改的字段权限映射,不为 null
* @throws NullPointerException node 为 null 时抛出
* @throws cn.hutool.json.JSONException 表单内容不能解析为 JSON 数组时抛出
*/
public static Map<String, String> publishedFormPermissions(Node node, String formItems) {
return formPermissions(node, formItems, ProcessInstanceConstant.FormPermClass.HIDE);
}
private static Map<String, String> formPermissions(Node node, String formItems, String defaultPermission) {
Map<String, String> permissions = new HashMap<>();
if (Objects.nonNull(node.getFormPerms())) {
permissions.putAll(node.getFormPerms());
}
if (StrUtil.isNotBlank(formItems)) {
fillMissingPermissions(JSONUtil.parseArray(formItems), permissions, defaultPermission);
}
return permissions;
}
private static void fillMissingPermissions(JSONArray items, Map<String, String> permissions,
String defaultPermission) {
for (int i = 0; i < items.size(); i++) {
JSONObject item = items.getJSONObject(i);
String field = item.getStr("field");
if (StrUtil.isNotBlank(field) && StrUtil.isBlank(permissions.get(field))) {
permissions.put(field, defaultPermission);
}
JSONArray children = item.getJSONArray("children");
if (Objects.nonNull(children)) {
fillMissingPermissions(children, permissions, defaultPermission);
}
}
}
/**
* 从流程定义ID中提取流程ID
* <p>
@@ -64,7 +64,8 @@ class ProcessInstanceServiceImplTests {
ProcessInstanceServiceImpl service = new ProcessInstanceServiceImpl(mock(RemoteUserService.class),
mock(TaskService.class), mock(RuntimeService.class), mock(HistoryService.class), recordService,
mock(IProcessCopyService.class), mock(IProcessService.class), mock(IProcessNodeRecordService.class),
mock(IProcessNodeRecordAssignUserService.class), new ObjectMapper());
mock(IProcessNodeRecordAssignUserService.class), new ObjectMapper(),
mock(com.pig4cloud.pigx.flow.service.IProcessNodeDataService.class));
ProcessInstanceParamDto request = new ProcessInstanceParamDto();
request.setProcessInstanceId("process-7598");
request.setParamMap(variables);
@@ -0,0 +1,290 @@
package com.pig4cloud.pigx.flow.service.impl;
import cn.hutool.extra.spring.SpringUtil;
import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder;
import com.baomidou.mybatisplus.core.config.GlobalConfig;
import com.baomidou.mybatisplus.core.handlers.MetaObjectHandler;
import com.baomidou.mybatisplus.core.toolkit.GlobalConfigUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pig4cloud.pigx.flow.dto.Node;
import com.pig4cloud.pigx.flow.dto.NodeUser;
import com.pig4cloud.pigx.flow.dto.ProcessNodeDataDto;
import com.pig4cloud.pigx.flow.entity.Process;
import com.pig4cloud.pigx.flow.mapper.ProcessNodeDataMapper;
import com.pig4cloud.pigx.flow.service.IProcessService;
import com.pig4cloud.pigx.flow.service.IProcessStarterService;
import com.pig4cloud.pigx.flow.support.node.MultiInstanceHandler;
import com.pig4cloud.pigx.flow.support.node.impl.AssignUserFixedStrategyImpl;
import com.pig4cloud.pigx.flow.support.utils.ModelUtil;
import com.pig4cloud.pigx.flow.support.utils.NodeUtil;
import org.apache.ibatis.mapping.Environment;
import org.apache.ibatis.reflection.MetaObject;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
import org.flowable.bpmn.model.BpmnModel;
import org.flowable.bpmn.model.UserTask;
import org.flowable.engine.ProcessEngine;
import org.flowable.engine.ProcessEngineConfiguration;
import org.flowable.engine.repository.ProcessDefinition;
import org.flowable.engine.runtime.ProcessInstance;
import org.flowable.spring.SpringProcessEngineConfiguration;
import org.flowable.task.api.Task;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.transaction.support.TransactionTemplate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.List;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
class ProcessNodeVersionTests {
private ProcessEngine engine;
private ProcessNodeDataServiceImpl nodes;
private GenericApplicationContext context;
private DataSourceTransactionManager transactions;
private JdbcTemplate jdbc;
@BeforeEach
void setUp() throws Exception {
TenantContextHolder.setTenantId(1L);
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
dataSource.setUser("sa");
jdbc = new JdbcTemplate(dataSource);
jdbc.execute("create table process_node_data(id bigint primary key, flow_id varchar(255), "
+ "process_definition_id varchar(64), tenant_id bigint, node_id varchar(50), data clob, "
+ "del_flag varchar(1) default '0', create_time timestamp, update_time timestamp, "
+ "unique(tenant_id, process_definition_id, node_id))");
transactions = new DataSourceTransactionManager(dataSource);
SpringProcessEngineConfiguration configuration = new SpringProcessEngineConfiguration();
configuration.setDataSource(dataSource);
configuration.setTransactionManager(transactions);
configuration.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE);
configuration.setDisableIdmEngine(true);
configuration.setDisableEventRegistry(true);
configuration.setAsyncExecutorActivate(false);
Map<Object, Object> beans = new HashMap<>();
configuration.setBeans(beans);
engine = configuration.buildProcessEngine();
MybatisConfiguration mybatis = new MybatisConfiguration();
mybatis.setMapUnderscoreToCamelCase(true);
mybatis.setEnvironment(new Environment("test", new SpringManagedTransactionFactory(), dataSource));
GlobalConfig global = new GlobalConfig();
global.setDbConfig(new GlobalConfig.DbConfig());
global.setMetaObjectHandler(new MetaObjectHandler() {
@Override
public void insertFill(MetaObject metaObject) {
setFieldValByName("delFlag", "0", metaObject);
}
@Override
public void updateFill(MetaObject metaObject) {
}
});
GlobalConfigUtils.setGlobalConfig(mybatis, global);
mybatis.addMapper(ProcessNodeDataMapper.class);
var factory = new MybatisSqlSessionFactoryBuilder().build(mybatis);
nodes = new ProcessNodeDataServiceImpl(engine.getRepositoryService());
ReflectionTestUtils.setField(nodes, "baseMapper",
new SqlSessionTemplate(factory).getMapper(ProcessNodeDataMapper.class));
beans.put("multiInstanceHandler", new MultiInstanceHandler(nodes,
Map.of("1AssignUserStrategy", new AssignUserFixedStrategyImpl(null)), new ObjectMapper()));
context = new GenericApplicationContext();
context.getBeanFactory().registerSingleton("processNodeDataService", nodes);
context.refresh();
SpringUtil spring = new SpringUtil();
spring.setApplicationContext(context);
spring.postProcessBeanFactory(context.getBeanFactory());
}
@AfterEach
void tearDown() {
TenantContextHolder.clear();
if (engine != null) {
engine.close();
}
if (context != null) {
context.close();
}
if (jdbc != null) {
jdbc.execute("shutdown");
}
}
@Test
void oldInstanceKeepsPublishedApproverAndPermissionsAfterRepublish() {
ProcessDefinition a = publish("leave", 1001L, "R");
ProcessInstance old = start("leave");
ProcessDefinition b = publish("leave", 2002L, "E");
assertThat(engine.getRuntimeService()
.createProcessInstanceQuery()
.processInstanceId(old.getId())
.singleResult()
.getProcessDefinitionId()).isEqualTo(a.getId());
assertThat(advance(old).getAssignee()).isEqualTo("1001");
assertThat(advance(start("leave")).getAssignee()).isEqualTo("2002");
assertThat(nodes.getPublishedNode(a.getId(), "approve").getData().getFormPerms()).containsEntry("amount", "R");
assertThat(nodes.getPublishedNode(b.getId(), "approve").getData().getFormPerms()).containsEntry("amount", "E");
assertThat(NodeUtil.publishedFormPermissions(nodes.getPublishedNode(a.getId(), "approve").getData(),
"[{\"field\":\"amount\"},{\"children\":[{\"field\":\"extra\"}]}]"))
.containsEntry("amount", "R")
.containsEntry("extra", "H");
assertThat(NodeUtil.publishedFormPermissions(nodes.getPublishedNode(a.getId(), "root").getData(),
"[{\"field\":\"amount\"},{\"field\":\"layout\",\"children\":[{\"field\":\"nested\"},{\"field\":\"extra\"}]}]"))
.containsEntry("amount", "E")
.containsEntry("layout", "E")
.containsEntry("nested", "E")
.containsEntry("extra", "H");
assertThat(nodes.getPublishedNode(a.getId(), "start").getData()).isNull();
jdbc.update("delete from process_node_data where process_definition_id = ? and node_id = ?", a.getId(),
"approve");
assertThatThrownBy(() -> nodes.getPublishedNode(a.getId(), "approve")).isInstanceOf(IllegalStateException.class)
.hasMessageContaining("缺少节点版本快照:approve");
assertThat(nodes.saveNodeData(new ProcessNodeDataDto()).getCode()).isEqualTo(1);
assertThatThrownBy(() -> nodes.getPublishedNode("leave:0:missing", "approve"))
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining("缺少节点版本快照");
}
@Test
void failureAfterSnapshotRollsBackDeploymentAndNodesWithoutChangingOldVersion() throws Exception {
ProcessDefinition a = publish("leave", 1001L, "R");
long before = jdbc.queryForObject("select count(*) from process_node_data", Long.class);
ProcessServiceImpl target = new ProcessServiceImpl(mock(IProcessStarterService.class),
engine.getRepositoryService(), nodes, new ObjectMapper(), null, null) {
@Override
public Process getByFlowId(String flowId) {
return null;
}
@Override
public boolean saveOrUpdate(Process process) {
assertThat(jdbc.queryForObject("select count(*) from process_node_data", Long.class))
.isGreaterThan(before);
ProcessDefinition published = engine.getRepositoryService()
.createProcessDefinitionQuery()
.processDefinitionKey("leave")
.latestVersion()
.singleResult();
assertThat(published.getId()).isNotEqualTo(a.getId());
assertThat(nodes.getPublishedNode(published.getId(), "root").getData().getFormPerms())
.containsEntry("amount", "E")
.containsEntry("layout", "E")
.containsEntry("nested", "E");
assertThat(nodes.getPublishedNode(published.getId(), "approve").getData().getFormPerms())
.containsEntry("amount", "R")
.containsEntry("secret", "H")
.containsEntry("layout", "E")
.containsEntry("nested", "E");
throw new IllegalStateException("simulated metadata write failure");
}
};
TransactionInterceptor interceptor = new TransactionInterceptor();
interceptor.setTransactionManager(transactions);
interceptor.setTransactionAttributeSource(new AnnotationTransactionAttributeSource());
ProxyFactory proxy = new ProxyFactory(target);
proxy.addAdvice(interceptor);
IProcessService service = (IProcessService) proxy.getProxy();
Process request = new Process();
request.setFlowId("leave");
request.setName("Leave B");
Node requestedRoot = tree(2002L, "R");
requestedRoot.getChildren().setFormPerms(Map.of("amount", "R", "secret", "H"));
request.setProcess(new ObjectMapper().writeValueAsString(requestedRoot));
request.setFormItems(
"[{\"field\":\"amount\"},{\"field\":\"secret\"},{\"field\":\"layout\",\"children\":[{\"field\":\"nested\"}]}]");
request.setAdminList("[{\"id\":9000,\"type\":\"user\"}]");
assertThatThrownBy(() -> service.create(request)).isInstanceOf(IllegalStateException.class)
.hasMessage("simulated metadata write failure");
assertThat(engine.getRepositoryService().createProcessDefinitionQuery().processDefinitionKey("leave").count())
.isEqualTo(1);
assertThat(jdbc.queryForObject("select count(*) from process_node_data", Long.class)).isEqualTo(before);
assertThat(advance(start("leave")).getAssignee()).isEqualTo("1001");
assertThat(nodes.getPublishedNode(a.getId(), "approve").getData().getFormPerms()).containsEntry("amount", "R");
}
private ProcessDefinition publish(String key, long approver, String permission) {
return new TransactionTemplate(transactions).execute(status -> {
Node root = tree(approver, permission);
NodeUtil.snapshotFormPermissions(root,
"[{\"field\":\"amount\"},{\"field\":\"layout\",\"children\":[{\"field\":\"nested\"}]}]");
BpmnModel model = ModelUtil.buildBpmnModel(root, key, key);
// 隔离通知/业务记录依赖;保留真实多实例表达式和节点配置读写。
model.getMainProcess().setEventListeners(new ArrayList<>());
for (var element : model.getMainProcess().getFlowElements()) {
if (element instanceof UserTask task) {
task.setTaskListeners(new ArrayList<>());
if ("root".equals(task.getId())) {
task.setAssignee("9000");
}
}
}
var deployment = engine.getRepositoryService()
.createDeployment()
.addBpmnModel(key + ".bpmn20.xml", model)
.tenantId("1")
.deploy();
ProcessDefinition definition = engine.getRepositoryService()
.createProcessDefinitionQuery()
.deploymentId(deployment.getId())
.singleResult();
nodes.savePublishedNodes(definition, root);
return definition;
});
}
private Node tree(long approver, String permission) {
Node root = new Node();
root.setId("root");
root.setType(0);
root.setName("Start");
Node approval = new Node();
approval.setId("approve");
approval.setName("Approve");
approval.setType(1);
approval.setAssignedType(1);
approval.setMultipleMode(1);
approval.setNodeUserList(List.of(NodeUser.builder().id(approver).type("user").build()));
approval.setFormPerms(Map.of("amount", permission));
root.setChildren(approval);
return root;
}
private ProcessInstance start(String key) {
return engine.getRuntimeService()
.startProcessInstanceByKeyAndTenantId(key,
Map.of("root", List.of(Map.of("id", 9000L, "name", "Starter", "type", "user"))), "1");
}
private Task advance(ProcessInstance instance) {
Task waiting = engine.getTaskService().createTaskQuery().processInstanceId(instance.getId()).singleResult();
engine.getTaskService().complete(waiting.getId());
return engine.getTaskService().createTaskQuery().processInstanceId(instance.getId()).singleResult();
}
}
@@ -105,42 +105,61 @@ class TaskServiceImplTests {
@Mock
private IProcessService processService;
@Mock
private IProcessCopyService processCopyService;
@Mock
private IProcessNodeDataService nodeDataService;
@Mock
private IProcessNodeRecordAssignUserService processNodeRecordAssignUserService;
@Mock
private IProcessInstanceRecordService processInstanceRecordService;
@Mock
private IProcessNodeRecordService processNodeRecordService;
@Mock
private IFlowableEngineService flowableEngineService;
@Mock
private TaskService taskService;
@Mock
private HistoryService historyService;
@Mock
private RuntimeService runtimeService;
@Spy
private ObjectMapper objectMapper = new ObjectMapper();
@Mock
private RemoteUserService remoteUserService;
@Mock
private ProcessNodeRecordMapper processNodeRecordMapper;
@Mock
private ProcessTaskAddSignRecordMapper processTaskAddSignRecordMapper;
@Mock(answer = Answers.RETURNS_SELF)
private TaskQuery taskQuery;
@Mock
private BaseMapper<ProcessInstanceRecord> processInstanceBaseMapper;
@Mock
private BaseMapper<ProcessNodeRecordAssignUser> assignmentBaseMapper;
@Mock
private Task sourceTask;
@Mock
private Task addedTask;
@Mock
private Execution addedExecution;
@@ -157,11 +176,10 @@ class TaskServiceImplTests {
@BeforeEach
void setCurrentUserAndTenant() {
PigxUser user = new PigxUser(USER_ID, "approver", List.of(), null, List.of(), List.of(), null, null,
"审批人", "审批人", null, TENANT_ID, "password", true, true, "sys", true,
LocalDateTime.now(), true, List.of());
PigxUser user = new PigxUser(USER_ID, "approver", List.of(), null, List.of(), List.of(), null, null, "审批人",
"审批人", null, TENANT_ID, "password", true, true, "sys", true, LocalDateTime.now(), true, List.of());
SecurityContextHolder.getContext()
.setAuthentication(UsernamePasswordAuthenticationToken.authenticated(user, "password", List.of()));
.setAuthentication(UsernamePasswordAuthenticationToken.authenticated(user, "password", List.of()));
TenantContextHolder.setTenantId(TENANT_ID);
}
@@ -185,10 +203,9 @@ class TaskServiceImplTests {
@Test
void withdrawProcessInstancePropagatesWithdrawReasonBeforeMovingToEnd() {
ProcessInstanceRecord record = new ProcessInstanceRecord()
.setProcessInstanceId(PROCESS_INSTANCE_ID)
.setUserId(USER_ID)
.setStatus(NodeStatusEnum.JXZ.getCode());
ProcessInstanceRecord record = new ProcessInstanceRecord().setProcessInstanceId(PROCESS_INSTANCE_ID)
.setUserId(USER_ID)
.setStatus(NodeStatusEnum.JXZ.getCode());
ChangeActivityStateBuilder stateChangeBuilder = prepareTerminationFlow(record);
TaskParamDto request = new TaskParamDto();
request.setProcessInstanceId(PROCESS_INSTANCE_ID);
@@ -236,7 +253,7 @@ class TaskServiceImplTests {
prepareTargetValidation(List.of(originalAssignment()), validUser(TARGET_USER_ID, TENANT_ID));
prepareAuditInsert();
when(runtimeService.addMultiInstanceExecution(eq(NODE_ID), eq(PROCESS_INSTANCE_ID), anyMap()))
.thenReturn(addedExecution);
.thenReturn(addedExecution);
when(assignmentBaseMapper.selectCount(any())).thenReturn(1L);
R result = taskServiceImpl.addSign(request(List.of(TARGET_USER_ID), "需要共同确认"));
@@ -249,8 +266,7 @@ class TaskServiceImplTests {
assertThat(addSignResult.getTasks().get(0).getTaskId()).isEqualTo(ADDED_TASK_ID);
assertThat(addSignResult.getTasks().get(0).getUserId()).isEqualTo(TARGET_USER_ID);
ArgumentCaptor<ProcessTaskAddSignRecord> recordCaptor = ArgumentCaptor
.forClass(ProcessTaskAddSignRecord.class);
ArgumentCaptor<ProcessTaskAddSignRecord> recordCaptor = ArgumentCaptor.forClass(ProcessTaskAddSignRecord.class);
verify(processTaskAddSignRecordMapper).insert(recordCaptor.capture());
ProcessTaskAddSignRecord record = recordCaptor.getValue();
assertThat(record.getSourceTaskId()).isEqualTo(TASK_ID);
@@ -264,7 +280,7 @@ class TaskServiceImplTests {
verify(runtimeService).addMultiInstanceExecution(eq(NODE_ID), eq(PROCESS_INSTANCE_ID),
argThat(variables -> Long.valueOf(TARGET_USER_ID).equals(variables.get(NODE_ID + "_assignee_temp"))
&& Long.valueOf(ADD_SIGN_RECORD_ID)
.equals(variables.get(ProcessInstanceConstant.VAR_ADD_SIGN_RECORD_ID))));
.equals(variables.get(ProcessInstanceConstant.VAR_ADD_SIGN_RECORD_ID))));
}
@Test
@@ -272,7 +288,7 @@ class TaskServiceImplTests {
prepareTaskAndNode(null);
prepareActiveAssignment();
when(processInstanceRecordService.lambdaQuery())
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(processInstanceBaseMapper));
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(processInstanceBaseMapper));
when(processInstanceBaseMapper.selectCount(any())).thenReturn(1L);
R result = taskServiceImpl.addSign(request(List.of(TARGET_USER_ID), "需要共同确认"));
@@ -300,8 +316,8 @@ class TaskServiceImplTests {
prepareTaskAndNode(null);
prepareOriginalApprover();
ProcessNodeRecordAssignUser existingParticipant = new ProcessNodeRecordAssignUser()
.setNodeRecordId(NODE_RECORD_ID)
.setUserId(TARGET_USER_ID);
.setNodeRecordId(NODE_RECORD_ID)
.setUserId(TARGET_USER_ID);
prepareTargetValidation(List.of(originalAssignment(), existingParticipant),
validUser(TARGET_USER_ID, TENANT_ID));
@@ -319,11 +335,10 @@ class TaskServiceImplTests {
prepareTargetValidation(List.of(originalAssignment()), validUser(TARGET_USER_ID, TENANT_ID));
prepareAuditInsert();
when(runtimeService.addMultiInstanceExecution(anyString(), anyString(), anyMap()))
.thenThrow(new FlowableOptimisticLockingException("conflict"));
.thenThrow(new FlowableOptimisticLockingException("conflict"));
RecordingTransactionManager transactionManager = new RecordingTransactionManager();
R result = transactionalService(transactionManager)
.addSign(request(List.of(TARGET_USER_ID), "需要共同确认"));
R result = transactionalService(transactionManager).addSign(request(List.of(TARGET_USER_ID), "需要共同确认"));
assertThat(result.getCode()).isEqualTo(1);
assertThat(result.getMsg()).isEqualTo("节点状态已变化,请刷新后重试");
@@ -349,27 +364,26 @@ class TaskServiceImplTests {
}
ProcessNodeRecord nodeRecord = new ProcessNodeRecord().setId(NODE_RECORD_ID);
when(processNodeRecordMapper.selectActiveByIdForUpdate(NODE_RECORD_ID, PROCESS_INSTANCE_ID, NODE_ID,
TENANT_ID))
.thenReturn(nodeRecord);
when(processNodeRecordMapper.selectActiveByIdForUpdate(NODE_RECORD_ID, PROCESS_INSTANCE_ID, NODE_ID, TENANT_ID))
.thenReturn(nodeRecord);
Node node = new Node();
node.setAssignedType(ProcessInstanceConstant.AssignedTypeClass.ROLE);
node.setMultipleMode(ProcessInstanceConstant.MULTIPLE_MODE_AL_SAME);
node.setAllowAddSign(true);
when(nodeDataService.getNodeData(FLOW_ID, NODE_ID)).thenReturn(R.ok(node));
when(nodeDataService.getPublishedNode(PROCESS_DEFINITION_ID, NODE_ID)).thenReturn(R.ok(node));
}
private void prepareOriginalApprover() {
prepareActiveAssignment();
when(sourceTask.getAssignee()).thenReturn(String.valueOf(USER_ID));
when(processInstanceRecordService.lambdaQuery())
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(processInstanceBaseMapper));
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(processInstanceBaseMapper));
}
private void prepareActiveAssignment() {
when(sourceTask.getId()).thenReturn(TASK_ID);
when(processNodeRecordAssignUserService.lambdaQuery())
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(assignmentBaseMapper));
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(assignmentBaseMapper));
when(assignmentBaseMapper.selectOne(any())).thenReturn(originalAssignment());
}
@@ -388,20 +402,20 @@ class TaskServiceImplTests {
private ChangeActivityStateBuilder prepareTerminationFlow(ProcessInstanceRecord activeRecord) {
when(processInstanceRecordService.lambdaQuery())
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(processInstanceBaseMapper));
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(processInstanceBaseMapper));
when(processInstanceBaseMapper.selectList(any())).thenReturn(List.of());
if (activeRecord != null) {
when(processInstanceBaseMapper.selectOne(any())).thenReturn(activeRecord);
when(processNodeRecordAssignUserService.lambdaQuery())
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(assignmentBaseMapper));
.thenAnswer(invocation -> new LambdaQueryChainWrapper<>(assignmentBaseMapper));
when(assignmentBaseMapper.selectCount(any())).thenReturn(0L);
}
when(processInstanceRecordService.lambdaUpdate())
.thenAnswer(invocation -> new LambdaUpdateChainWrapper<>(processInstanceBaseMapper));
.thenAnswer(invocation -> new LambdaUpdateChainWrapper<>(processInstanceBaseMapper));
when(processNodeRecordService.lambdaUpdate())
.thenAnswer(invocation -> new LambdaUpdateChainWrapper<>(processNodeRecordMapper));
.thenAnswer(invocation -> new LambdaUpdateChainWrapper<>(processNodeRecordMapper));
when(processNodeRecordAssignUserService.lambdaUpdate())
.thenAnswer(invocation -> new LambdaUpdateChainWrapper<>(assignmentBaseMapper));
.thenAnswer(invocation -> new LambdaUpdateChainWrapper<>(assignmentBaseMapper));
ProcessInstanceQuery processInstanceQuery = mock(ProcessInstanceQuery.class, Answers.RETURNS_SELF);
when(runtimeService.createProcessInstanceQuery()).thenReturn(processInstanceQuery);
@@ -422,20 +436,19 @@ class TaskServiceImplTests {
String finishReason) {
var inOrder = inOrder(runtimeService, stateChangeBuilder);
inOrder.verify(runtimeService)
.setVariable(PROCESS_INSTANCE_ID, ProcessInstanceConstant.FINISH_REASON_VAR, finishReason);
.setVariable(PROCESS_INSTANCE_ID, ProcessInstanceConstant.FINISH_REASON_VAR, finishReason);
inOrder.verify(runtimeService).createChangeActivityStateBuilder();
inOrder.verify(stateChangeBuilder).moveExecutionsToSingleActivityId(List.of("execution-1"), "end");
inOrder.verify(stateChangeBuilder).changeState();
}
private ProcessNodeRecordAssignUser originalAssignment() {
return new ProcessNodeRecordAssignUser()
.setNodeRecordId(NODE_RECORD_ID)
.setTaskId(TASK_ID)
.setUserId(USER_ID)
.setInitialUserId(USER_ID)
.setAssignmentOrigin(ProcessInstanceConstant.ASSIGNMENT_ORIGIN_ORIGINAL)
.setStatus(NodeStatusEnum.JXZ.getCode());
return new ProcessNodeRecordAssignUser().setNodeRecordId(NODE_RECORD_ID)
.setTaskId(TASK_ID)
.setUserId(USER_ID)
.setInitialUserId(USER_ID)
.setAssignmentOrigin(ProcessInstanceConstant.ASSIGNMENT_ORIGIN_ORIGINAL)
.setStatus(NodeStatusEnum.JXZ.getCode());
}
private SysUser validUser(long userId, long tenantId) {
@@ -452,8 +465,8 @@ class TaskServiceImplTests {
private ITaskService transactionalService(RecordingTransactionManager transactionManager) {
ProxyFactory proxyFactory = new ProxyFactory(taskServiceImpl);
proxyFactory.addInterface(ITaskService.class);
proxyFactory.addAdvice(new TransactionInterceptor(transactionManager,
new AnnotationTransactionAttributeSource()));
proxyFactory
.addAdvice(new TransactionInterceptor(transactionManager, new AnnotationTransactionAttributeSource()));
return (ITaskService) proxyFactory.getProxy();
}
@@ -0,0 +1,40 @@
package com.pig4cloud.pigx.flow.support.node;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pig4cloud.pigx.common.core.util.R;
import com.pig4cloud.pigx.flow.constant.ProcessInstanceConstant;
import com.pig4cloud.pigx.flow.dto.Node;
import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
import org.flowable.engine.impl.persistence.entity.ExecutionEntityImpl;
import org.junit.jupiter.api.Test;
import java.util.Collections;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
class MultiInstanceHandlerTests {
@Test
void countersignWaitsForDynamicallyExpandedInstanceCount() {
IProcessNodeDataService nodeDataService = mock(IProcessNodeDataService.class);
Node node = new Node();
node.setMultipleMode(ProcessInstanceConstant.MULTIPLE_MODE_AL_SAME);
when(nodeDataService.getPublishedNode("flow-1:1:deployment", "approve-1")).thenReturn(R.ok(node));
MultiInstanceHandler handler = new MultiInstanceHandler(nodeDataService, Collections.emptyMap(),
new ObjectMapper());
ExecutionEntityImpl execution = mock(ExecutionEntityImpl.class);
when(execution.getProcessDefinitionId()).thenReturn("flow-1:1:deployment");
when(execution.getCurrentActivityId()).thenReturn("approve-1");
when(execution.getVariable("approve-1_approve_condition")).thenReturn(true);
when(execution.getVariable("nrOfInstances")).thenReturn(3);
when(execution.getVariable("nrOfCompletedInstances")).thenReturn(2);
assertThat(handler.completionCondition(execution)).isFalse();
when(execution.getVariable("nrOfCompletedInstances")).thenReturn(3);
assertThat(handler.completionCondition(execution)).isTrue();
}
}
@@ -26,6 +26,7 @@ class ApproveServiceTaskTests {
ExecutionEntityImpl execution = mock(ExecutionEntityImpl.class);
when(execution.getActivityId()).thenReturn("approve_service_task_approve-1");
when(execution.getProcessDefinitionKey()).thenReturn("leave");
when(execution.getProcessDefinitionId()).thenReturn("leave:1:deployment");
when(execution.getProcessInstanceId()).thenReturn("process-7598");
when(execution.getVariable("approve-1_approve_condition", Boolean.class)).thenReturn(false);
@@ -34,14 +35,14 @@ class ApproveServiceTaskTests {
when(runtimeService.createChangeActivityStateBuilder()).thenReturn(stateChangeBuilder);
when(stateChangeBuilder.processInstanceId("process-7598")).thenReturn(stateChangeBuilder);
when(stateChangeBuilder.moveActivityIdTo("approve_service_task_approve-1", "end"))
.thenReturn(stateChangeBuilder);
.thenReturn(stateChangeBuilder);
Refuse refuse = new Refuse();
refuse.setHandler("TO_END");
Node node = new Node();
node.setRefuse(refuse);
IProcessNodeDataService nodeDataService = mock(IProcessNodeDataService.class);
when(nodeDataService.getNodeData("leave", "approve-1")).thenReturn(R.ok(node));
when(nodeDataService.getPublishedNode("leave:1:deployment", "approve-1")).thenReturn(R.ok(node));
ApplicationContext previousContext = SpringUtil.getApplicationContext();
GenericApplicationContext context = new GenericApplicationContext();
@@ -58,8 +59,9 @@ class ApproveServiceTaskTests {
}
InOrder inOrder = inOrder(runtimeService, stateChangeBuilder);
inOrder.verify(runtimeService).setVariable("process-7598", ProcessInstanceConstant.FINISH_REASON_VAR,
ProcessInstanceConstant.FINISH_REASON_REFUSE);
inOrder.verify(runtimeService)
.setVariable("process-7598", ProcessInstanceConstant.FINISH_REASON_VAR,
ProcessInstanceConstant.FINISH_REASON_REFUSE);
inOrder.verify(stateChangeBuilder).moveActivityIdTo("approve_service_task_approve-1", "end");
inOrder.verify(stateChangeBuilder).changeState();
verify(runtimeService).createChangeActivityStateBuilder();