161 lines
7.1 KiB
Java
161 lines
7.1 KiB
Java
package com.yunhai.flowcenter.engine;
|
||
|
||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
import com.yunhai.platform.openapi.core.TargetSpec;
|
||
import com.yunhai.platform.openapi.core.YschOpenCall;
|
||
import com.yunhai.flowcenter.entity.FlwRouter;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import java.util.HashMap;
|
||
import java.util.List;
|
||
import java.util.Map;
|
||
|
||
/**
|
||
* 条件路由判断调用客户端（2026-09-24 §4.3）：POST {目录解析 host}{basePath}{path}?schId=，
|
||
* 身份即本应用 appKey（100019）——签名拦截器由 {@link YschOpenCall} 装配链注入，与回调推送同构。
|
||
* 寻址：service_key 走服务目录三级解析（targets > services > 枚举兜底）；
|
||
* flw_router.base_path 非空时覆盖目录条目的 context-path（网关形态目录无前缀、实际服务有的场景），
|
||
* 漏填必 401（规范 §4.2 同款坑）。<b>direct_host 非空的自注册行跳过目录解析直连</b>（2026-10-12 收口）。
|
||
* 信封解析复用 SDK（{code,msg,data}）。
|
||
*
|
||
* <p>重试纪律：同步 3s×3 次退避重试（500ms/1500ms），业务方义务=幂等纯读轻实现。
|
||
* 全部失败抛 {@link JudgeUnavailableException}（由调用方落 outlet=NULL 留痕并挂起）。
|
||
*/
|
||
@Slf4j
|
||
@Service
|
||
public class RouteJudgeClient {
|
||
|
||
/** 判断接口单次超时（毫秒）与重试次数/退避（设计 §4.4：总预算 3s×3 退避重试） */
|
||
static final int TIMEOUT_MS = 3_000;
|
||
static final int MAX_ATTEMPTS = 3;
|
||
private static final long[] BACKOFF_MS = {500L, 1500L};
|
||
|
||
private final YschOpenCall openCall;
|
||
private final ObjectMapper objectMapper;
|
||
|
||
public RouteJudgeClient(YschOpenCall openCall, ObjectMapper objectMapper) {
|
||
this.openCall = openCall;
|
||
this.objectMapper = objectMapper;
|
||
}
|
||
|
||
/** 判断请求体（契约冻结：query 带 schId，业务方按 outlets 返回 {outlet}；submitterId 2026-10-10 增补，老实现忽略即可） */
|
||
public static class JudgeBody {
|
||
public final Long schId;
|
||
public final Long instanceId;
|
||
public final Integer instanceVersion;
|
||
public final String bizType;
|
||
public final String bizId;
|
||
public final Integer submitterId;
|
||
public final String nodeId;
|
||
public final String nodeName;
|
||
public final String routerKey;
|
||
public final List<String> outlets;
|
||
public final Map<String, Object> variables;
|
||
|
||
public JudgeBody(Long schId, Long instanceId, Integer instanceVersion, String bizType, String bizId,
|
||
Integer submitterId, String nodeId, String nodeName, String routerKey,
|
||
List<String> outlets, Map<String, Object> variables) {
|
||
this.schId = schId;
|
||
this.instanceId = instanceId;
|
||
this.instanceVersion = instanceVersion;
|
||
this.bizType = bizType;
|
||
this.bizId = bizId;
|
||
this.submitterId = submitterId;
|
||
this.nodeId = nodeId;
|
||
this.nodeName = nodeName;
|
||
this.routerKey = routerKey;
|
||
this.outlets = outlets;
|
||
this.variables = variables;
|
||
}
|
||
}
|
||
|
||
/** 判断结果：出口名 + 原始返回（data 节点序列化，审计落 flw_route_log.raw_response） */
|
||
public static class JudgeResult {
|
||
public final String outlet;
|
||
public final String rawResponse;
|
||
|
||
public JudgeResult(String outlet, String rawResponse) {
|
||
this.outlet = outlet;
|
||
this.rawResponse = rawResponse;
|
||
}
|
||
}
|
||
|
||
/** 判断服务不可用（重试耗尽/网络/信封错误）；outlet 未命中由调用方判定，不走本异常 */
|
||
public static class JudgeUnavailableException extends RuntimeException {
|
||
public JudgeUnavailableException(String message, Throwable cause) {
|
||
super(message, cause);
|
||
}
|
||
}
|
||
|
||
@SuppressWarnings("unchecked")
|
||
public JudgeResult call(FlwRouter router, JudgeBody body) {
|
||
TargetSpec spec = resolveTarget(router);
|
||
Map<String, String> query = new HashMap<>();
|
||
query.put("schId", String.valueOf(body.schId));
|
||
RuntimeException last = null;
|
||
for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
|
||
try {
|
||
Map<String, Object> data = openCall.post(spec, router.getPath(), query, body, Map.class);
|
||
String raw = objectMapper.writeValueAsString(data);
|
||
Object outlet = data == null ? null : data.get("outlet");
|
||
if (outlet == null || String.valueOf(outlet).trim().isEmpty()) {
|
||
throw new IllegalStateException("判断接口未返回 outlet：" + raw);
|
||
}
|
||
return new JudgeResult(String.valueOf(outlet).trim(), raw);
|
||
} catch (RuntimeException e) {
|
||
last = e;
|
||
log.warn("[flow-center] 路由判断调用失败（第 {}/{} 次）：router={} {}",
|
||
attempt, MAX_ATTEMPTS, router.getRouterKey(), e.toString());
|
||
if (attempt < MAX_ATTEMPTS) {
|
||
sleep(BACKOFF_MS[attempt - 1]);
|
||
}
|
||
} catch (Exception e) {
|
||
last = new IllegalStateException(e.toString(), e);
|
||
log.warn("[flow-center] 路由判断调用异常（第 {}/{} 次）：router={} {}",
|
||
attempt, MAX_ATTEMPTS, router.getRouterKey(), e.toString());
|
||
if (attempt < MAX_ATTEMPTS) {
|
||
sleep(BACKOFF_MS[attempt - 1]);
|
||
}
|
||
}
|
||
}
|
||
throw new JudgeUnavailableException("判断服务调用失败（已重试 " + MAX_ATTEMPTS + " 次）", last);
|
||
}
|
||
|
||
/**
|
||
* 目标解析：
|
||
* <ul>
|
||
* <li>{@code direct_host} 非空（2026-10-12 starter 自注册行）→ {@link TargetSpec#fromUrl(String)}
|
||
* 拆基址直连，<b>跳过服务目录三级解析</b>（service_key/base_path 手工核对坑消失）；</li>
|
||
* <li>否则（存量手工行）：目录三级 + flw_router.base_path 覆盖（空=沿用目录条目前缀），行为逐字不变。</li>
|
||
* </ul>
|
||
* 两分支同设判断专用 3s 超时。
|
||
*/
|
||
private TargetSpec resolveTarget(FlwRouter router) {
|
||
String directHost = router.getDirectHost();
|
||
if (directHost != null && !directHost.trim().isEmpty()) {
|
||
return withJudgeTimeout(TargetSpec.fromUrl(directHost.trim()));
|
||
}
|
||
TargetSpec resolved = openCall.resolveTarget(router.getServiceKey());
|
||
String basePath = resolved.getBasePath();
|
||
if (router.getBasePath() != null && !router.getBasePath().trim().isEmpty()) {
|
||
basePath = router.getBasePath().trim();
|
||
}
|
||
return withJudgeTimeout(TargetSpec.of(resolved.getBaseHost(), basePath));
|
||
}
|
||
|
||
private static TargetSpec withJudgeTimeout(TargetSpec spec) {
|
||
spec.setConnectTimeoutMs((long) TIMEOUT_MS);
|
||
spec.setReadTimeoutMs((long) TIMEOUT_MS);
|
||
return spec;
|
||
}
|
||
|
||
private static void sleep(long ms) {
|
||
try {
|
||
Thread.sleep(ms);
|
||
} catch (InterruptedException e) {
|
||
Thread.currentThread().interrupt();
|
||
}
|
||
}
|
||
}
|