zgty/zt-dsc
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'dev' into test

Browse Source
This commit is contained in:
chenbowen
2025-12-02 17:49:15 +08:00
parent 8d1a6ed045 0846e4a56a
commit 6eda833c55
23 changed files with 736 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
deployment.yaml
View File

@@ -422,3 +422,77 @@ spec:
port: 48100
targetPort: 48100
nodePort: 30090
---
# zt-module-template
apiVersion: apps/v1
kind: Deployment
metadata:
namespace: ns-d6a0e78ebd674c279614498e4c57b133
name: zt-module-template
labels:
app: zt-module-template
annotations:
version: "VERSION_PLACEHOLDER"
description: DESC_PLACEHOLDER
rollout.kubernetes.io/change-cause: "DESC_PLACEHOLDER:VERSION_PLACEHOLDER"
spec:
replicas: 1
selector:
matchLabels:
app: zt-module-template
template:
metadata:
labels:
app: zt-module-template
spec:
containers:
- name: zt-module-template
image: 172.16.46.66:10043/zt/zt-module-template:VERSION_PLACEHOLDER
imagePullPolicy: Always
env:
- name: TZ
value: Asia/Shanghai
readinessProbe:
httpGet:
path: /actuator/health
port: 48100
initialDelaySeconds: 50
periodSeconds: 5
failureThreshold: 3
livenessProbe:
httpGet:
path: /actuator/health
port: 48100
initialDelaySeconds: 50
periodSeconds: 10
failureThreshold: 5
resources:
requests:
cpu: "500m"
memory: "1024Mi"
limits:
cpu: "700m"
memory: "1024Mi"
terminationGracePeriodSeconds: 30
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
---
apiVersion: v1
kind: Service
metadata:
namespace: ns-d6a0e78ebd674c279614498e4c57b133
name: zt-module-template
spec:
type: NodePort
selector:
app: zt-module-template
ports:
- protocol: TCP
port: 48100
targetPort: 48100
nodePort: 30889

2
zt-framework/zt-spring-boot-starter-web/src/test/java/com/zt/plat/framework/desensitize/core/DesensitizeTest.java
View File

