feat(flow): 条件分支支持多选(包容网关)

- 条件分支节点新增 multiMatch,开启后分叉与合并网关改为 InclusiveGateway,未开启保持排他网关
- 驳回到发起人时收回实例下全部执行流,修复并行、多选分支内驳回后兄弟分支残留导致汇聚异常
- 新增 H2 引擎测试覆盖命中多条、兜底、空分支、嵌套条件及分支内驳回

(cherry picked from commit b982714e1875632fde3338bc69f27052cad98a9d)
This commit is contained in:
PIGCLOUD committed 2026-11-03 14:56:21 +08:00
1 parent 93c52e41dc
commit 2f089f029f
4 files changed
+288 -4

No files matched your search

@@ -124,6 +124,11 @@ public class Node {
*/
private List<Node> conditionNodes;
/**
* 条件分支是否允许同时命中多条 true=包容网关,满足条件的分支都执行;空或 false=排他网关,只走一条
*/
private Boolean multiMatch;
/**
* 表单字段权限配置 key=表单字段ID,value=权限类型(H=隐藏,R=只读,E=编辑)
*/
@@ -10,6 +10,9 @@ import org.flowable.engine.RuntimeService;
import org.flowable.engine.delegate.DelegateExecution;
import org.flowable.engine.delegate.JavaDelegate;
import org.flowable.engine.impl.persistence.entity.ExecutionEntityImpl;
import org.flowable.engine.runtime.Execution;
import java.util.List;
/**
* 审批服务任务
@@ -96,8 +99,15 @@ public class ApproveServiceTask implements JavaDelegate {
}
else if (StrUtil.equals(rejectionHandler, "TO_START")) {
// 驳回到发起人:设置驳回标识变量,跳转到 root 节点
// 并行、多选分支内驳回时,兄弟分支执行流需一并收回,否则重新提交后汇聚网关会多出令牌
runtimeService.setVariable(processInstanceId, "rejectToStarter", true);
stateChangeBuilder.moveActivityIdTo(currentActivityId, "root").changeState();
List<String> executionIds = runtimeService.createExecutionQuery()
.parentId(processInstanceId)
.list()
.stream()
.map(Execution::getId)
.toList();
stateChangeBuilder.moveExecutionsToSingleActivityId(executionIds, "root").changeState();
}
else {
// 默认行为:重定向到结束节点
@@ -522,6 +522,7 @@ public class ModelUtil {
* 构建排他网关
* <p>
* 创建排他网关(条件分支)节点和对应的合并网关。 排他网关用于根据条件只选择一条执行路径,避免多个条件同时为真时并行进入多个分支。
* 开启 multiMatch 时改用包容网关,满足条件的分支都执行,合并网关等待所有已激活分支到达。
* @param node 排他网关节点数据
* @return 包含分叉网关和合并网关的元素列表
*/
@@ -530,14 +531,15 @@ public class ModelUtil {
node.setTailId(StrUtil.format("{}_merge_gateway", node.getId()));
List<FlowElement> flowElementList = new ArrayList<>();
boolean multiMatch = Boolean.TRUE.equals(node.getMultiMatch());
ExclusiveGateway exclusiveGateway = new ExclusiveGateway();
Gateway exclusiveGateway = multiMatch ? new InclusiveGateway() : new ExclusiveGateway();
exclusiveGateway.setId(node.getId());
exclusiveGateway.setName(node.getName());
flowElementList.add(exclusiveGateway);
// 合并网关
ExclusiveGateway gateway = new ExclusiveGateway();
// 合并网关,必须与分叉网关同类型,否则包容分叉的多条分支会让后续节点重复执行
Gateway gateway = multiMatch ? new InclusiveGateway() : new ExclusiveGateway();
gateway.setId(StrUtil.format("{}_merge_gateway", node.getId()));
gateway.setName(StrUtil.format("{}_合并网关", node.getName()));
flowElementList.add(gateway);
@@ -0,0 +1,267 @@
package com.pig4cloud.pigx.flow.support.utils;
import cn.hutool.extra.spring.SpringUtil;
import com.pig4cloud.pigx.common.core.util.R;
import com.pig4cloud.pigx.flow.dto.Node;
import com.pig4cloud.pigx.flow.dto.Refuse;
import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
import org.flowable.bpmn.model.BpmnModel;
import org.flowable.bpmn.model.ExclusiveGateway;
import org.flowable.bpmn.model.InclusiveGateway;
import org.flowable.bpmn.model.UserTask;
import org.flowable.engine.ProcessEngine;
import org.flowable.engine.ProcessEngineConfiguration;
import org.flowable.engine.runtime.ProcessInstance;
import org.flowable.task.api.Task;
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.springframework.context.ApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.flowable.spring.SpringProcessEngineConfiguration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* 条件分支多选(包容网关)在真实 Flowable 引擎上的执行语义。
*/
class ModelUtilMultiMatchTests {
private ProcessEngine engine;
private IProcessNodeDataService nodes;
private ApplicationContext previousContext;
private GenericApplicationContext context;
@BeforeEach
void setUp() {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
dataSource.setUser("sa");
SpringProcessEngineConfiguration configuration = new SpringProcessEngineConfiguration();
configuration.setDataSource(dataSource);
configuration.setTransactionManager(new DataSourceTransactionManager(dataSource));
configuration.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE);
configuration.setDisableIdmEngine(true);
configuration.setDisableEventRegistry(true);
configuration.setAsyncExecutorActivate(false);
engine = configuration.buildProcessEngine();
nodes = mock(IProcessNodeDataService.class);
previousContext = SpringUtil.getApplicationContext();
context = new GenericApplicationContext();
context.getBeanFactory().registerSingleton("runtimeService", engine.getRuntimeService());
context.getBeanFactory().registerSingleton("processNodeDataService", nodes);
context.refresh();
new SpringUtil().setApplicationContext(context);
}
@AfterEach
void tearDown() {
new SpringUtil().setApplicationContext(previousContext);
context.close();
engine.close();
}
@Test
void multiMatchBuildsPairedInclusiveGateways() {
BpmnModel model = ModelUtil.buildBpmnModel(tree(true, approval("a"), approval("b"), approval("other")), "f",
"f");
assertThat(model.getMainProcess().getFlowElement("gw")).isInstanceOf(InclusiveGateway.class);
assertThat(model.getMainProcess().getFlowElement("gw_merge_gateway")).isInstanceOf(InclusiveGateway.class);
BpmnModel single = ModelUtil.buildBpmnModel(tree(false, approval("a"), approval("b"), approval("other")),
"f", "f");
assertThat(single.getMainProcess().getFlowElement("gw")).isInstanceOf(ExclusiveGateway.class);
assertThat(single.getMainProcess().getFlowElement("gw_merge_gateway")).isInstanceOf(ExclusiveGateway.class);
}
@Test
void multiMatchRunsEveryHitBranchAndJoinsOnce() {
ProcessInstance instance = submit(tree(true, approval("a"), approval("b"), approval("other")), 200, "it");
assertThat(activeTaskKeys(instance)).containsExactlyInAnyOrder("a", "b");
complete(instance, "a");
assertThat(activeTaskKeys(instance)).containsExactly("b");
complete(instance, "b");
assertThat(activeTaskKeys(instance)).containsExactly("after");
}
@Test
void multiMatchFallsBackToOtherBranchWhenNothingHits() {
ProcessInstance instance = submit(tree(true, approval("a"), approval("b"), approval("other")), 1, "hr");
assertThat(activeTaskKeys(instance)).containsExactly("other");
complete(instance, "other");
assertThat(activeTaskKeys(instance)).containsExactly("after");
}
@Test
void multiMatchEmptyHitBranchGoesStraightToJoin() {
ProcessInstance instance = submit(tree(true, approval("a"), null, approval("other")), 200, "it");
assertThat(activeTaskKeys(instance)).containsExactly("a");
complete(instance, "a");
assertThat(activeTaskKeys(instance)).containsExactly("after");
}
@Test
void multiMatchSupportsNestedConditionBranch() {
Node nested = new Node();
nested.setId("inner");
nested.setType(4);
nested.setName("inner");
nested.setConditionNodes(List.of(branch("inner_hit", "amount > 100", approval("b1")),
branch("inner_else", null, approval("b2"))));
ProcessInstance instance = submit(tree(true, approval("a"), nested, approval("other")), 200, "it");
assertThat(activeTaskKeys(instance)).containsExactlyInAnyOrder("a", "b1");
complete(instance, "b1");
complete(instance, "a");
assertThat(activeTaskKeys(instance)).containsExactly("after");
}
@Test
void singleMatchKeepsOnlyFirstHitBranch() {
ProcessInstance instance = submit(tree(false, approval("a"), approval("b"), approval("other")), 200, "it");
assertThat(activeTaskKeys(instance)).containsExactly("a");
}
@Test
void rejectToStarterInsideMultiMatchDropsSiblingBranches() {
Refuse refuse = new Refuse();
refuse.setHandler("TO_START");
Node rejectConfig = new Node();
rejectConfig.setRefuse(refuse);
when(nodes.getPublishedNode(anyString(), eq("a"))).thenReturn(R.ok(rejectConfig));
ProcessInstance instance = submit(tree(true, approval("a"), approval("b"), approval("other")), 200, "it");
engine.getTaskService().complete(task(instance, "a").getId(), Map.of("a_approve_condition", false));
assertThat(activeTaskKeys(instance)).containsExactly("root");
engine.getTaskService().complete(task(instance, "root").getId(), Map.of("a_approve_condition", true));
assertThat(activeTaskKeys(instance)).containsExactlyInAnyOrder("a", "b");
complete(instance, "a");
complete(instance, "b");
assertThat(activeTaskKeys(instance)).containsExactly("after");
}
@Test
void rejectToStarterInsideParallelDropsSiblingBranches() {
Refuse refuse = new Refuse();
refuse.setHandler("TO_START");
Node rejectConfig = new Node();
rejectConfig.setRefuse(refuse);
when(nodes.getPublishedNode(anyString(), eq("a"))).thenReturn(R.ok(rejectConfig));
Node root = tree(false, approval("a"), approval("b"), approval("other"));
Node parallel = root.getChildren();
parallel.setType(5);
parallel.setConditionNodes(List.of(branch("ca", null, approval("a")), branch("cb", null, approval("b"))));
ProcessInstance instance = submit(root, 200, "it");
engine.getTaskService().complete(task(instance, "a").getId(), Map.of("a_approve_condition", false));
assertThat(activeTaskKeys(instance)).containsExactly("root");
engine.getTaskService().complete(task(instance, "root").getId(), Map.of("a_approve_condition", true));
complete(instance, "a");
complete(instance, "b");
assertThat(activeTaskKeys(instance)).containsExactly("after");
}
/**
* root → gw(a: amount > 100 | b: dept == 'it' | 其他情况) → after → end
*/
private Node tree(boolean multiMatch, Node a, Node b, Node other) {
Node gateway = new Node();
gateway.setId("gw");
gateway.setType(4);
gateway.setName("gw");
gateway.setMultiMatch(multiMatch);
gateway.setConditionNodes(List.of(branch("ca", "amount > 100", a), branch("cb", "dept == 'it'", b),
branch("cother", null, other)));
gateway.setChildren(approval("after"));
Node root = new Node();
root.setId("root");
root.setType(0);
root.setName("root");
root.setChildren(gateway);
return root;
}
private Node branch(String id, String expression, Node child) {
Node branch = new Node();
branch.setId(id);
branch.setType(3);
branch.setName(id);
branch.setConditionType(1);
branch.setExpressionText(expression);
branch.setChildren(child);
return branch;
}
private Node approval(String id) {
Node approval = new Node();
approval.setId(id);
approval.setType(1);
approval.setName(id);
approval.setMultipleMode(1);
return approval;
}
/**
* 部署后发起并提交发起人节点;剥离监听器和多实例,只保留真实网关、连线和审批服务任务。
*/
private ProcessInstance submit(Node root, int amount, String dept) {
BpmnModel model = ModelUtil.buildBpmnModel(root, "f", "f");
model.getMainProcess().setEventListeners(new ArrayList<>());
for (var element : model.getMainProcess().getFlowElements()) {
if (element instanceof UserTask task) {
task.setTaskListeners(new ArrayList<>());
task.setLoopCharacteristics(null);
task.setAssignee("9000");
}
}
engine.getRepositoryService().createDeployment().addBpmnModel("f.bpmn20.xml", model).deploy();
ProcessInstance instance = engine.getRuntimeService()
.startProcessInstanceByKey("f", Map.of("amount", amount, "dept", dept));
complete(instance, "root");
return instance;
}
private void complete(ProcessInstance instance, String key) {
engine.getTaskService().complete(task(instance, key).getId());
}
private Task task(ProcessInstance instance, String key) {
return engine.getTaskService()
.createTaskQuery()
.processInstanceId(instance.getId())
.taskDefinitionKey(key)
.singleResult();
}
private List<String> activeTaskKeys(ProcessInstance instance) {
return engine.getTaskService()
.createTaskQuery()
.processInstanceId(instance.getId())
.list()
.stream()
.map(Task::getTaskDefinitionKey)
.toList();
}
}