Files
flow-center-service/src/main/java/com/yunhai/flowcenter/engine/BranchScope.java
T
chenxing d2bef20e13
build / build-push (push) Successful in 47s
feat(engine): NOTIFY 抄送 + CONDITION 条件路由 + JOIN scope 修复(两批全栈)
- NOTIFY:TYPE_NOTIFY 瞬时抄送(flw_cc 逐人留痕 + CC_SENT 事件 + 站内通知不加待办角标)、/api/flow/cc/my、轨迹 kind=CC 与 oapi progress 抄送步、示意图青框卡片
- CONDITION:flw_router 判断服务注册 + flw_route_log 决策留痕(REQUIRES_NEW,reroute 回滚不丢失败行)+ 同步判断调用(服务目录寻址 + 平台签名,3s×3 退避重试)+ 失败挂起/管理端重跑(不配默认出口)+ 可选事件 flow.route.decided
- JOIN scope:branch_key 路径前缀编码 + scope 级到齐放行 + DefValidator SPLIT/JOIN 配对与键长守卫(存量无点号键零迁移,历史直连图回退实例级口径)
- 管理端:/api/flow/admin/router/* CRUD + POST /api/flow/admin/instance/{id}/reroute
- 生产 SQL:sql/20260923_create_flw_cc.sql(含 flw_task 时间列升 DATETIME(3))、20260924_create_flw_router_and_route_log.sql
- 测试:全量 125 绿(FlowEngineIT 24,含条件/嵌套并行/退回重走用例)
2026-09-24 08:58:26 +08:00

145 lines
6.6 KiB
Java
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package com.yunhai.flowcenter.engine;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
/**
* 分支归属判定(2026-09-24 §4.6 升级为路径前缀编码):节点属于哪个并行分支——
* 顶层分支键=分支头节点 id(无点号,与存量单层完全一致),内层分支键=「外层键.分支头id」
* (如 tb.tc,外层是内层的前缀),主线与汇聚链之后属 null。
* 运行时用于推导分支上下文({@link FlowEngine}:建行/到达标记/scope 级放行),
* 发布期用于拒绝跨分支连线({@link DefValidator})——两处必须同口径,故提取为静态工具。
*
* <p>嵌套口径(内层 JOIN 先于外层放行的关键):上溯遇 JOIN 视为透明块——跳到该 JOIN
* 匹配 SPLIT 的上游继续走(内层块整体属外层某分支,块的出口链沿用外层上下文)。
*/
public final class BranchScope {
/** 上溯递归深度上限:发布期已拦环,此处仅对未校验脏图防御性兜底 */
private static final int MAX_WALK_DEPTH = 64;
private BranchScope() {
}
/**
* 节点的完整分支键(路径前缀编码)。沿入边反向 BFS 到最近的上游 PARALLEL_SPLIT,
* 返回「该 split 自身所在分支键.本链分支头id」;顶层 split 的分支头即键本身(存量兼容)。
* 遇 START 或(透明跳越后的)耗尽即停——主线节点返回 null。
*/
public static String keyOf(FlowDefModel model, String nodeId) {
return keyOf(model, nodeId, new HashMap<String, String>(), new HashMap<String, FlowDefModel.Node>(), 0);
}
/**
* JOIN 的匹配分发节点:沿入边上溯(内层 JOIN 块透明跳越)首个 PARALLEL_SPLIT 即匹配者。
* 嵌套结构下内层 JOIN 必然先命中内层 split(内层先放行),外层 JOIN 透过内层块命中外层 split。
* 找不到(如主线节点直连 JOIN 的历史脏图)返回 null,由调用方决定兜底口径。
*/
public static FlowDefModel.Node matchedSplitOf(FlowDefModel model, String joinNodeId) {
return matchedSplitOf(model, joinNodeId, new HashMap<String, String>(), new HashMap<String, FlowDefModel.Node>(), 0);
}
/** 两节点是否同属一个分支上下文(含"都属主线"=都属于 null) */
public static boolean sameScope(FlowDefModel model, String a, String b) {
String ka = keyOf(model, a);
String kb = keyOf(model, b);
return ka == null ? kb == null : ka.equals(kb);
}
private static String keyOf(FlowDefModel model, String nodeId,
Map<String, String> keyMemo, Map<String, FlowDefModel.Node> splitMemo, int depth) {
if (model == null || nodeId == null || depth > MAX_WALK_DEPTH) {
return null;
}
if (keyMemo.containsKey(nodeId)) {
return keyMemo.get(nodeId);
}
keyMemo.put(nodeId, null); // 占位防环:递归回访时按主线(null)处理并终止
String result = null;
Set<String> visited = new HashSet<>();
visited.add(nodeId);
Deque<String> queue = new ArrayDeque<>();
queue.add(nodeId);
while (!queue.isEmpty() && result == null) {
String cur = queue.poll();
for (FlowDefModel.Edge e : model.inEdges(cur)) {
FlowDefModel.Node from = model.node(e.from);
if (from == null) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_SPLIT.equals(from.type)) {
// 本链分支头确定:再取该 split 自身所在的外层上下文,拼成路径前缀
String outer = keyOf(model, from.id, keyMemo, splitMemo, depth + 1);
result = outer == null ? e.to : outer + "." + e.to;
break;
}
if (FlowDefModel.TYPE_START.equals(from.type)) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(from.type)) {
// 内层汇聚块透明:从其匹配 split 的上游继续上溯(块出口链沿用外层上下文)
FlowDefModel.Node innerSplit = matchedSplitOf(model, from.id, keyMemo, splitMemo, depth + 1);
if (innerSplit != null && visited.add(innerSplit.id)) {
queue.add(innerSplit.id);
}
continue;
}
if (visited.add(e.from)) {
queue.add(e.from);
}
}
}
keyMemo.put(nodeId, result);
return result;
}
private static FlowDefModel.Node matchedSplitOf(FlowDefModel model, String joinNodeId,
Map<String, String> keyMemo,
Map<String, FlowDefModel.Node> splitMemo, int depth) {
if (model == null || joinNodeId == null || depth > MAX_WALK_DEPTH) {
return null;
}
if (splitMemo.containsKey(joinNodeId)) {
return splitMemo.get(joinNodeId);
}
splitMemo.put(joinNodeId, null); // 占位防环
FlowDefModel.Node result = null;
Set<String> visited = new HashSet<>();
visited.add(joinNodeId);
Deque<String> queue = new ArrayDeque<>();
queue.add(joinNodeId);
while (!queue.isEmpty() && result == null) {
String cur = queue.poll();
for (FlowDefModel.Edge e : model.inEdges(cur)) {
FlowDefModel.Node from = model.node(e.from);
if (from == null) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_SPLIT.equals(from.type)) {
result = from;
break;
}
if (FlowDefModel.TYPE_START.equals(from.type)) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(from.type)) {
FlowDefModel.Node innerSplit = matchedSplitOf(model, from.id, keyMemo, splitMemo, depth + 1);
if (innerSplit != null && visited.add(innerSplit.id)) {
queue.add(innerSplit.id);
}
continue;
}
if (visited.add(e.from)) {
queue.add(e.from);
}
}
}
splitMemo.put(joinNodeId, result);
return result;
}
}