This commit is contained in:
2026-08-25 16:40:48 +08:00
parent 1f6f1a0677
commit 87c5e18ca6
93 changed files with 1154 additions and 527 deletions
+1 -1
View File
@@ -12,7 +12,7 @@
|---|---|
| 接收、状态、人工补投 API | `src/main/java/com/baiyue/eventbridge/controller/EventBridgeController.java` |
| Inbox 接收与幂等 | `service/EventInboxService.java` |
| 按目标区域调度与重试 | `service/OverseasEventDispatcher.java`;`targetRegion=CN` 回投 Firstleg;海外业务事件投 Lastleg,三类 IAM 安全事件按 eventType 投 `overseas-security-service`,传输层不解析 payload |
| 按目标区域调度与重试 | `service/OverseasEventDispatcher.java`;`targetRegion=CN` 回投 Firstleg;海外业务事件投 Lastleg 通用入口,三类 IAM 安全事件按 eventType 投 Lastleg `/internal/lastleg/security/v1/events`,传输层不解析 payload |
| 尾程 HTTP 客户端 | `service/LastlegEventClient.java`、`HttpLastlegEventClient.java` |
| 国内头程 HTTP 客户端 | `service/FirstlegEventClient.java`、`HttpFirstlegEventClient.java`;默认内部入口 `/internal/firstleg/v1/container-events` |
| 人工补投 | `service/EventReplayService.java`;只允许 `DEAD` 转回 `RETRY`,记录操作人与原因,并保留原 `eventId`、payload 与 `payloadHash` |
@@ -6,6 +6,7 @@ import com.baiyue.eventbridge.domain.BridgeInbox;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.Clock;
@@ -41,6 +42,7 @@ public class EventBridgeMetrics {
private final MeterRegistry registry;
private final Clock clock;
@Autowired
public EventBridgeMetrics(MeterRegistry registry) {
this(registry, Clock.systemDefaultZone());
}
@@ -47,10 +47,14 @@ public class HttpSecurityProjectionEventClient implements SecurityProjectionEven
JsonNode body;
try { body = objectMapper.readTree(response.body()); }
catch (Exception exception) { body = null; }
if (response.statusCode() / 100 != 2 || body == null || !"200".equals(body.path("code").asText())
|| !row.getEventId().equals(body.path("data").path("eventId").asText())) {
throw new IllegalStateException("海外安全投影消费失败: HTTP " + response.statusCode());
}
requireAcknowledgement(response.statusCode(), body, row.getEventId());
return String.valueOf(response.statusCode());
}
static void requireAcknowledgement(int statusCode, JsonNode body, String eventId) {
if (statusCode / 100 != 2 || body == null || !"200".equals(body.path("code").asText())
|| !eventId.equals(body.path("data").path("eventId").asText())) {
throw new IllegalStateException("Lastleg 安全投影消费失败: HTTP " + statusCode);
}
}
}
@@ -33,7 +33,7 @@ baiyue:
event-dispatcher:
consumer-url: ${EVENT_CONSUMER_URL:${LASTLEG_CONSUMER_URL:http://127.0.0.1:9303/internal/lastleg/events}}
firstleg-consumer-url: ${FIRSTLEG_CONTAINER_EVENT_CONSUMER_URL:http://127.0.0.1:9202/internal/firstleg/v1/container-events}
security-consumer-url: ${OVERSEAS_SECURITY_EVENT_CONSUMER_URL:http://127.0.0.1:9307/internal/overseas-security/v1/events}
security-consumer-url: ${LASTLEG_SECURITY_EVENT_CONSUMER_URL:http://127.0.0.1:9303/internal/lastleg/security/v1/events}
fixed-delay: ${EVENT_DISPATCH_FIXED_DELAY:3000}
batch-size: ${EVENT_DISPATCH_BATCH_SIZE:50}
max-attempts: ${EVENT_DISPATCH_MAX_ATTEMPTS:10}
@@ -8,12 +8,32 @@ import org.junit.jupiter.api.Test;
import java.lang.reflect.Modifier;
import java.util.Arrays;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.mockito.Mockito.mock;
class EventBridgeMetricsTest {
@Test
void productionConstructorIsExplicitForSpringWhenTestClockConstructorAlsoExists() throws Exception {
assertThat(EventBridgeMetrics.class.getConstructor(MeterRegistry.class)
.isAnnotationPresent(Autowired.class)).isTrue();
}
@Test
void springContextCreatesMetricsBeanWithTheAnnotatedProductionConstructor() {
try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) {
context.registerBean(MeterRegistry.class, SimpleMeterRegistry::new);
context.register(EventBridgeMetrics.class);
context.refresh();
assertThat(context.getBean(EventBridgeMetrics.class)).isNotNull();
}
}
@Test
void collapsesUnboundedRegionAndEventTypeLabelsToOther() {
SimpleMeterRegistry registry = new SimpleMeterRegistry();
@@ -0,0 +1,44 @@
package com.baiyue.eventbridge.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import java.nio.file.Files;
import java.nio.file.Path;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class HttpSecurityProjectionEventClientTest {
private final ObjectMapper objectMapper = new ObjectMapper();
@Test
void onlyMatchingLastlegAcknowledgementCompletesBridgeDelivery() throws Exception {
var acknowledged = objectMapper.readTree("""
{"code":"200","data":{"eventId":"event-1","result":"APPLIED"}}
""");
var businessFailure = objectMapper.readTree("""
{"code":"A0409","message":"payloadHash 冲突","data":{"eventId":"event-1"}}
""");
var wrongEvent = objectMapper.readTree("""
{"code":"200","data":{"eventId":"event-2","result":"DUPLICATE"}}
""");
assertThatCode(() -> HttpSecurityProjectionEventClient.requireAcknowledgement(
200, acknowledged, "event-1")).doesNotThrowAnyException();
assertThatThrownBy(() -> HttpSecurityProjectionEventClient.requireAcknowledgement(
200, businessFailure, "event-1")).isInstanceOf(IllegalStateException.class);
assertThatThrownBy(() -> HttpSecurityProjectionEventClient.requireAcknowledgement(
200, wrongEvent, "event-1")).isInstanceOf(IllegalStateException.class);
}
@Test
void defaultConsumerUrlTargetsLastlegSecurityEndpoint() throws Exception {
String application = Files.readString(Path.of("src/main/resources/application.yml"));
assertThat(application).contains(
"LASTLEG_SECURITY_EVENT_CONSUMER_URL:http://127.0.0.1:9303/internal/lastleg/security/v1/events")
.doesNotContain("/internal/overseas-security/v1/events");
}
}