shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: reuse mqtt identity resolution

Browse Source
This commit is contained in:
lingniu
2026-07-01 10:12:52 +08:00
parent 418aefc189
commit 74e01acc7a
4 changed files with 126 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java
View File

@@ -141,6 +141,7 @@ public final class MqttEndpointManager implements AutoCloseable {
MqttPayload parsed = profileRegistry.parse(ep.getName(), ep.getProfile(), topic, payload);
String externalDeviceId = firstNonBlank(parsed.deviceId(), parsed.vin());
IdentityResolution identity = resolveIdentity(parsed);
parsed = parsed.withIdentity(identity.identity(), identity.errorMessage());
Map<String, String> meta = new HashMap<>(4);
// metadata 同时保留外部设备号和内部 VIN，便于排查绑定表错误导致的车辆归属问题。
meta.put("vin", identity.identity().vin());

11
modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/mapper/YutongEventMapper.java
View File

@@ -109,6 +109,17 @@ public final class YutongEventMapper implements EventMapper<MqttPayload> {
}
private IdentityResolution resolveIdentity(MqttPayload payload) {
if (payload.hasIdentitySnapshot()) {
VehicleIdentitySource source = VehicleIdentitySource.UNKNOWN;
try {
source = VehicleIdentitySource.valueOf(payload.identitySource());
} catch (IllegalArgumentException ignored) {
// 保守兜底，外部化的 payload 如果带了未知 source 不应影响接收主链路。
}
return new IdentityResolution(
new VehicleIdentity(payload.resolvedVin(), payload.identityResolved(), source),
payload.identityErrorMessage().isBlank() ? null : payload.identityErrorMessage());
}
String externalDeviceId = firstNonBlank(payload.deviceId(), payload.vin());
String externalVin = payload.externalVin() == null ? "" : payload.externalVin();
String phone = payload.phone() == null ? "" : payload.phone();

41
modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/model/MqttPayload.java
View File

@@ -1,6 +1,9 @@
package com.lingniu.ingest.inbound.mqtt.model;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.databind.JsonNode;
import com.lingniu.ingest.identity.VehicleIdentity;
import com.lingniu.ingest.identity.VehicleIdentitySource;
import java.time.Instant;
@@ -15,6 +18,7 @@ import java.time.Instant;
* @param phone 外部手机号 / SIM 字段
* @param deviceId 外部设备号 / 终端 ID / IMEI 字段
* @param plateNo 外部车牌字段
* @param resolvedVin 已解析的内部 VIN，供 RAW 和业务事件复用同一次身份查询
* @param deviceTime 设备上报时刻
* @param data 原始 JSON 树，供 Mapper 按 profile 提取字段
*/
@@ -27,9 +31,33 @@ public record MqttPayload(
String phone,
String deviceId,
String plateNo,
@JsonIgnore String resolvedVin,
@JsonIgnore boolean identityResolved,
@JsonIgnore String identitySource,
@JsonIgnore String identityErrorMessage,
Instant deviceTime,
JsonNode data
) {
public MqttPayload {
resolvedVin = resolvedVin == null ? "" : resolvedVin.trim();
identitySource = identitySource == null ? "" : identitySource.trim();
identityErrorMessage = identityErrorMessage == null ? "" : identityErrorMessage.trim();
}
public MqttPayload(String endpointName,
String profile,
String topic,
String vin,
String externalVin,
String phone,
String deviceId,
String plateNo,
Instant deviceTime,
JsonNode data) {
this(endpointName, profile, topic, vin, externalVin, phone, deviceId, plateNo,
"", false, "", "", deviceTime, data);
}
public MqttPayload(String endpointName,
String profile,
String topic,
@@ -38,4 +66,17 @@ public record MqttPayload(
JsonNode data) {
this(endpointName, profile, topic, vin, "", "", vin, "", deviceTime, data);
}
public MqttPayload withIdentity(VehicleIdentity identity, String errorMessage) {
VehicleIdentity safe = identity == null
? new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN)
: identity;
return new MqttPayload(endpointName, profile, topic, vin, externalVin, phone, deviceId, plateNo,
safe.vin(), safe.resolved(), safe.source().name(), errorMessage, deviceTime, data);
}
@JsonIgnore
public boolean hasIdentitySnapshot() {
return !identitySource.isBlank();
}
}

73
modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java
View File

@@ -12,7 +12,11 @@ import com.lingniu.ingest.core.dispatcher.HandlerInvoker;
import com.lingniu.ingest.core.dispatcher.HandlerRegistry;
import com.lingniu.ingest.core.pipeline.InterceptorChain;
import com.lingniu.ingest.identity.InMemoryVehicleIdentityService;
import com.lingniu.ingest.identity.VehicleIdentity;
import com.lingniu.ingest.identity.VehicleIdentityBinding;
import com.lingniu.ingest.identity.VehicleIdentityLookup;
import com.lingniu.ingest.identity.VehicleIdentityResolver;
import com.lingniu.ingest.identity.VehicleIdentitySource;
import com.lingniu.ingest.inbound.mqtt.client.MqttEndpointManager;
import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties;
import com.lingniu.ingest.inbound.mqtt.handler.MqttRealtimeHandler;
@@ -30,6 +34,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -86,6 +91,55 @@ class MqttEndpointManagerTest {
eventBus.close();
}
@Test
void inboundMessageReusesResolvedIdentityForRawArchiveAndMappedEvents() throws Exception {
MqttInboundProperties.Endpoint endpoint = new MqttInboundProperties.Endpoint();
endpoint.setName("endpoint-a");
endpoint.setTopic("/ytforward/#");
endpoint.setProfile("yutong");
byte[] payload = """
{"device":"YT-DEVICE-001","time":"20260622150000","data":{"METER_SPEED":31,"TOTAL_MILEAGE":1024}}
""".trim().getBytes(StandardCharsets.UTF_8);
CountingIdentityResolver identities = new CountingIdentityResolver(new VehicleIdentity(
"LNVIN000000MQTT01", true, VehicleIdentitySource.BOUND_DEVICE_ID));
RecordingSink sink = new RecordingSink(2);
DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink));
AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
MqttProfileRegistry profileRegistry = new MqttProfileRegistry(List.of(
new YutongMqttProfile(
new MqttPayloadParser(new ObjectMapper()),
new YutongEventMapper(identities))));
Dispatcher dispatcher = new Dispatcher(
registry(profileRegistry),
new InterceptorChain(List.of()),
new HandlerInvoker(),
eventBus,
batchExecutor);
MqttEndpointManager manager = new MqttEndpointManager(
new MqttInboundProperties(), profileRegistry, identities, dispatcher);
invokeOnMessage(manager, endpoint, "/ytforward/vehicle/YT-DEVICE-001", payload);
assertThat(sink.await()).isTrue();
assertThat(identities.calls()).isEqualTo(1);
assertThat(sink.events)
.extracting(VehicleEvent::vin)
.contains("LNVIN000000MQTT01");
assertThat(sink.events).anySatisfy(event -> {
assertThat(event).isInstanceOf(VehicleEvent.RawArchive.class);
VehicleEvent.RawArchive archive = (VehicleEvent.RawArchive) event;
assertThat(archive.parsedJson())
.contains("\"deviceId\":\"YT-DEVICE-001\"")
.doesNotContain("resolvedVin")
.doesNotContain("identityResolved")
.doesNotContain("identitySource")
.doesNotContain("identityErrorMessage");
});
batchExecutor.close();
eventBus.close();
}
@Test
void malformedInboundMessageArchivesParseFailureMetadataAndPassthrough() throws Exception {
MqttInboundProperties.Endpoint endpoint = new MqttInboundProperties.Endpoint();
@@ -381,6 +435,25 @@ class MqttEndpointManagerTest {
}
}
private static final class CountingIdentityResolver implements VehicleIdentityResolver {
private final VehicleIdentity identity;
private final AtomicInteger calls = new AtomicInteger();
private CountingIdentityResolver(VehicleIdentity identity) {
this.identity = identity;
}
@Override
public VehicleIdentity resolve(VehicleIdentityLookup lookup) {
calls.incrementAndGet();
return identity;
}
private int calls() {
return calls.get();
}
}
private static final class BrokenProfile implements com.lingniu.ingest.inbound.mqtt.profile.MqttProfile {
@Override
public String name() {