diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java index 2c58db2e2..4a075e447 100644 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java @@ -124,6 +124,11 @@ public class Node { */ private List conditionNodes; + /** + * 条件分支是否允许同时命中多条 true=包容网关，满足条件的分支都执行；空或 false=排他网关，只走一条 + */ + private Boolean multiMatch; + /** * 表单字段权限配置 key=表单字段ID，value=权限类型（H=隐藏，R=只读，E=编辑） */ diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/tasks/ApproveServiceTask.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/tasks/ApproveServiceTask.java index 719d64c11..aa41fdc95 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/tasks/ApproveServiceTask.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/tasks/ApproveServiceTask.java @@ -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 executionIds = runtimeService.createExecutionQuery() + .parentId(processInstanceId) + .list() + .stream() + .map(Execution::getId) + .toList(); + stateChangeBuilder.moveExecutionsToSingleActivityId(executionIds, "root").changeState(); } else { // 默认行为：重定向到结束节点 diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/ModelUtil.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/ModelUtil.java index ff0163b4c..e53472930 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/ModelUtil.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/ModelUtil.java @@ -522,6 +522,7 @@ public class ModelUtil { * 构建排他网关 *

* 创建排他网关（条件分支）节点和对应的合并网关。 排他网关用于根据条件只选择一条执行路径，避免多个条件同时为真时并行进入多个分支。 + * 开启 multiMatch 时改用包容网关，满足条件的分支都执行，合并网关等待所有已激活分支到达。 * @param node 排他网关节点数据 * @return 包含分叉网关和合并网关的元素列表 */ @@ -530,14 +531,15 @@ public class ModelUtil { node.setTailId(StrUtil.format("{}_merge_gateway", node.getId())); List 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); diff --git a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/support/utils/ModelUtilMultiMatchTests.java b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/support/utils/ModelUtilMultiMatchTests.java new file mode 100644 index 000000000..8c101f98f --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/support/utils/ModelUtilMultiMatchTests.java @@ -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 activeTaskKeys(ProcessInstance instance) { + return engine.getTaskService() + .createTaskQuery() + .processInstanceId(instance.getId()) + .list() + .stream() + .map(Task::getTaskDefinitionKey) + .toList(); + } + +}