@@ -48,7 +48,7 @@ public class DesensitizeTest {
DesensitizeDemo d = JsonUtils.parseObject(JsonUtils.toJsonString(desensitizeDemo), DesensitizeDemo.class);
// 断言
assertNotNull(d);
assertEquals("芋***", d.getNickname());
assertEquals("Z***", d.getNickname());
assertEquals("998800********31", d.getBankCard());
assertEquals("粤A6***6", d.getCarLicense());
assertEquals("0108*****22", d.getFixedPhone());

8
zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/security/GatewaySecurityFilter.java
View File

@@ -273,10 +273,10 @@ public class GatewaySecurityFilter extends OncePerRequestFilter {
String signatureType = resolveSignatureType(credential, security);
try {
// boolean valid = CryptoSignatureUtils.verifySignature(signaturePayload, signatureType);
// if (!valid) {
// throw new SecurityValidationException(HttpStatus.UNAUTHORIZED, "签名校验失败");
// }
boolean valid = CryptoSignatureUtils.verifySignature(signaturePayload, signatureType);
if (!valid) {
throw new SecurityValidationException(HttpStatus.UNAUTHORIZED, "签名校验失败");
}
} catch (IllegalArgumentException ex) {
throw new SecurityValidationException(HttpStatus.INTERNAL_SERVER_ERROR, "签名算法配置异常");
}

7
zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/framework/integration/gateway/step/impl/HttpStepHandler.java
View File

@@ -393,9 +393,14 @@ public class HttpStepHandler implements ApiStepHandler {
}
private boolean supportsRequestBody(HttpMethod method) {
// 所有请求都要传递请求体
if (method == null) {
return true;
}
return !(HttpMethod.GET.equals(method)
|| HttpMethod.HEAD.equals(method)
|| HttpMethod.OPTIONS.equals(method)
|| HttpMethod.TRACE.equals(method));
}
private Mono<Object> applyResilientRetry(Mono<Object> responseMono, ApiStepDefinition stepDefinition) {
return responseMono.retryWhen(Retry.fixedDelay(RETRY_ATTEMPTS, RETRY_DELAY)

12
zt-module-databus/zt-module-databus-server/src/test/java/com/zt/plat/module/databus/controller/admin/gateway/ApiGatewayControllerTest.java
View File

@@ -2,8 +2,11 @@ package com.zt.plat.module.databus.controller.admin.gateway;
import com.zt.plat.module.databus.controller.admin.gateway.vo.ApiGatewayInvokeReqVO;
import com.zt.plat.module.databus.framework.integration.gateway.core.ApiGatewayExecutionService;
import com.zt.plat.module.databus.framework.integration.gateway.core.IntegrationFlowManager;
import com.zt.plat.module.databus.framework.integration.gateway.model.ApiGatewayResponse;
import com.zt.plat.module.databus.framework.integration.config.ApiGatewayProperties;
import com.zt.plat.module.databus.framework.integration.gateway.core.ApiGatewayAccessLogger;
import com.zt.plat.module.databus.service.gateway.ApiAnonymousUserService;
import com.zt.plat.module.databus.service.gateway.ApiClientCredentialService;
import com.zt.plat.module.databus.service.gateway.ApiDefinitionService;
import org.junit.jupiter.api.Test;
@@ -48,9 +51,18 @@ class ApiGatewayControllerTest {
@MockBean
private StringRedisTemplate stringRedisTemplate;
@MockBean
private IntegrationFlowManager integrationFlowManager;
@MockBean
private ApiClientCredentialService apiClientCredentialService;
@MockBean
private ApiAnonymousUserService apiAnonymousUserService;
@MockBean
private ApiGatewayAccessLogger apiGatewayAccessLogger;
@Test
void invokeShouldReturnGatewayEnvelope() throws Exception {
ApiGatewayResponse response = ApiGatewayResponse.builder()

17
zt-module-databus/zt-module-databus-server/src/test/java/com/zt/plat/module/databus/framework/integration/gateway/sample/DatabusApiInvocationExample.java
View File

@@ -28,16 +28,10 @@ import java.util.UUID;
public final class DatabusApiInvocationExample {
public static final String TIMESTAMP = Long.toString(System.currentTimeMillis());
// private static final String APP_ID = "ztmy";
// private static final String APP_SECRET = "zFre/nTRGi7LpoFjN7oQkKeOT09x1fWTyIswrc702QQ=";
// private static final String APP_ID = "test";
// private static final String APP_SECRET = "RSYtKXrXPLMy3oeh0cOro6QCioRUgqfnKCkDkNq78sI=";
private static final String APP_ID = "testAnnoy";
private static final String APP_SECRET = "jyGCymUjCFL2i3a4Tm3qBIkUrUl4ZgKPYvOU/47ZWcM=";
private static final String APP_ID = "ztmy";
private static final String APP_SECRET = "zFre/nTRGi7LpoFjN7oQkKeOT09x1fWTyIswrc702QQ=";
private static final String ENCRYPTION_TYPE = CryptoSignatureUtils.ENCRYPT_TYPE_AES;
// private static final String TARGET_API = "http://172.16.46.63:30081/admin-api/databus/api/portal/lgstOpenApi/v1";
private static final String TARGET_API = "http://127.0.0.1:48080/admin-api/databus/api/portal/test/1";
// private static final String TARGET_API = "http://127.0.0.1:48080/admin-api/databus/api/portal/lgstOpenApi/v1";
private static final String TARGET_API = "http://172.16.46.63:30081/admin-api/databus/api/portal/callback/v1";
private static final HttpClient HTTP_CLIENT = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(5))
.build();
@@ -88,8 +82,11 @@ public final class DatabusApiInvocationExample {
Map<String, Object> queryParams = new LinkedHashMap<>();
long extraTimestamp = 1761556157185L;
// String bodyJson = String.format("""
// {"operateFlag":"I","__interfaceType__":"R_MY_JY_03","data":{"endAddressName":"1","customerCompanyName":"中铜国贸","endAddressDetail":"测试地址","remark":" ","custSuppType":"1","shipperCompanyName":"中铜国贸","consigneeCorpCode":" ","consignerContactPhone":" 11","importFlag":"10","businessSupplierCode":" ","entrustMainCode":"WT3162251027027","endAddressCode":" ","specifyCarrierCorpCode":"10086689","materDetail":[{"detailStatus":"10","batchNo":"ZLTD2510ZTGM0017001","measureCodeMdm":"CU032110001","packType":" ","quantityPlanDetail":1,"deliveryOrderNo":"ZLTD2510ZTGM0017001","measureCode":"CU032110001","goodsSpecification":" ","measureUnitCode":"PAC","entrustDetailCode":"WT3162251027027001","brand":" ","soNumber":"68ecf0055502d565d22b378a"}],"operateFlag":1,"custSuppName":"上海锦生金属有限公司","startAddressCode":" ","planStartTime":1761556166000,"customerCompanyCode":0,"importMethod":"EXW","startAddressType":"10","shipperCompanyCode":"3162","deliverCondition":"20","businessSupplierName":" ","startAddressDetail":" 111","transType":"30","endAddressType":"20","planEndTime":1761556166000,"specifyCarrierCorpName":null,"custSuppFlag":"0101","businessType":"20","consigneeCorpName":" ","custSuppCode":"10086689","startAddressName":" 111","consignerContactName":" 11"},"datetime":"20251027170929","busiBillCode":"WT3162251027027","system":"BRMS","__requestId__":"f918841c-14fb-49eb-9640-c5d1b3d46bd1"}
// """, extraTimestamp);
String bodyJson = String.format("""
{"operateFlag":"I","__interfaceType__":"R_MY_JY_03","data":{"endAddressName":"1","customerCompanyName":"中铜国贸","endAddressDetail":"测试地址","remark":" ","custSuppType":"1","shipperCompanyName":"中铜国贸","consigneeCorpCode":" ","consignerContactPhone":" 11","importFlag":"10","businessSupplierCode":" ","entrustMainCode":"WT3162251027027","endAddressCode":" ","specifyCarrierCorpCode":"10086689","materDetail":[{"detailStatus":"10","batchNo":"ZLTD2510ZTGM0017001","measureCodeMdm":"CU032110001","packType":" ","quantityPlanDetail":1,"deliveryOrderNo":"ZLTD2510ZTGM0017001","measureCode":"CU032110001","goodsSpecification":" ","measureUnitCode":"PAC","entrustDetailCode":"WT3162251027027001","brand":" ","soNumber":"68ecf0055502d565d22b378a"}],"operateFlag":1,"custSuppName":"上海锦生金属有限公司","startAddressCode":" ","planStartTime":1761556166000,"customerCompanyCode":0,"importMethod":"EXW","startAddressType":"10","shipperCompanyCode":"3162","deliverCondition":"20","businessSupplierName":" ","startAddressDetail":" 111","transType":"30","endAddressType":"20","planEndTime":1761556166000,"specifyCarrierCorpName":null,"custSuppFlag":"0101","businessType":"20","consigneeCorpName":" ","custSuppCode":"10086689","startAddressName":" 111","consignerContactName":" 11"},"datetime":"20251027170929","busiBillCode":"WT3162251027027","system":"BRMS","__requestId__":"f918841c-14fb-49eb-9640-c5d1b3d46bd1"}
{}
""", extraTimestamp);
Map<String, Object> bodyParams = parseBodyJson(bodyJson);

45
zt-module-databus/zt-module-databus-server/src/test/java/com/zt/plat/module/databus/framework/integration/gateway/security/GatewaySecurityFilterTest.java
View File

@@ -9,6 +9,7 @@ import com.zt.plat.framework.security.core.util.SecurityFrameworkUtils;
import com.zt.plat.framework.web.core.util.WebFrameworkUtils;
import com.zt.plat.module.databus.dal.dataobject.gateway.ApiClientCredentialDO;
import com.zt.plat.module.databus.framework.integration.config.ApiGatewayProperties;
import com.zt.plat.module.databus.framework.integration.gateway.core.ApiGatewayAccessLogger;
import com.zt.plat.module.databus.service.gateway.ApiAnonymousUserService;
import com.zt.plat.module.databus.service.gateway.ApiClientCredentialService;
import jakarta.servlet.http.HttpServletRequest;
@@ -25,7 +26,9 @@ import org.springframework.security.core.context.SecurityContextHolder;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
import java.util.UUID;
import static com.zt.plat.module.databus.framework.integration.config.ApiGatewayProperties.*;
@@ -50,8 +53,9 @@ class GatewaySecurityFilterTest {
ApiClientCredentialService credentialService = mock(ApiClientCredentialService.class);
ApiAnonymousUserService anonymousUserService = mock(ApiAnonymousUserService.class);
when(anonymousUserService.issueAccessToken(any())).thenReturn(Optional.empty());
when(anonymousUserService.issueAccessToken(any())).thenReturn(Optional.empty());
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper());
ApiGatewayAccessLogger accessLogger = mock(ApiGatewayAccessLogger.class);
when(accessLogger.logEntrance(any())).thenReturn(1L);
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper(), accessLogger);
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/admin-api/databus/api/portal/demo/v1");
request.setRemoteAddr("127.0.0.1");
@@ -74,7 +78,9 @@ class GatewaySecurityFilterTest {
ApiClientCredentialService credentialService = mock(ApiClientCredentialService.class);
ApiAnonymousUserService anonymousUserService = mock(ApiAnonymousUserService.class);
when(anonymousUserService.issueAccessToken(any())).thenReturn(Optional.empty());
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper());
ApiGatewayAccessLogger accessLogger = mock(ApiGatewayAccessLogger.class);
when(accessLogger.logEntrance(any())).thenReturn(1L);
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper(), accessLogger);
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/admin-api/databus/api/portal/demo/v1");
request.setRemoteAddr("10.0.0.1");
@@ -97,6 +103,8 @@ class GatewaySecurityFilterTest {
ApiClientCredentialService credentialService = mock(ApiClientCredentialService.class);
ApiAnonymousUserService anonymousUserService = mock(ApiAnonymousUserService.class);
ApiGatewayAccessLogger accessLogger = mock(ApiGatewayAccessLogger.class);
when(accessLogger.logEntrance(any())).thenReturn(1L);
when(anonymousUserService.issueAccessToken(any())).thenReturn(Optional.empty());
ApiClientCredentialDO credential = new ApiClientCredentialDO();
credential.setAppId("demo-app");
@@ -108,13 +116,13 @@ class GatewaySecurityFilterTest {
properties.getSecurity().setRequireBodyEncryption(false);
properties.getSecurity().setEncryptResponse(false);
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper());
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper(), accessLogger);
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/admin-api/databus/api/portal/demo/v1");
request.setRemoteAddr("127.0.0.1");
long timestamp = System.currentTimeMillis();
String nonce = UUID.randomUUID().toString().replaceAll("-", "");
String signature = signatureForApp("demo-app");
String signature = signatureForApp("demo-app", timestamp);
request.addHeader(APP_ID_HEADER, "demo-app");
request.addHeader(TIMESTAMP_HEADER, String.valueOf(timestamp));
request.addHeader(NONCE_HEADER, nonce);
@@ -142,13 +150,15 @@ class GatewaySecurityFilterTest {
ApiClientCredentialService credentialService = mock(ApiClientCredentialService.class);
ApiAnonymousUserService anonymousUserService = mock(ApiAnonymousUserService.class);
ApiGatewayAccessLogger accessLogger = mock(ApiGatewayAccessLogger.class);
when(accessLogger.logEntrance(any())).thenReturn(1L);
ApiClientCredentialDO credential = new ApiClientCredentialDO();
credential.setAppId("demo-app");
credential.setEncryptionKey("demo-secret-key");
credential.setEncryptionType(CryptoSignatureUtils.ENCRYPT_TYPE_AES);
when(credentialService.findActiveCredential("demo-app")).thenReturn(Optional.of(credential));
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper());
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper(), accessLogger);
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/admin-api/databus/api/portal/demo/v1");
request.setRemoteAddr("127.0.0.1");
@@ -183,6 +193,8 @@ class GatewaySecurityFilterTest {
ApiClientCredentialService credentialService = mock(ApiClientCredentialService.class);
ApiAnonymousUserService anonymousUserService = mock(ApiAnonymousUserService.class);
ApiGatewayAccessLogger accessLogger = mock(ApiGatewayAccessLogger.class);
when(accessLogger.logEntrance(any())).thenReturn(1L);
ApiClientCredentialDO credential = new ApiClientCredentialDO();
credential.setAppId("demo-app");
credential.setSignatureType(null);
@@ -200,7 +212,7 @@ class GatewaySecurityFilterTest {
when(anonymousUserService.find(99L)).thenReturn(Optional.of(details));
when(anonymousUserService.issueAccessToken(details)).thenReturn(Optional.of("mock-token"));
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper());
GatewaySecurityFilter filter = new GatewaySecurityFilter(properties, redisTemplate, credentialService, anonymousUserService, new ObjectMapper(), accessLogger);
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/admin-api/databus/api/portal/demo/v1");
request.setRemoteAddr("127.0.0.1");
@@ -209,7 +221,7 @@ class GatewaySecurityFilterTest {
request.addHeader(APP_ID_HEADER, "demo-app");
request.addHeader(TIMESTAMP_HEADER, String.valueOf(timestamp));
request.addHeader(NONCE_HEADER, nonce);
request.addHeader(SIGNATURE_HEADER, signatureForApp("demo-app"));
request.addHeader(SIGNATURE_HEADER, signatureForApp("demo-app", timestamp));
MockHttpServletResponse response = new MockHttpServletResponse();
MockFilterChain chain = new MockFilterChain();
@@ -231,7 +243,20 @@ class GatewaySecurityFilterTest {
return properties;
}
private String signatureForApp(String appId) {
return SecureUtil.md5("appId=" + appId);
private String signatureForApp(String appId, long timestamp) {
Map<String, Object> payload = new TreeMap<>();
payload.put(APP_ID_HEADER, appId);
payload.put(TIMESTAMP_HEADER, String.valueOf(timestamp));
StringBuilder sb = new StringBuilder();
payload.forEach((key, value) -> {
if (value == null) {
return;
}
if (sb.length() > 0) {
sb.append('&');
}
sb.append(key).append('=').append(value);
});
return SecureUtil.md5(sb.toString());
}
}

360
zt-module-databus/zt-module-databus-server/src/test/java/com/zt/plat/module/databus/framework/integration/gateway/step/impl/HttpStepHandlerConnectionResetScenarioTest.java
View File

@@ -1,360 +0,0 @@
package com.zt.plat.module.databus.framework.integration.gateway.step.impl;
import com.zt.plat.framework.common.exception.ServiceException;
import org.junit.jupiter.api.Test;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.PrematureCloseException;
import reactor.netty.resources.ConnectionProvider;
import reactor.util.retry.Retry;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Demonstrates the stale-connection scenario using the legacy vs the deferred retry pipeline.
*/
class HttpStepHandlerConnectionResetScenarioTest {
private static final Duration RETRY_DELAY = Duration.ofMillis(200);
private static final int RETRY_ATTEMPTS = 3;
private static final Duration BLOCK_TIMEOUT = Duration.ofSeconds(5);
private static final Duration RESET_WAIT = Duration.ofMillis(300);
@Test
void legacyPipelineLosesSuccessfulRetry() throws Exception {
try (ResetOnceHttpServer server = new ResetOnceHttpServer()) {
WebClient webClient = createWebClient();
URI uri = server.uri("/demo");
warmUp(server, webClient, uri);
server.awaitWarmupConnectionReset(RESET_WAIT);
legacyInvoke(webClient, uri, Map.of("mode", "legacy"));
server.awaitFreshResponses(1, Duration.ofSeconds(2));
assertThat(server.getFreshResponseCount()).isEqualTo(1);
assertThat(server.getServedBodies()).contains("reset", "fresh");
}
}
@Test
void deferredPipelinePropagatesSuccessfulRetry() throws Exception {
try (ResetOnceHttpServer server = new ResetOnceHttpServer()) {
WebClient webClient = createWebClient();
URI uri = server.uri("/demo");
warmUp(server, webClient, uri);
server.awaitWarmupConnectionReset(RESET_WAIT);
Object result = deferredInvoke(webClient, uri, Map.of("mode", "defer"));
assertThat(result).isInstanceOf(Map.class);
Map<?, ?> resultMap = (Map<?, ?>) result;
assertThat(resultMap.get("stage")).isEqualTo("fresh");
server.awaitFreshResponses(1, Duration.ofSeconds(2));
assertThat(server.getFreshResponseCount()).isEqualTo(1);
assertThat(server.getServedBodies()).contains("reset", "fresh");
}
}
private WebClient createWebClient() {
ConnectionProvider provider = ConnectionProvider.builder("http-step-handler-demo")
.maxConnections(1)
.pendingAcquireMaxCount(-1)
.maxIdleTime(Duration.ofSeconds(5))
.build();
HttpClient httpClient = HttpClient.create(provider).compress(true);
return WebClient.builder()
.clientConnector(new ReactorClientHttpConnector(httpClient))
.build();
}
private void warmUp(ResetOnceHttpServer server, WebClient webClient, URI uri) {
webClient.post()
.uri(uri)
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(Map.of("warm", true))
.retrieve()
.bodyToMono(Object.class)
.block(BLOCK_TIMEOUT);
server.awaitWarmupResponse(Duration.ofSeconds(2));
}
private Object legacyInvoke(WebClient webClient, URI uri, Object body) {
WebClient.RequestHeadersSpec<?> spec = webClient.post()
.uri(uri)
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(body);
Mono<Object> responseMono = spec.retrieve()
.bodyToMono(Object.class)
// 模拟业务中首次订阅后缓存失败结果的场景
.cache();
return responseMono.retryWhen(Retry.fixedDelay(RETRY_ATTEMPTS, RETRY_DELAY)
.filter(this::isRetryableException)
.onRetryExhaustedThrow((specification, signal) -> signal.failure()))
.block(BLOCK_TIMEOUT);
}
private Object deferredInvoke(WebClient webClient, URI uri, Object body) {
Mono<Object> responseMono = Mono.defer(() -> webClient.post()
.uri(uri)
.contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON)
.bodyValue(body)
.retrieve()
.bodyToMono(Object.class)
// 通过 defer，每次重试都会重新创建带缓存的响应 Mono
.cache());
return responseMono.retryWhen(Retry.fixedDelay(RETRY_ATTEMPTS, RETRY_DELAY)
.filter(this::isRetryableException))
.block(BLOCK_TIMEOUT);
}
private boolean isRetryableException(Throwable throwable) {
if (throwable == null) {
return false;
}
Throwable cursor = throwable;
while (cursor != null) {
if (cursor instanceof ServiceException) {
return false;
}
if (cursor instanceof PrematureCloseException) {
return true;
}
if (cursor instanceof IOException) {
return true;
}
cursor = cursor.getCause();
}
return false;
}
private static final class ResetOnceHttpServer implements AutoCloseable {
private static final Duration RESET_DELAY = Duration.ofMillis(250);
private final ServerSocket serverSocket;
private final ExecutorService acceptExecutor;
private final ScheduledExecutorService scheduler;
private final AtomicInteger connectionCount = new AtomicInteger();
private final AtomicInteger freshResponses = new AtomicInteger();
private final CountDownLatch warmupResponseSent = new CountDownLatch(1);
private final CountDownLatch warmupReset = new CountDownLatch(1);
private final List<String> servedBodies = new CopyOnWriteArrayList<>();
private volatile boolean running = true;
private volatile Socket warmupSocket;
ResetOnceHttpServer() throws IOException {
this.serverSocket = new ServerSocket(0, 50, InetAddress.getByName("127.0.0.1"));
this.serverSocket.setReuseAddress(true);
this.acceptExecutor = Executors.newSingleThreadExecutor(r -> {
Thread t = new Thread(r, "reset-once-http-accept");
t.setDaemon(true);
return t;
});
this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
Thread t = new Thread(r, "reset-once-http-scheduler");
t.setDaemon(true);
return t;
});
acceptExecutor.submit(this::acceptLoop);
}
URI uri(String path) {
Objects.requireNonNull(path, "path");
if (!path.startsWith("/")) {
path = "/" + path;
}
return URI.create("http://127.0.0.1:" + serverSocket.getLocalPort() + path);
}
List<String> getServedBodies() {
return new ArrayList<>(servedBodies);
}
int getFreshResponseCount() {
return freshResponses.get();
}
void awaitWarmupResponse(Duration timeout) {
awaitLatch(warmupResponseSent, timeout);
}
void awaitWarmupConnectionReset(Duration timeout) {
awaitLatch(warmupReset, timeout);
}
void awaitFreshResponses(int expected, Duration timeout) {
long deadline = System.nanoTime() + timeout.toNanos();
while (freshResponses.get() < expected && System.nanoTime() < deadline) {
try {
Thread.sleep(10);
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
return;
}
}
}
private void awaitLatch(CountDownLatch latch, Duration timeout) {
try {
if (!latch.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
throw new IllegalStateException("Timed out waiting for latch");
}
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
throw new IllegalStateException(ex);
}
}
private void acceptLoop() {
try {
while (running) {
Socket socket = serverSocket.accept();
int index = connectionCount.incrementAndGet();
handle(socket, index);
}
} catch (SocketException ex) {
if (running) {
throw new IllegalStateException("Unexpected server socket error", ex);
}
} catch (IOException ex) {
if (running) {
throw new IllegalStateException("I/O error in server", ex);
}
}
}
private void handle(Socket socket, int index) {
try {
socket.setTcpNoDelay(true);
RequestMetadata metadata = readRequest(socket);
if (index == 1) {
warmupSocket = socket;
String body = "{\"stage\":\"warmup\",\"path\":\"" + metadata.path + "\"}";
writeResponse(socket, body, true);
servedBodies.add("warmup");
warmupResponseSent.countDown();
scheduler.schedule(() -> forceReset(socket), RESET_DELAY.toMillis(), TimeUnit.MILLISECONDS);
} else if (index == 2) {
// 模拟客户端复用到仍在连接池中的旧连接，但服务端已在请求到达后立即复位。
servedBodies.add("reset");
scheduler.schedule(() -> closeWithReset(socket), 10, TimeUnit.MILLISECONDS);
} else {
String body = "{\"stage\":\"fresh\",\"attempt\":" + index + "}";
writeResponse(socket, body, false);
servedBodies.add("fresh");
freshResponses.incrementAndGet();
socket.close();
}
} catch (IOException ex) {
// ignore for the purpose of the test
}
}
private void forceReset(Socket socket) {
try {
if (!socket.isClosed()) {
servedBodies.add("reset");
closeWithReset(socket);
}
} finally {
warmupReset.countDown();
}
}
private void closeWithReset(Socket socket) {
try {
if (!socket.isClosed()) {
socket.setSoLinger(true, 0);
socket.close();
}
} catch (IOException ignored) {
// ignore
}
}
private RequestMetadata readRequest(Socket socket) throws IOException {
BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII));
String requestLine = reader.readLine();
if (requestLine == null) {
return new RequestMetadata("unknown", 0);
}
String path = requestLine.split(" ", 3)[1];
int contentLength = 0;
String line;
while ((line = reader.readLine()) != null && !line.isEmpty()) {
if (line.toLowerCase(Locale.ROOT).startsWith("content-length:")) {
contentLength = Integer.parseInt(line.substring(line.indexOf(':') + 1).trim());
}
}
if (contentLength > 0) {
char[] buffer = new char[contentLength];
int read = 0;
while (read < contentLength) {
int r = reader.read(buffer, read, contentLength - read);
if (r < 0) {
break;
}
read += r;
}
}
return new RequestMetadata(path, contentLength);
}
private void writeResponse(Socket socket, String body, boolean keepAlive) throws IOException {
byte[] bodyBytes = body.getBytes(StandardCharsets.UTF_8);
StringBuilder builder = new StringBuilder()
.append("HTTP/1.1 200 OK\r\n")
.append("Content-Type: application/json\r\n")
.append("Content-Length: ").append(bodyBytes.length).append("\r\n");
if (keepAlive) {
builder.append("Connection: keep-alive\r\n");
} else {
builder.append("Connection: close\r\n");
}
builder.append("\r\n");
OutputStream outputStream = socket.getOutputStream();
outputStream.write(builder.toString().getBytes(StandardCharsets.US_ASCII));
outputStream.write(bodyBytes);
outputStream.flush();
}
@Override
public void close() throws Exception {
running = false;
try {
serverSocket.close();
} catch (IOException ignored) {
}
if (warmupSocket != null && !warmupSocket.isClosed()) {
try {
warmupSocket.close();
} catch (IOException ignored) {
}
}
scheduler.shutdownNow();
acceptExecutor.shutdownNow();
}
private record RequestMetadata(String path, int contentLength) {
}
}
}

4
zt-module-databus/zt-module-databus-server/src/test/java/com/zt/plat/module/databus/service/gateway/ApiDefinitionServiceImplTest.java
View File

@@ -16,6 +16,7 @@ import com.zt.plat.module.databus.dal.mysql.gateway.ApiPolicyRateLimitMapper;
import com.zt.plat.module.databus.dal.mysql.gateway.ApiStepMapper;
import com.zt.plat.module.databus.dal.mysql.gateway.ApiTransformMapper;
import com.zt.plat.module.databus.enums.gateway.ApiStatusEnum;
import com.zt.plat.module.databus.service.gateway.ApiVersionService;
import com.zt.plat.module.databus.service.gateway.impl.ApiDefinitionServiceImpl;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.AfterEach;
@@ -63,6 +64,9 @@ class ApiDefinitionServiceImplTest extends BaseDbUnitTest {
@MockBean
private StringRedisTemplate stringRedisTemplate;
@MockBean
private ApiVersionService apiVersionService;
@TestConfiguration
static class JacksonTestConfiguration {

11
zt-module-infra/zt-module-infra-api/src/main/java/com/zt/plat/module/infra/api/businessfile/BusinessFileApi.java
View File

@@ -6,7 +6,7 @@ import com.zt.plat.module.infra.api.businessfile.dto.BusinessFilePageReqDTO;
import com.zt.plat.module.infra.api.businessfile.dto.BusinessFileRespDTO;
import com.zt.plat.module.infra.api.businessfile.dto.BusinessFileSaveReqDTO;
import com.zt.plat.module.infra.api.businessfile.dto.BusinessFileWithUrlRespDTO;
import com.zt.plat.module.infra.enums.ApiConstants;
import com.zt.plat.framework.common.enums.RpcConstants;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
@@ -19,11 +19,11 @@ import java.util.List;
/**
* @author chenbowen
*/
@FeignClient(name = ApiConstants.NAME)
@FeignClient(name = RpcConstants.INFRA_NAME)
@Tag(name = "RPC 服务 - 业务附件关联")
public interface BusinessFileApi {
String PREFIX = ApiConstants.PREFIX + "/business-file";
String PREFIX = RpcConstants.INFRA_PREFIX + "/business-file";
@PostMapping(PREFIX + "/create")
@Operation(summary = "创建业务附件关联")
@@ -52,6 +52,11 @@ public interface BusinessFileApi {
@Parameter(name = "id", description = "编号", required = true)
CommonResult<BusinessFileRespDTO> getBusinessFile(@RequestParam("id") Long id);
@GetMapping(PREFIX + "/get-by-code")
@Operation(summary = "根据业务编码获得业务附件关联")
@Parameter(name = "businessCode", description = "业务编码", required = true)
CommonResult<BusinessFileRespDTO> getBusinessFileByBusinessCode(@RequestParam("businessCode") String businessCode);
@PostMapping(PREFIX + "/page")
@Operation(summary = "获得业务附件关联分页")
CommonResult<PageResult<BusinessFileRespDTO>> getBusinessFilePage(@RequestBody BusinessFilePageReqDTO pageReqDTO);

6
zt-module-infra/zt-module-infra-server/src/main/java/com/zt/plat/module/infra/api/businessfile/BusinessFileApiImpl.java
View File

@@ -67,6 +67,12 @@ public class BusinessFileApiImpl implements BusinessFileApi {
return success(BeanUtils.toBean(businessFile, BusinessFileRespDTO.class));
}
@Override
public CommonResult<BusinessFileRespDTO> getBusinessFileByBusinessCode(String businessCode) {
BusinessFileDO businessFile = businessFileService.getBusinessFileByBusinessCode(businessCode);
return success(BeanUtils.toBean(businessFile, BusinessFileRespDTO.class));
}
@Override
public CommonResult<PageResult<BusinessFileRespDTO>> getBusinessFilePage(BusinessFilePageReqDTO pageReqDTO) {
PageResult<BusinessFileDO> pageResult = businessFileService.getBusinessFilePage(BeanUtils.toBean(pageReqDTO, BusinessFilePageReqVO.class));

9
zt-module-infra/zt-module-infra-server/src/main/java/com/zt/plat/module/infra/controller/admin/businessfile/BusinessFileController.java
View File

@@ -90,6 +90,15 @@ public class BusinessFileController {
return success(BeanUtils.toBean(businessFile, BusinessFileRespVO.class));
}
@GetMapping("/get-by-code")
@Operation(summary = "根据业务编码获得业务附件关联")
@Parameter(name = "businessCode", description = "业务编码", required = true)
@PreAuthorize("@ss.hasPermission('infra:business-file:query')")
public CommonResult<BusinessFileRespVO> getBusinessFileByBusinessCode(@RequestParam("businessCode") String businessCode) {
BusinessFileDO businessFile = businessFileService.getBusinessFileByBusinessCode(businessCode);
return success(BeanUtils.toBean(businessFile, BusinessFileRespVO.class));
}
@GetMapping("/page")
@Operation(summary = "获得业务附件关联分页")
@PreAuthorize("@ss.hasAnyPermissions({'infra:business-file:query','supply:purchase-credit-granting-form-template:query-list','supply:purchase-amount-request-form-template:query-list'})")

4
zt-module-infra/zt-module-infra-server/src/main/java/com/zt/plat/module/infra/dal/mysql/businessfile/BusinessFileMapper.java
View File

@@ -27,4 +27,8 @@ public interface BusinessFileMapper extends BaseMapperX<BusinessFileDO> {
.orderByDesc(BusinessFileDO::getId));
}
default BusinessFileDO selectByBusinessCode(String businessCode) {
return selectFirstOne(BusinessFileDO::getBusinessCode, businessCode);
}
}

8
zt-module-infra/zt-module-infra-server/src/main/java/com/zt/plat/module/infra/service/businessfile/BusinessFileService.java
View File

@@ -53,6 +53,14 @@ public interface BusinessFileService {
*/
BusinessFileDO getBusinessFile(Long id);
/**
* 根据业务编码获得业务附件关联
*
* @param businessCode 业务编码
* @return 业务附件关联
*/
BusinessFileDO getBusinessFileByBusinessCode(String businessCode);
/**
* 获得业务附件关联分页
*

13
zt-module-infra/zt-module-infra-server/src/main/java/com/zt/plat/module/infra/service/businessfile/BusinessFileServiceImpl.java
View File

@@ -18,6 +18,7 @@ import com.zt.plat.module.system.api.user.AdminUserApi;
import com.zt.plat.module.system.api.user.dto.AdminUserRespDTO;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.validation.annotation.Validated;
import java.util.*;
@@ -99,6 +100,18 @@ public class BusinessFileServiceImpl implements BusinessFileService {
return businessFileMapper.selectById(id);
}
@Override
public BusinessFileDO getBusinessFileByBusinessCode(String businessCode) {
if (!StringUtils.hasText(businessCode)) {
throw exception(BUSINESS_FILE_NOT_EXISTS);
}
BusinessFileDO businessFile = businessFileMapper.selectByBusinessCode(businessCode.trim());
if (businessFile == null) {
throw exception(BUSINESS_FILE_NOT_EXISTS);
}
return businessFile;
}
@Override
public PageResult<BusinessFileDO> getBusinessFilePage(BusinessFilePageReqVO pageReqVO) {
return businessFileMapper.selectPage(pageReqVO);

6
zt-module-system/zt-module-system-server/src/main/java/com/zt/plat/module/system/controller/admin/integration/iwork/vo/IWorkFileCallbackReqVO.java
View File

@@ -12,9 +12,9 @@ public class IWorkFileCallbackReqVO {
@NotBlank(message = "文件 URL 不能为空")
private String fileUrl;
@Schema(description = "业务 ID", requiredMode = Schema.RequiredMode.REQUIRED, example = "123456")
@NotBlank(message = "业务 ID 不能为空")
private String businessId;
@Schema(description = "业务编码", requiredMode = Schema.RequiredMode.REQUIRED, example = "DJ-2025-0001")
@NotBlank(message = "业务编码不能为空")
private String businessCode;
@Schema(description = "文件名称，可选", example = "合同附件.pdf")
private String fileName;

31
zt-module-system/zt-module-system-server/src/main/java/com/zt/plat/module/system/service/integration/iwork/impl/IWorkIntegrationServiceImpl.java
View File

@@ -11,6 +11,7 @@ import com.zt.plat.framework.common.exception.ServiceException;
import com.zt.plat.framework.common.exception.util.ServiceExceptionUtil;
import com.zt.plat.framework.common.pojo.CommonResult;
import com.zt.plat.module.infra.api.businessfile.BusinessFileApi;
import com.zt.plat.module.infra.api.businessfile.dto.BusinessFileRespDTO;
import com.zt.plat.module.infra.api.businessfile.dto.BusinessFileSaveReqDTO;
import com.zt.plat.module.infra.api.file.FileApi;
import com.zt.plat.module.infra.api.file.dto.FileCreateReqDTO;
@@ -161,20 +162,16 @@ public class IWorkIntegrationServiceImpl implements IWorkIntegrationService {
throw new ServiceException(IWORK_CONFIGURATION_INVALID.getCode(), "回调请求不能为空");
}
String fileUrl = Optional.ofNullable(reqVO.getFileUrl()).map(String::trim).orElse("");
String businessIdStr = Optional.ofNullable(reqVO.getBusinessId()).map(String::trim).orElse("");
String businessCode = Optional.ofNullable(reqVO.getBusinessCode()).map(String::trim).orElse("");
if (!StringUtils.hasText(fileUrl)) {
throw new ServiceException(IWORK_CONFIGURATION_INVALID.getCode(), "文件 URL 不能为空");
}
if (!StringUtils.hasText(businessIdStr)) {
throw new ServiceException(IWORK_CONFIGURATION_INVALID.getCode(), "业务 ID 不能为空");
if (!StringUtils.hasText(businessCode)) {
throw new ServiceException(IWORK_CONFIGURATION_INVALID.getCode(), "业务编码不能为空");
}
Long businessId;
try {
businessId = Long.valueOf(businessIdStr);
} catch (NumberFormatException ex) {
throw new ServiceException(IWORK_CONFIGURATION_INVALID.getCode(), "业务 ID 必须是数字: " + businessIdStr);
}
BusinessFileRespDTO referenceBusinessFile = loadBusinessFileByBusinessCode(businessCode);
Long businessId = referenceBusinessFile.getBusinessId();
// 通过文件 API 创建文件
FileCreateReqDTO fileCreateReqDTO = new FileCreateReqDTO();
@@ -192,6 +189,7 @@ public class IWorkIntegrationServiceImpl implements IWorkIntegrationService {
BusinessFileSaveReqDTO businessReq = BusinessFileSaveReqDTO.builder()
.businessId(businessId)
.businessCode(referenceBusinessFile.getBusinessCode())
.fileId(fileId)
.fileName(fileResult.getData().getName())
.source("iwork")
@@ -204,6 +202,21 @@ public class IWorkIntegrationServiceImpl implements IWorkIntegrationService {
return bizResult.getData();
}
private BusinessFileRespDTO loadBusinessFileByBusinessCode(String businessCode) {
CommonResult<BusinessFileRespDTO> businessResult = businessFileApi.getBusinessFileByBusinessCode(businessCode);
if (businessResult == null || !businessResult.isSuccess() || businessResult.getData() == null) {
throw new ServiceException(IWORK_CONFIGURATION_INVALID.getCode(),
"根据业务编码获取业务附件关联失败: " + Optional.ofNullable(businessResult)
.map(CommonResult::getMsg)
.orElse("未知错误"));
}
BusinessFileRespDTO businessFile = businessResult.getData();
if (businessFile.getBusinessId() == null) {
throw new ServiceException(IWORK_CONFIGURATION_INVALID.getCode(), "业务编码未绑定业务 ID: " + businessCode);
}
return businessFile;
}
private byte[] downloadFileBytes(String fileUrl) {
OkHttpClient client = okHttpClient();
Request request = new Request.Builder().url(fileUrl).get().build();

2
zt-module-template/zt-module-template-server/src/main/java/com/zt/plat/module/template/TemplateServerApplication.java
View File

@@ -2,6 +2,7 @@ package com.zt.plat.module.template;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
/**
* 项目的启动类
@@ -9,6 +10,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
* @author 周迪
*/
@SpringBootApplication
@EnableScheduling
public class TemplateServerApplication {
public static void main(String[] args) {

126
zt-module-template/zt-module-template-server/src/main/java/com/zt/plat/module/template/dal/dataobject/databus/TemplateDatabusRequestLogDO.java Normal file
View File

@@ -0,0 +1,126 @@
package com.zt.plat.module.template.dal.dataobject.databus;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.zt.plat.framework.mybatis.core.dataobject.BaseDO;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.ToString;
/**
* Databus 请求与响应日志表。
*/
@TableName("template_databus_request_log")
@Data
@EqualsAndHashCode(callSuper = true)
@ToString(callSuper = true)
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class TemplateDatabusRequestLogDO extends BaseDO {
@TableId(type = IdType.ASSIGN_ID)
private Long id;
/**
* 请求唯一标识。
*/
@TableField("REQUEST_ID")
private String requestId;
/**
* 实际请求 URL。
*/
@TableField("TARGET_URL")
private String targetUrl;
/**
* HTTP 方法。
*/
@TableField("HTTP_METHOD")
private String httpMethod;
/**
* Query 参数 JSON。
*/
@TableField("QUERY_PARAMS")
private String queryParams;
/**
* 请求头 JSON。
*/
@TableField("REQUEST_HEADERS")
private String requestHeaders;
/**
* 原始请求体。
*/
@TableField("REQUEST_BODY")
private String requestBody;
/**
* 加密后的请求体。
*/
@TableField("ENCRYPTED_REQUEST_BODY")
private String encryptedRequestBody;
/**
* 签名。
*/
@TableField("SIGNATURE")
private String signature;
/**
* 随机串。
*/
@TableField("NONCE")
private String nonce;
/**
* 时间戳。
*/
@TableField("ZT_TIMESTAMP")
private String ztTimestamp;
/**
* HTTP 返回状态码。
*/
@TableField("RESPONSE_STATUS")
private Integer responseStatus;
/**
* 加密响应体。
*/
@TableField("ENCRYPTED_RESPONSE_BODY")
private String encryptedResponseBody;
/**
* 解密后的响应体。
*/
@TableField("RESPONSE_BODY")
private String responseBody;
/**
* 是否调用成功。
*/
@TableField("SUCCESS")
private Boolean success;
/**
* 错误信息。
*/
@TableField("ERROR_MESSAGE")
private String errorMessage;
/**
* 调用耗时（毫秒）。
*/
@TableField("DURATION_MS")
private Long durationMs;
}

12
zt-module-template/zt-module-template-server/src/main/java/com/zt/plat/module/template/dal/mysql/databus/TemplateDatabusRequestLogMapper.java Normal file
View File

@@ -0,0 +1,12 @@
package com.zt.plat.module.template.dal.mysql.databus;
import com.zt.plat.framework.mybatis.core.mapper.BaseMapperX;
import com.zt.plat.module.template.dal.dataobject.databus.TemplateDatabusRequestLogDO;
import org.apache.ibatis.annotations.Mapper;
/**
* Databus 请求日志 Mapper。
*/
@Mapper
public interface TemplateDatabusRequestLogMapper extends BaseMapperX<TemplateDatabusRequestLogDO> {
}

21
zt-module-template/zt-module-template-server/src/main/java/com/zt/plat/module/template/job/databus/TemplateDatabusScheduler.java Normal file
View File

@@ -0,0 +1,21 @@
package com.zt.plat.module.template.job.databus;
import com.zt.plat.module.template.service.databus.TemplateDatabusInvokeService;
import lombok.RequiredArgsConstructor;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* 基于 Spring 的 Databus 调度任务。
*/
@Component
@RequiredArgsConstructor
public class TemplateDatabusScheduler {
private final TemplateDatabusInvokeService invokeService;
@Scheduled(cron = "0 0/10 * * * ?")
public void execute() {
invokeService.invokeAndRecord();
}
}

304
zt-module-template/zt-module-template-server/src/main/java/com/zt/plat/module/template/service/databus/TemplateDatabusInvokeService.java Normal file
View File

@@ -0,0 +1,304 @@
package com.zt.plat.module.template.service.databus;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zt.plat.framework.common.util.security.CryptoSignatureUtils;
import com.zt.plat.framework.tenant.core.aop.TenantIgnore;
import com.zt.plat.module.template.dal.dataobject.databus.TemplateDatabusRequestLogDO;
import com.zt.plat.module.template.dal.mysql.databus.TemplateDatabusRequestLogMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
/**
* 调用 Databus 接口并记录日志。
*/
@Service
@Slf4j
@RequiredArgsConstructor
public class TemplateDatabusInvokeService {
private static final TypeReference<LinkedHashMap<String, Object>> MAP_TYPE = new TypeReference<>() {
};
private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
private static final String HEADER_ZT_APP_ID = "ZT-App-Id";
private static final String HEADER_ZT_TIMESTAMP = "ZT-Timestamp";
private static final String HEADER_ZT_NONCE = "ZT-Nonce";
private static final String HEADER_ZT_SIGNATURE = "ZT-Signature";
private static final String HEADER_ZT_AUTH_TOKEN = "ZT-Auth-Token";
private static final String HEADER_CONTENT_TYPE = "Content-Type";
private static final String TARGET_URL = "http://172.16.46.63:30081/admin-api/databus/api/portal/callback/v1";
private static final String APP_ID = "ztmy";
private static final String APP_SECRET = "zFre/nTRGi7LpoFjN7oQkKeOT09x1fWTyIswrc702QQ=";
private static final String AUTH_TOKEN = "a5d7cf609c0b47038ea405c660726ee9";
private static final String DEFAULT_HTTP_METHOD = "POST";
private static final String ENCRYPTION_TYPE = CryptoSignatureUtils.ENCRYPT_TYPE_AES;
private static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(5);
private static final Duration READ_TIMEOUT = Duration.ofSeconds(10);
private static final Map<String, String> BASE_QUERY_PARAMS = Map.of(
"businessCode", "11",
"fileId", "11"
);
private static final Map<String, String> EXTRA_HEADERS = Map.of();
private static final String DEFAULT_REQUEST_BODY = """
{
}
""";
private static final HttpClient HTTP_CLIENT = HttpClient.newBuilder()
.connectTimeout(CONNECT_TIMEOUT)
.build();
private final TemplateDatabusRequestLogMapper requestLogMapper;
private final ObjectMapper objectMapper;
@TenantIgnore
public void invokeAndRecord() {
TemplateDatabusRequestLogDO logDO = TemplateDatabusRequestLogDO.builder()
.requestId(generateRequestId())
.httpMethod(DEFAULT_HTTP_METHOD)
.targetUrl(TARGET_URL)
.success(Boolean.FALSE)
.build();
Instant start = Instant.now();
try {
Map<String, Object> queryParams = buildQueryParams();
Map<String, Object> bodyParams = buildBodyParams(logDO.getRequestId());
String requestBody = toJson(bodyParams);
String serializedQuery = toJson(queryParams);
logDO.setRequestBody(requestBody);
logDO.setQueryParams(serializedQuery);
String timestamp = Long.toString(System.currentTimeMillis());
logDO.setZtTimestamp(timestamp);
String nonce = generateNonce();
logDO.setNonce(nonce);
String signature = generateSignature(queryParams, bodyParams, timestamp);
logDO.setSignature(signature);
String encryptedBody = encryptPayload(requestBody);
logDO.setEncryptedRequestBody(encryptedBody);
URI uri = buildUri(TARGET_URL, queryParams);
logDO.setTargetUrl(uri.toString());
Map<String, String> headers = buildHeaders(nonce, timestamp, signature);
logDO.setRequestHeaders(toJson(headers));
HttpRequest request = buildHttpRequest(uri, headers, encryptedBody);
HttpResponse<String> response = HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
logDO.setResponseStatus(response.statusCode());
logDO.setEncryptedResponseBody(response.body());
logDO.setResponseBody(tryDecrypt(response.body()));
boolean success = response.statusCode() >= 200 && response.statusCode() < 300;
logDO.setSuccess(success);
if (!success) {
log.warn("Databus API 返回非 2xx 状态码: {}", response.statusCode());
}
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
logDO.setErrorMessage(truncate("Interrupted: " + ex.getMessage(), 1000));
log.warn("Databus 调度被中断: {}", ex.getMessage());
} catch (Exception ex) {
logDO.setErrorMessage(truncate(ex.getMessage(), 1000));
log.error("Databus 调度执行异常", ex);
} finally {
logDO.setDurationMs(Duration.between(start, Instant.now()).toMillis());
requestLogMapper.insert(logDO);
}
}
private Map<String, Object> buildQueryParams() {
Map<String, Object> params = new LinkedHashMap<>();
BASE_QUERY_PARAMS.forEach(params::put);
return params;
}
private Map<String, Object> buildBodyParams(String requestId) {
Map<String, Object> body = new LinkedHashMap<>(parseTemplateBody());
body.put("__requestId__", requestId);
body.put("datetime", DATETIME_FORMATTER.format(LocalDateTime.now()));
body.putIfAbsent("system", "TEMPLATE");
return body;
}
private Map<String, Object> parseTemplateBody() {
try {
return objectMapper.readValue(DEFAULT_REQUEST_BODY, MAP_TYPE);
} catch (JsonProcessingException ex) {
throw new IllegalStateException("内置 Databus 请求体 JSON 解析失败", ex);
}
}
private Map<String, String> buildHeaders(String nonce, String timestamp, String signature) {
Map<String, String> headers = new LinkedHashMap<>();
headers.put(HEADER_ZT_APP_ID, APP_ID);
headers.put(HEADER_ZT_TIMESTAMP, timestamp);
headers.put(HEADER_ZT_NONCE, nonce);
headers.put(HEADER_ZT_SIGNATURE, signature);
if (StringUtils.hasText(AUTH_TOKEN)) {
headers.put(HEADER_ZT_AUTH_TOKEN, AUTH_TOKEN);
}
headers.put(HEADER_CONTENT_TYPE, "application/json");
EXTRA_HEADERS.forEach((key, value) -> {
if (StringUtils.hasText(key) && value != null) {
headers.put(key, value);
}
});
return headers;
}
private HttpRequest buildHttpRequest(URI uri, Map<String, String> headers, String encryptedBody) {
HttpRequest.Builder builder = HttpRequest.newBuilder(uri)
.timeout(READ_TIMEOUT);
headers.forEach(builder::header);
HttpRequest.BodyPublisher publisher = HttpRequest.BodyPublishers.ofString(encryptedBody, StandardCharsets.UTF_8);
switch (DEFAULT_HTTP_METHOD) {
case "GET":
builder.GET();
break;
case "PUT":
builder.PUT(publisher);
break;
case "PATCH":
builder.method("PATCH", publisher);
break;
case "DELETE":
builder.method("DELETE", publisher);
break;
default:
builder.POST(publisher);
}
return builder.build();
}
private String encryptPayload(String plaintext) {
try {
return CryptoSignatureUtils.encrypt(plaintext, APP_SECRET, ENCRYPTION_TYPE);
} catch (Exception ex) {
throw new IllegalStateException("请求体加密失败", ex);
}
}
private String tryDecrypt(String cipherText) {
if (!StringUtils.hasText(cipherText)) {
return cipherText;
}
try {
return CryptoSignatureUtils.decrypt(cipherText, APP_SECRET, ENCRYPTION_TYPE);
} catch (Exception ex) {
return "<unable to decrypt> " + ex.getMessage();
}
}
private URI buildUri(String baseUrl, Map<String, Object> queryParams) {
if (queryParams.isEmpty()) {
return URI.create(baseUrl);
}
StringBuilder builder = new StringBuilder(baseUrl);
builder.append(baseUrl.contains("?") ? '&' : '?');
boolean first = true;
for (Map.Entry<String, Object> entry : queryParams.entrySet()) {
if (!first) {
builder.append('&');
}
first = false;
builder.append(encode(entry.getKey()))
.append('=')
.append(encode(Objects.toString(entry.getValue(), "")));
}
return URI.create(builder.toString());
}
private String generateSignature(Map<String, Object> queryParams, Map<String, Object> bodyParams, String timestamp) {
TreeMap<String, Object> sorted = new TreeMap<>();
queryParams.forEach((key, value) -> sorted.put(key, normalizeValue(value)));
bodyParams.forEach((key, value) -> sorted.put(key, normalizeValue(value)));
sorted.put(HEADER_ZT_APP_ID, APP_ID);
sorted.put(HEADER_ZT_TIMESTAMP, timestamp);
StringBuilder canonical = new StringBuilder();
sorted.forEach((key, value) -> {
if (value == null) {
return;
}
if (canonical.length() > 0) {
canonical.append('&');
}
canonical.append(key).append('=').append(value);
});
return md5Hex(canonical.toString());
}
private Object normalizeValue(Object value) {
if (value == null) {
return null;
}
if (value instanceof Map || value instanceof Iterable) {
try {
return objectMapper.writeValueAsString(value);
} catch (JsonProcessingException ex) {
return value.toString();
}
}
return value;
}
private String md5Hex(String source) {
try {
MessageDigest digest = MessageDigest.getInstance("MD5");
byte[] bytes = digest.digest(source.getBytes(StandardCharsets.UTF_8));
StringBuilder builder = new StringBuilder(bytes.length * 2);
for (byte aByte : bytes) {
String part = Integer.toHexString(aByte & 0xFF);
if (part.length() == 1) {
builder.append('0');
}
builder.append(part);
}
return builder.toString();
} catch (NoSuchAlgorithmException ex) {
throw new IllegalStateException("MD5 算法不可用", ex);
}
}
private String encode(String value) {
return URLEncoder.encode(value, StandardCharsets.UTF_8);
}
private String generateRequestId() {
return UUID.randomUUID().toString().replace("-", "");
}
private String generateNonce() {
return UUID.randomUUID().toString().replace("-", "");
}
private String toJson(Object value) {
try {
return objectMapper.writeValueAsString(value);
} catch (JsonProcessingException ex) {
return String.valueOf(value);
}
}
private String truncate(String value, int maxLength) {
if (!StringUtils.hasText(value) || value.length() <= maxLength) {
return value;
}
return value.substring(0, maxLength);
}
}

49
zt-module-template/zt-module-template-server/src/main/resources/sql/template_databus_request_log.sql Normal file
View File

@@ -0,0 +1,49 @@
CREATE TABLE template_databus_request_log (
ID BIGINT NOT NULL,
REQUEST_ID VARCHAR(64) NOT NULL,
TARGET_URL VARCHAR(512) NOT NULL,
HTTP_METHOD VARCHAR(16) NOT NULL,
QUERY_PARAMS TEXT,
REQUEST_HEADERS TEXT,
REQUEST_BODY TEXT,
ENCRYPTED_REQUEST_BODY TEXT,
SIGNATURE VARCHAR(128),
NONCE VARCHAR(64),
ZT_TIMESTAMP VARCHAR(32),
RESPONSE_STATUS INT,
ENCRYPTED_RESPONSE_BODY TEXT,
RESPONSE_BODY TEXT,
SUCCESS BIT DEFAULT '0' NOT NULL,
ERROR_MESSAGE TEXT,
DURATION_MS BIGINT,
CREATE_TIME DATETIME(6) DEFAULT CURRENT_TIMESTAMP NOT NULL,
UPDATE_TIME DATETIME(6) DEFAULT CURRENT_TIMESTAMP NOT NULL,
CREATOR VARCHAR(64),
UPDATER VARCHAR(64),
DELETED BIT DEFAULT '0' NOT NULL,
PRIMARY KEY (ID)
);
COMMENT ON TABLE template_databus_request_log IS 'Databus 请求调度日志表';
COMMENT ON COLUMN template_databus_request_log.ID IS '主键';
COMMENT ON COLUMN template_databus_request_log.REQUEST_ID IS '请求唯一标识';
COMMENT ON COLUMN template_databus_request_log.TARGET_URL IS '目标地址（含 Query）';
COMMENT ON COLUMN template_databus_request_log.HTTP_METHOD IS 'HTTP 方法';
COMMENT ON COLUMN template_databus_request_log.QUERY_PARAMS IS 'Query 参数 JSON';
COMMENT ON COLUMN template_databus_request_log.REQUEST_HEADERS IS '请求头 JSON';
COMMENT ON COLUMN template_databus_request_log.REQUEST_BODY IS '原始请求体';
COMMENT ON COLUMN template_databus_request_log.ENCRYPTED_REQUEST_BODY IS '加密请求体';
COMMENT ON COLUMN template_databus_request_log.SIGNATURE IS '签名';
COMMENT ON COLUMN template_databus_request_log.NONCE IS '随机串';
COMMENT ON COLUMN template_databus_request_log.ZT_TIMESTAMP IS '时间戳';
COMMENT ON COLUMN template_databus_request_log.RESPONSE_STATUS IS '响应状态码';
COMMENT ON COLUMN template_databus_request_log.ENCRYPTED_RESPONSE_BODY IS '加密响应体';
COMMENT ON COLUMN template_databus_request_log.RESPONSE_BODY IS '解密后的响应体';
COMMENT ON COLUMN template_databus_request_log.SUCCESS IS '是否成功';
COMMENT ON COLUMN template_databus_request_log.ERROR_MESSAGE IS '错误信息';
COMMENT ON COLUMN template_databus_request_log.DURATION_MS IS '耗时（毫秒）';
COMMENT ON COLUMN template_databus_request_log.CREATE_TIME IS '创建时间';
COMMENT ON COLUMN template_databus_request_log.UPDATE_TIME IS '更新时间';
COMMENT ON COLUMN template_databus_request_log.CREATOR IS '创建人';
COMMENT ON COLUMN template_databus_request_log.UPDATER IS '更新人';
COMMENT ON COLUMN template_databus_request_log.DELETED IS '逻辑删除标记';