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 org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import jakarta.annotation.PostConstruct; import jakarta.annotation.Resource; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; @Component public class HttpLastlegEventClient implements LastlegEventClient { @Resource private ObjectMapper objectMapper; @Resource private EventDispatcherProperties properties; @Value("${baiyue.security.internal-service-token:}") private String serviceToken; private HttpClient httpClient; @PostConstruct void initializeHttpClient() { this.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()); String body = objectMapper.writeValueAsString(envelope); HttpRequest request = HttpRequest.newBuilder(properties.consumerUrl()) .timeout(properties.requestTimeout()) .header("Content-Type", "application/json") .header("X-Internal-Service-Token", serviceToken) .POST(HttpRequest.BodyPublishers.ofString(body)) .build(); HttpResponse 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()); } }