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

feat: split history and analytics consumers

Browse Source
This commit is contained in:
lingniu
2026-06-23 17:43:28 +08:00
parent f9aa214dc2
commit 6a8b53bab6
6 changed files with 143 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java
View File

@@ -60,8 +60,14 @@ public final class TelemetryEnvelopeRecordMapper {
ProtocolId protocol = protocol(envelope.getSource());
RawArchiveRef rawArchive = envelope.getRawArchive();
Map<String, String> metadata = metadata(envelope, protocol);
String rawArchiveUri = rawArchive.getUri();
String rawArchiveKey = rawArchiveKey(rawArchive.getUri(), metadata);
String rawArchiveUri = rawArchive.getUri().isBlank() && !rawArchiveKey.isBlank()
? RawArchiveKeys.logicalUri(rawArchiveKey)
: rawArchive.getUri();
metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, envelope.getEventId());
if (!rawArchiveKey.isBlank()) {
metadata.putIfAbsent(RawArchiveKeys.META_KEY, rawArchiveKey);
}
if (!rawArchiveUri.isBlank()) {
metadata.putIfAbsent(RawArchiveKeys.META_URI, rawArchiveUri);
}
@@ -74,10 +80,21 @@ public final class TelemetryEnvelopeRecordMapper {
Instant.ofEpochMilli(envelope.getIngestTimeMs()),
rawArchiveUri,
metadata,
rawArchiveJson(envelope, metadata)
rawArchiveJson(envelope, metadata, rawArchiveKey, rawArchiveUri)
);
}
private static String rawArchiveKey(String rawArchiveUri, Map<String, String> metadata) {
String key = metadata.getOrDefault(RawArchiveKeys.META_KEY, "");
if (!key.isBlank()) {
return key;
}
if (rawArchiveUri != null && rawArchiveUri.startsWith("archive://")) {
return rawArchiveUri.substring("archive://".length());
}
return "";
}
private static Map<String, String> metadata(VehicleEnvelope envelope, ProtocolId protocol) {
Map<String, String> metadata = new LinkedHashMap<>(envelope.getMetadataMap());
if (!envelope.getProtocolVersion().isBlank()) {
@@ -109,7 +126,10 @@ public final class TelemetryEnvelopeRecordMapper {
}
}
private static String rawArchiveJson(VehicleEnvelope envelope, Map<String, String> metadata) {
private static String rawArchiveJson(VehicleEnvelope envelope,
Map<String, String> metadata,
String rawArchiveKey,
String rawArchiveUri) {
Map<String, Object> payload = new LinkedHashMap<>();
RawArchiveRef rawArchive = envelope.getRawArchive();
payload.put("eventId", envelope.getEventId());
@@ -121,7 +141,14 @@ public final class TelemetryEnvelopeRecordMapper {
payload.put("eventType", "RAW_ARCHIVE");
payload.put("eventTime", Instant.ofEpochMilli(envelope.getEventTimeMs()).toString());
payload.put("ingestTime", Instant.ofEpochMilli(envelope.getIngestTimeMs()).toString());
payload.put("rawArchiveUri", rawArchive.getUri());
payload.put("rawArchiveUri", rawArchiveUri);
payload.put("rawArchiveKey", rawArchiveKey);
if (metadata.containsKey("command")) {
payload.put("command", metadata.get("command"));
}
if (metadata.containsKey("infoType")) {
payload.put("infoType", metadata.get("infoType"));
}
payload.put("rawSizeBytes", rawArchive.getSizeBytes());
payload.put("metadata", metadata);
try {

39
modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java
View File

@@ -1,7 +1,11 @@
package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.eventfilestore.EventFileQuery;
import com.lingniu.ingest.eventfilestore.EventFileRecord;
import com.lingniu.ingest.eventfilestore.EventFileStore;
@@ -12,6 +16,7 @@ import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
@@ -20,6 +25,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
class EventHistoryEnvelopeIngestorTest {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
@Test
void parsesKafkaEnvelopeBytesAndAppendsRecord() throws Exception {
CapturingStore store = new CapturingStore();
@@ -57,10 +64,12 @@ class EventHistoryEnvelopeIngestorTest {
}
@Test
void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() {
void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() throws Exception {
CapturingStore store = new CapturingStore();
EventHistoryEnvelopeIngestor ingestor =
new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
String rawArchiveKey = "2026/06/23/GB32960/VINRAW001/raw-event-1.bin";
String rawArchiveUri = "archive://" + rawArchiveKey;
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
@@ -70,9 +79,11 @@ class EventHistoryEnvelopeIngestorTest {
.setProtocolVersion("V2016")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.putMetadata("archiveKey", "gb32960/2026/06/23/raw-event-1.bin")
.putMetadata(RawArchiveKeys.META_KEY, rawArchiveKey)
.putMetadata(RawArchiveKeys.META_URI, rawArchiveUri)
.putMetadata("command", "0x0002")
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://gb32960/2026/06/23/raw-event-1.bin")
.setUri(rawArchiveUri)
.setSizeBytes(128)
.build())
.build();
@@ -83,14 +94,28 @@ class EventHistoryEnvelopeIngestorTest {
assertThat(store.records).hasSize(1);
EventFileRecord record = store.records.getFirst();
assertThat(record.eventId()).isEqualTo("raw-event-1");
assertThat(record.protocol()).isEqualTo(ProtocolId.GB32960);
assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
assertThat(record.vin()).isEqualTo("VINRAW001");
assertThat(record.rawArchiveUri()).isEqualTo("archive://gb32960/2026/06/23/raw-event-1.bin");
assertThat(record.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_112_400_000L));
assertThat(record.rawArchiveUri()).isEqualTo(rawArchiveUri);
assertThat(record.metadata())
.containsEntry("protocolVersion", "V2016")
.containsEntry("archiveKey", "gb32960/2026/06/23/raw-event-1.bin");
assertThat(record.payloadJson()).contains("\"rawSizeBytes\":128");
assertThat(record.payloadJson()).doesNotContain("rawBytes");
.containsEntry(RawArchiveKeys.META_EVENT_ID, "raw-event-1")
.containsEntry(RawArchiveKeys.META_KEY, rawArchiveKey)
.containsEntry(RawArchiveKeys.META_URI, rawArchiveUri)
.containsEntry("command", "0x0002");
JsonNode payload = OBJECT_MAPPER.readTree(record.payloadJson());
assertThat(payload.get("eventId").asText()).isEqualTo("raw-event-1");
assertThat(payload.get("protocol").asText()).isEqualTo("GB32960");
assertThat(payload.get("eventType").asText()).isEqualTo("RAW_ARCHIVE");
assertThat(payload.get("vin").asText()).isEqualTo("VINRAW001");
assertThat(payload.get("eventTime").asText()).isEqualTo("2026-06-22T07:13:20Z");
assertThat(payload.get("rawArchiveUri").asText()).isEqualTo(rawArchiveUri);
assertThat(payload.get("rawArchiveKey").asText()).isEqualTo(rawArchiveKey);
assertThat(payload.get("rawSizeBytes").asLong()).isEqualTo(128);
assertThat(payload.get("metadata").get("command").asText()).isEqualTo("0x0002");
assertThat(payload.toString()).doesNotContain("rawBytes");
}
private static VehicleEnvelope envelope(String eventId) {

6
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java
View File

@@ -17,7 +17,11 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor {
}
public void ingest(byte[] kafkaValue) {
processor.process(parse(kafkaValue));
VehicleEnvelope envelope = parse(kafkaValue);
if (!envelope.hasTelemetrySnapshot()) {
return;
}
processor.process(envelope);
}
@Override

32
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java
View File

@@ -2,6 +2,7 @@ package com.lingniu.ingest.vehiclestat;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
@@ -74,6 +75,23 @@ class VehicleStatEnvelopeIngestorTest {
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
}
@Test
void rawArchiveEnvelopeIsIgnoredWithoutAppendingMileagePoint() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
VehicleStatEnvelopeIngestor ingestor =
new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(repository));
VehicleEnvelope envelope = rawArchiveEnvelope();
ingestor.ingest(envelope.toByteArray());
var result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED);
assertThat(result.eventId()).isEqualTo("raw-event-1");
assertThat(result.vin()).isEqualTo("VIN001");
assertThat(result.message()).contains("telemetry_snapshot");
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
}
private static VehicleEnvelope envelope() {
return VehicleEnvelope.newBuilder()
.setEventId("event-1")
@@ -91,6 +109,20 @@ class VehicleStatEnvelopeIngestorTest {
.build();
}
private static VehicleEnvelope rawArchiveEnvelope() {
return VehicleEnvelope.newBuilder()
.setEventId("raw-event-1")
.setVin("VIN001")
.setSource("GB32960")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://2026/06/23/GB32960/VIN001/raw-event-1.bin")
.setSizeBytes(128)
.build())
.build();
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
private final java.util.Map<String, java.util.List<MileagePoint>> points = new java.util.HashMap<>();

10
modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java
View File

@@ -17,7 +17,11 @@ public final class VehicleStateEnvelopeIngestor implements EnvelopeIngestor {
}
public void ingest(byte[] kafkaValue) {
updater.update(parse(kafkaValue));
VehicleEnvelope envelope = parse(kafkaValue);
if (!envelope.hasTelemetrySnapshot()) {
return;
}
updater.update(envelope);
}
@Override
@@ -25,6 +29,10 @@ public final class VehicleStateEnvelopeIngestor implements EnvelopeIngestor {
VehicleEnvelope envelope = null;
try {
envelope = parse(kafkaValue);
if (!envelope.hasTelemetrySnapshot()) {
return EnvelopeIngestResult.skipped(
envelope.getEventId(), envelope.getVin(), "envelope telemetry_snapshot is required");
}
// Kafka 消费路径只接受标准 VehicleEnvelope；坏消息返回明确结果给处理器写 DLQ。
updater.update(envelope);
return EnvelopeIngestResult.processed(envelope.getEventId(), envelope.getVin());

34
modules/services/vehicle-state-service/src/test/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestorTest.java
View File

@@ -2,6 +2,7 @@ package com.lingniu.ingest.vehiclestate;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
@@ -47,6 +48,25 @@ class VehicleStateEnvelopeIngestorTest {
assertThat(repository.getState("VIN001")).isEmpty();
}
@Test
void rawArchiveEnvelopeIsIgnoredWithoutUpdatingRepository() {
InMemoryVehicleStateRepository repository = new InMemoryVehicleStateRepository();
VehicleStateEnvelopeIngestor ingestor = new VehicleStateEnvelopeIngestor(new VehicleStateUpdater(repository));
VehicleEnvelope envelope = rawArchiveEnvelope();
ingestor.ingest(envelope.toByteArray());
var result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED);
assertThat(result.eventId()).isEqualTo("raw-event-1");
assertThat(result.vin()).isEqualTo("VIN001");
assertThat(result.message()).contains("telemetry_snapshot");
assertThat(repository.getState("VIN001")).isEmpty();
assertThat(repository.getLocation("VIN001")).isEmpty();
assertThat(repository.getSafety("VIN001")).isEmpty();
assertThat(repository.getLastEvent("VIN001")).isEmpty();
}
private static VehicleEnvelope envelope() {
return VehicleEnvelope.newBuilder()
.setEventId("event-1")
@@ -64,6 +84,20 @@ class VehicleStateEnvelopeIngestorTest {
.build();
}
private static VehicleEnvelope rawArchiveEnvelope() {
return VehicleEnvelope.newBuilder()
.setEventId("raw-event-1")
.setVin("VIN001")
.setSource("GB32960")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://2026/06/23/GB32960/VIN001/raw-event-1.bin")
.setSizeBytes(128)
.build())
.build();
}
private static final class EmptyRepository implements VehicleStateRepository {
@Override public void putState(String vin, String json) {}
@Override public void putLocation(String vin, String json) {}