1. 新增分页接口聚合查询注解支持
2. 优化 databus api 日志记录的字段缺失问题 3. 新增 eplat sso 页面登录校验 4. 用户、部门编辑新增 seata 事务支持 5. 新增 iwork 流程发起接口 6. 新增 eban 同步用户时的岗位处理逻辑 7. 新增无 skywalking 时的 traceId 支持
This commit is contained in:
chenbowen
@@ -0,0 +1,360 @@
|
||||
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) {
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user