java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev' of http://8.163.30.128:10082/java-project/baiyue-firstleg into dev

Browse Source 
# Conflicts:
#	config/nacos/cn/event-bridge-service.yml
#	config/nacos/overseas/lastleg-service.yml
#	deploy/sql/头程数据库统一增量-MySQL8.sql
#	event-bridge-service/src/main/resources/application.yml
#	iam/src/main/resources/db/lastleg-permissions.sql
#	lastleg-service/AGENTS.md
#	lastleg-service/src/main/java/com/baiyue/lastleg/BaiyueLastlegApplication.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/controller/InboundPlanController.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/inventory/command/controller/InventoryCommandController.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/inventory/query/controller/InventoryQueryController.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/procurement/controller/PlatformOverseasProcurementController.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/procurement/service/OverseasProcurementService.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/relabel/controller/RelabelApplicationController.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/trucking/controller/PlatformUsTruckOrderController.java
#	lastleg-service/src/main/java/com/baiyue/lastleg/trucking/service/UsTruckOrderService.java
#	lastleg-service/src/main/resources/application.yml
#	lastleg-service/src/main/resources/db/lastleg.sql
#	scripts/db/incremental.sql
#	warehouse-overseas-service/src/main/java/com/baiyue/warehouse/controller/PlatformWarehouseController.java
#	warehouse-overseas-service/src/main/java/com/baiyue/warehouse/controller/PlatformWarehouseResourceController.java
This commit is contained in:
ouyangxq
2026-08-20 18:14:55 +08:00
parent 1cd8b6a473 c9a05b108b
commit 382ffc7f55
27 changed files with 415 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
event-bridge-service/src/main/java/com/baiyue/eventbridge/mapper/BridgeInboxMapper.java
+2 -3
View File
@@ -16,8 +16,7 @@ import java.util.List;
public interface BridgeInboxMapper extends BaseMapper<BridgeInbox> {
@Select("""
SELECT event_id FROM bridge_inbox
WHERE target_region = 'OVERSEAS'
AND ((status IN ('RECEIVED','RETRY') AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP(3)))
WHERE ((status IN ('RECEIVED','RETRY') AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP(3)))
OR (status = 'PROCESSING' AND lease_until < CURRENT_TIMESTAMP(3)))
ORDER BY received_at LIMIT #{limit}
""")
@@ -26,7 +25,7 @@ public interface BridgeInboxMapper extends BaseMapper<BridgeInbox> {
@Update("""
UPDATE bridge_inbox SET status='PROCESSING', lease_owner=#{owner},
lease_until=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{leaseSeconds} SECOND)
WHERE event_id=#{eventId} AND target_region='OVERSEAS'
WHERE event_id=#{eventId}
AND ((status IN ('RECEIVED','RETRY') AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP(3)))
OR (status='PROCESSING' AND lease_until < CURRENT_TIMESTAMP(3)))
""")
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/FirstlegEventClient.java
+7
View File
@@ -0,0 +1,7 @@
package com.baiyue.eventbridge.service;
import com.baiyue.eventbridge.domain.BridgeInbox;
public interface FirstlegEventClient {
String deliver(BridgeInbox event) throws Exception;
}
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/HttpFirstlegEventClient.java
+52
View File
@@ -0,0 +1,52 @@
package com.baiyue.eventbridge.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.RegionCode;
import com.baiyue.eventbridge.config.EventDispatcherProperties;
import com.baiyue.eventbridge.domain.BridgeInbox;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
@Component
public class HttpFirstlegEventClient implements FirstlegEventClient {
@Resource private ObjectMapper objectMapper;
@Resource private EventDispatcherProperties properties;
@Value("${baiyue.event-dispatcher.firstleg-consumer-url:http://127.0.0.1:9203/internal/firstleg/v1/container-events}")
private URI consumerUrl;
@Value("${baiyue.security.internal-service-token:}") private String serviceToken;
private HttpClient httpClient;
@PostConstruct
void initializeHttpClient() {
httpClient = HttpClient.newBuilder().connectTimeout(properties.connectTimeout()).build();
}
@Override
public String deliver(BridgeInbox row) throws Exception {
EventEnvelope envelope = new EventEnvelope(row.getEventId(), row.getEventType(), row.getEventVersion(),
row.getOccurredAt(), RegionCode.valueOf(row.getSourceRegion()), RegionCode.valueOf(row.getTargetRegion()),
row.getCustomerId(), row.getAggregateType(), row.getAggregateId(), row.getAggregateVersion(),
row.getHandoverVersion(), row.getTraceId(), row.getCorrelationId(), row.getCausationId(),
objectMapper.readTree(row.getPayloadJson()), row.getPayloadHash());
HttpRequest request = HttpRequest.newBuilder(consumerUrl).timeout(properties.requestTimeout())
.header("Content-Type", "application/json")
.header("X-Internal-Service-Token", serviceToken)
.POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(envelope))).build();
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
JsonNode responseBody = objectMapper.readTree(response.body());
if (response.statusCode() / 100 != 2 || responseBody == null
|| !"200".equals(responseBody.path("code").asText())) {
throw new IllegalStateException("头程消费失败: HTTP " + response.statusCode());
}
return String.valueOf(response.statusCode());
}
}
event-bridge-service/src/main/java/com/baiyue/eventbridge/service/OverseasEventDispatcher.java
+8 -2
View File
@@ -20,6 +20,7 @@ public class OverseasEventDispatcher {
@Resource private BridgeInboxMapper inboxMapper;
@Resource private BridgeDeliveryAttemptMapper attemptMapper;
@Resource private LastlegEventClient eventClient;
@Resource private FirstlegEventClient firstlegEventClient;
@Resource private EventDispatcherProperties properties;
private final String leaseOwner = ManagementFactory.getRuntimeMXBean().getName() + ":" + UUID.randomUUID();
@@ -40,7 +41,7 @@ public class OverseasEventDispatcher {
int deliveryAttemptNo = inboxMapper.nextAttemptNo(eventId);
BridgeDeliveryAttempt attempt = startAttempt(event, deliveryAttemptNo);
try {
String responseCode = eventClient.deliver(event);
String responseCode = deliver(event);
if (inboxMapper.markProcessed(eventId, leaseOwner) != 1) {
throw new IllegalStateException("事件租约已失效，无法确认投递结果");
}
@@ -49,11 +50,16 @@ public class OverseasEventDispatcher {
String error = abbreviate(exception.getMessage(), 1000);
inboxMapper.markFailed(eventId, leaseOwner, properties.maxAttempts(), backoffSeconds(retryAttempt), error);
finishAttempt(attempt, retryAttempt >= properties.maxAttempts() ? "DEAD" : "RETRY", null, error);
log.warn("尾程事件投递失败 eventId={}, retryAttempt={}, deliveryAttempt={}, reason={}",
log.warn("跨区事件投递失败 eventId={}, retryAttempt={}, deliveryAttempt={}, reason={}",
eventId, retryAttempt, deliveryAttemptNo, error);
}
}
private String deliver(BridgeInbox event) throws Exception {
return "CN".equals(event.getTargetRegion())
? firstlegEventClient.deliver(event) : eventClient.deliver(event);
}
private BridgeDeliveryAttempt startAttempt(BridgeInbox event, int attemptNo) {
BridgeDeliveryAttempt attempt = new BridgeDeliveryAttempt();
attempt.setEventId(event.getEventId());
event-bridge-service/src/main/resources/application.yml
+1
View File
@@ -32,6 +32,7 @@ baiyue:
region: ${REGION_CODE:OVERSEAS}
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}
fixed-delay: ${EVENT_DISPATCH_FIXED_DELAY:3000}
batch-size: ${EVENT_DISPATCH_BATCH_SIZE:50}
max-attempts: ${EVENT_DISPATCH_MAX_ATTEMPTS:10}
event-bridge-service/src/test/java/com/baiyue/eventbridge/service/OverseasEventDispatcherTest.java
+17
View File
@@ -21,6 +21,7 @@ class OverseasEventDispatcherTest {
private final BridgeInboxMapper inboxMapper = mock(BridgeInboxMapper.class);
private final BridgeDeliveryAttemptMapper attemptMapper = mock(BridgeDeliveryAttemptMapper.class);
private final LastlegEventClient eventClient = mock(LastlegEventClient.class);
private final FirstlegEventClient firstlegEventClient = mock(FirstlegEventClient.class);
private OverseasEventDispatcher dispatcher;
private BridgeInbox event;
@@ -34,6 +35,7 @@ class OverseasEventDispatcherTest {
ReflectionTestUtils.setField(dispatcher, "inboxMapper", inboxMapper);
ReflectionTestUtils.setField(dispatcher, "attemptMapper", attemptMapper);
ReflectionTestUtils.setField(dispatcher, "eventClient", eventClient);
ReflectionTestUtils.setField(dispatcher, "firstlegEventClient", firstlegEventClient);
ReflectionTestUtils.setField(dispatcher, "properties", properties);
event = new BridgeInbox();
event.setEventId("event-1");
@@ -83,9 +85,24 @@ class OverseasEventDispatcherTest {
dispatcher.dispatchOne("event-1");
verifyNoInteractions(eventClient);
verifyNoInteractions(firstlegEventClient);
verify(attemptMapper, never()).insert(any(BridgeDeliveryAttempt.class));
}
@Test
void cnTargetIsDeliveredToFirstleg() throws Exception {
event.setSourceRegion("OVERSEAS");
event.setTargetRegion("CN");
when(firstlegEventClient.deliver(event)).thenReturn("200");
when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(1);
dispatcher.dispatchOne("event-1");
verify(firstlegEventClient).deliver(event);
verifyNoInteractions(eventClient);
assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED");
}
private BridgeDeliveryAttempt updatedAttempt() {
ArgumentCaptor<BridgeDeliveryAttempt> captor = ArgumentCaptor.forClass(BridgeDeliveryAttempt.class);
verify(attemptMapper).updateById(captor.capture());