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

refactor: remove event file store contracts

Browse Source
This commit is contained in:
lingniu
2026-07-01 11:18:02 +08:00
parent 7ce64e9963
commit 986fc2fac3
20 changed files with 151 additions and 1315 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
docs/module-data-flow.html
View File

@@ -317,7 +317,7 @@ flowchart TB
end
subgraph sink["输出与明细存储 modules/sinks"]
mq["sink-kafka<br/>VehicleEvent 到 Protobuf Envelope 到 Kafka"]
kafka["sink-kafka<br/>VehicleEvent 到 Protobuf Envelope 到 Kafka"]
archive["sink-archive<br/>RawArchive 到本地/S3/OSS 冷存"]
tdengineStore["tdengine-history-store<br/>TDengine raw_frames + locations"]
end
@@ -362,13 +362,13 @@ flowchart TB
api --> registry
registry --> dispatcher
dispatcher --> bus
bus --> mq
bus --> kafka
bus --> archive
mq --> historyApp
mq --> analyticsApp
kafka --> historyApp
kafka --> analyticsApp
historyApp --> history
analyticsApp --> stat
mq -.可选热状态.-> state
kafka -.可选热状态.-> state
history --> tdengineStore
gateway --> session
@@ -379,7 +379,7 @@ flowchart TB
obs -.监控.-> dispatcher
obs -.监控.-> bus
obs -.监控.-> mq
obs -.监控.-> kafka
classDef entry fill:#eff6ff,stroke:#2563eb,color:#1e3a8a;
classDef core fill:#ecfdf3,stroke:#0f766e,color:#134e48;
@@ -389,7 +389,7 @@ flowchart TB
class gb,jt808,jt1078,jsatl12,mqtt,mqttProfile,gbApp,jtApp,yutongApp,historyApp,analyticsApp entry;
class api,registry,dispatcher,bus,identity core;
class mq,archive,tdengineStore sink;
class kafka,archive,tdengineStore sink;
class gateway,session command;
class codec,obs support;
</pre>

36
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java
View File

@@ -270,6 +270,26 @@ class MavenModuleProfileTest {
.isFalse();
}
@Test
void legacyEventFileStoreContractsAreRemovedFromProductionSources() throws Exception {
Path root = repositoryRoot();
assertThat(Files.exists(root.resolve("modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history")))
.isFalse();
assertThat(Files.exists(root.resolve(
"modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java")))
.isFalse();
assertThat(Files.exists(root.resolve(
"modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java")))
.isFalse();
assertThat(productionJavaContains("EventFileStore"))
.isFalse();
assertThat(productionJavaContains("EventFileRecord"))
.isFalse();
assertThat(productionJavaContains("EventFileQuery"))
.isFalse();
}
private static Document rootPom() throws Exception {
return modulePom("pom.xml");
}
@@ -411,6 +431,22 @@ class MavenModuleProfileTest {
return false;
}
private static boolean productionJavaContains(String token) throws Exception {
try (Stream<Path> files = Files.walk(repositoryRoot().resolve("modules"))) {
return files
.filter(Files::isRegularFile)
.filter(path -> path.toString().endsWith(".java"))
.filter(path -> !path.toString().contains("/src/test/"))
.anyMatch(path -> {
try {
return Files.readString(path).contains(token);
} catch (Exception e) {
throw new IllegalStateException("failed to read " + path, e);
}
});
}
}
private static Element firstDirectChild(Element parent, String tagName) {
NodeList children = parent.getChildNodes();
for (int i = 0; i < children.getLength(); i++) {

75
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java
View File

@@ -3,7 +3,6 @@ package com.lingniu.ingest.historyapp;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
@@ -19,14 +18,9 @@ import com.lingniu.ingest.sink.kafka.KafkaEnvelopeDeadLetterSink;
import com.lingniu.ingest.sink.kafka.KafkaEventSink;
import com.lingniu.ingest.sink.kafka.KafkaEnvelopeConsumerRunner;
import com.lingniu.ingest.sink.kafka.KafkaSinkProperties;
import com.lingniu.ingest.sink.kafka.proto.ParseStatusProto;
import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef;
import com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload;
import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
import com.lingniu.ingest.sink.kafka.KafkaSinkAutoConfiguration;
import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow;
import com.lingniu.ingest.tdenginehistory.config.TdengineHistoryAutoConfiguration;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.junit.jupiter.api.Test;
@@ -37,16 +31,12 @@ import org.springframework.context.annotation.Bean;
import java.lang.reflect.Method;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
class VehicleHistoryAppCompositionTest {
@@ -171,32 +161,6 @@ class VehicleHistoryAppCompositionTest {
});
}
@Test
void historyIngestorIgnoresEventFileStoreBeanAndWritesTdengineOnly() {
EventFileStore eventFileStore = mock(EventFileStore.class);
CapturingTdengineWriter writer = new CapturingTdengineWriter();
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class)
.withBean(EventFileStore.class, () -> eventFileStore)
.withBean(TdengineHistoryWriter.class, () -> writer)
.withBean(EnvelopeDeadLetterSink.class, () -> mock(EnvelopeDeadLetterSink.class))
.withPropertyValues(
"lingniu.ingest.event-history.enabled=true",
"lingniu.ingest.sink.kafka.consumer.enabled=false")
.run(context -> {
EventHistoryEnvelopeIngestor ingestor = context.getBean(EventHistoryEnvelopeIngestor.class);
ingestor.tryIngest(rawFrameEnvelope().toByteArray());
verify(eventFileStore, never()).appendAll(any());
assertThat(writer.rawFrames)
.extracting(TdengineRawFrameRow::frameId)
.containsExactly("raw-frame-1");
});
}
@SuppressWarnings("unchecked")
private static KafkaProducer<String, byte[]> kafkaProducer() {
return mock(KafkaProducer.class);
@@ -226,43 +190,4 @@ class VehicleHistoryAppCompositionTest {
return binding;
}
private static VehicleEnvelope rawFrameEnvelope() {
return VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
.setEventId("raw-event-1")
.setVin("VINRAW001")
.setSource("JT808")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://jt808/2026/06/29/raw-frame-1.bin")
.setSizeBytes(68))
.setRawFrameFact(RawFrameFactPayload.newBuilder()
.setFrameId("raw-frame-1")
.setVehicleKey("jt808:013800000000")
.setVin("VINRAW001")
.setPhone("013800000000")
.setMessageId(0x0200)
.setRawUri("archive://jt808/2026/06/29/raw-frame-1.bin")
.setRawSizeBytes(68)
.setParseStatus(ParseStatusProto.PARSE_STATUS_SUCCEEDED))
.build();
}
private static final class CapturingTdengineWriter implements TdengineHistoryWriter {
private final List<TdengineRawFrameRow> rawFrames = new ArrayList<>();
@Override
public void appendRawFrames(List<TdengineRawFrameRow> rows) {
rawFrames.addAll(rows);
}
@Override
public void appendLocations(List<com.lingniu.ingest.tdenginehistory.TdengineLocationRow> rows) {
}
@Override
public void appendTelemetryFields(List<com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow> rows) {
}
}
}

4
modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java
View File

@@ -144,8 +144,8 @@ public sealed interface VehicleEvent
* 原始报文归档事件：每条成功解码的入站帧由 Dispatcher 产出一条，携带原始字节和归档索引信息。
* 32960 历史全字段查询依赖它对应的 {@code archive://...} 文件能够被回读。
*
* <p>当前 {@code KafkaEventSink} 默认不发送本类型，{@code EventFileStoreSink} 只保存索引不保存
* bytes；因此启用新 raw 落盘/转发实现时，必须同时保证 {@link RawArchiveKeys} 的 key 规则一致。
* <p>当前 {@code KafkaEventSink} 会发送本类型的索引信息，RAW bytes 由 archive 落盘保存；
* 因此启用 raw 落盘/转发实现时，必须同时保证 {@link RawArchiveKeys} 的 key 规则一致。
*
* @param command 协议主命令码（如 32960 0x02/0x03 等），冗余在事件里便于按命令分片归档
* @param infoType 协议子类型（可为 0），同上

93
modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java
View File

@@ -1,93 +0,0 @@
package com.lingniu.ingest.api.history;
import com.lingniu.ingest.api.ProtocolId;
import java.time.Instant;
import java.time.LocalDate;
/**
* Query criteria for the optional file-backed historical event index.
*/
public record EventFileQuery(
ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Instant eventTimeFrom,
Instant eventTimeTo,
Order order,
int limit,
String vin,
String eventType,
Instant cursorEventTime,
Instant cursorIngestTime,
String cursorEventId
) {
public enum Order {
ASC,
DESC
}
public EventFileQuery {
if (protocol == null) {
throw new IllegalArgumentException("protocol must not be null");
}
if (dateFrom == null || dateTo == null) {
throw new IllegalArgumentException("date range must not be null");
}
if (dateTo.isBefore(dateFrom)) {
throw new IllegalArgumentException("dateTo must not be before dateFrom");
}
if (eventTimeFrom != null && eventTimeTo != null && eventTimeTo.isBefore(eventTimeFrom)) {
throw new IllegalArgumentException("eventTimeTo must not be before eventTimeFrom");
}
order = order == null ? Order.ASC : order;
limit = limit <= 0 ? 100 : limit;
vin = vin == null || vin.isBlank() ? null : vin.trim();
eventType = eventType == null || eventType.isBlank() ? null : eventType.trim();
cursorEventId = cursorEventId == null || cursorEventId.isBlank() ? null : cursorEventId.trim();
if ((cursorEventTime == null || cursorIngestTime == null || cursorEventId == null)
&& !(cursorEventTime == null && cursorIngestTime == null && cursorEventId == null)) {
throw new IllegalArgumentException("cursor eventTime, ingestTime and eventId must be provided together");
}
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Instant eventTimeFrom,
Instant eventTimeTo,
Order order,
int limit,
String vin,
String eventType) {
this(protocol, dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, limit,
vin, eventType, null, null, null);
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Order order,
int limit,
String vin,
String eventType) {
this(protocol, dateFrom, dateTo, null, null, order, limit, vin, eventType);
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Order order,
int limit) {
this(protocol, dateFrom, dateTo, order, limit, null, null);
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Order order,
int limit,
String vin) {
this(protocol, dateFrom, dateTo, order, limit, vin, null);
}
}

41
modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java
View File

@@ -1,41 +0,0 @@
package com.lingniu.ingest.api.history;
import com.lingniu.ingest.api.ProtocolId;
import java.time.Instant;
import java.util.Map;
/**
* Minimal historical event record shared by optional compatibility stores and history queries.
*/
public record EventFileRecord(
String eventId,
ProtocolId protocol,
String eventType,
String vin,
Instant eventTime,
Instant ingestTime,
String rawArchiveUri,
Map<String, String> metadata,
String payloadJson
) {
public EventFileRecord {
if (eventId == null || eventId.isBlank()) {
throw new IllegalArgumentException("eventId must not be blank");
}
if (protocol == null) {
throw new IllegalArgumentException("protocol must not be null");
}
if (eventTime == null) {
throw new IllegalArgumentException("eventTime must not be null");
}
if (ingestTime == null) {
throw new IllegalArgumentException("ingestTime must not be null");
}
eventType = eventType == null ? "" : eventType;
vin = vin == null ? "" : vin;
rawArchiveUri = rawArchiveUri == null ? "" : rawArchiveUri;
metadata = metadata == null ? Map.of() : Map.copyOf(metadata);
payloadJson = payloadJson == null || payloadJson.isBlank() ? "{}" : payloadJson;
}
}

26
modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java
View File

@@ -1,26 +0,0 @@
package com.lingniu.ingest.api.history;
import java.io.IOException;
import java.util.List;
/**
* Historical event index contract.
*
* <p>The default production history path uses TDengine directly. This contract remains in core so
* optional compatibility implementations can still plug into query code without pulling their
* storage engines into the default reactor.
*/
public interface EventFileStore {
default void append(EventFileRecord record) throws IOException {
appendAll(List.of(record));
}
void appendAll(List<EventFileRecord> records) throws IOException;
List<EventFileRecord> query(EventFileQuery query) throws IOException;
default EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
return null;
}
}

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

@@ -1,122 +0,0 @@
package com.lingniu.ingest.eventhistory;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.util.List;
import java.util.Map;
/**
* 通用历史事件查询接口。
*
* <p>这是跨协议的低层记录查询，返回的是 EventFileStore 中的 envelope/snapshot 记录。
* GB32960 业务查询应优先使用 {@link Gb32960FrameController}，因为它会回读 RAW 并按协议展开全部字段。
*/
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
@ConditionalOnBean(EventFileStore.class)
@RequestMapping("/api/event-history")
public class EventHistoryController {
private static final ZoneId DEFAULT_ZONE = ZoneId.of("Asia/Shanghai");
private final EventFileStore store;
public EventHistoryController(EventFileStore store) {
this.store = store;
}
@GetMapping("/records")
public List<RecordResponse> query(
@RequestParam ProtocolId protocol,
@RequestParam String dateFrom,
@RequestParam String dateTo,
@RequestParam(defaultValue = "ASC") EventFileQuery.Order order,
@RequestParam(defaultValue = "100") int limit,
@RequestParam(required = false) String vin,
@RequestParam(required = false) String eventType,
@RequestParam(required = false) String cursorEventTime,
@RequestParam(required = false) String cursorIngestTime,
@RequestParam(required = false) String cursorEventId) throws IOException {
return queryRecords(protocol, dateFrom, dateTo, order, limit, vin, eventType,
cursorEventTime, cursorIngestTime, cursorEventId).stream()
.map(RecordResponse::from)
.toList();
}
private List<EventFileRecord> queryRecords(ProtocolId protocol,
String dateFrom,
String dateTo,
EventFileQuery.Order order,
int limit,
String vin,
String eventType,
String cursorEventTime,
String cursorIngestTime,
String cursorEventId) throws IOException {
QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
return store.query(new EventFileQuery(
protocol,
range.dateFrom(),
range.dateTo(),
range.eventTimeFrom(),
range.eventTimeTo(),
order,
limit,
vin,
eventType,
instantParam(cursorEventTime),
instantParam(cursorIngestTime),
cursorEventId));
}
private static Instant instantParam(String raw) {
if (raw == null || raw.isBlank()) {
return null;
}
String value = raw.trim();
try {
return OffsetDateTime.parse(value).toInstant();
} catch (RuntimeException ignored) {
return LocalDateTime.parse(value).atZone(DEFAULT_ZONE).toInstant();
}
}
public record RecordResponse(
String eventId,
ProtocolId protocol,
String eventType,
String vin,
String eventTime,
String ingestTime,
String rawArchiveUri,
Map<String, String> metadata,
String payloadJson) {
private static RecordResponse from(EventFileRecord record) {
return new RecordResponse(
record.eventId(),
record.protocol(),
record.eventType(),
record.vin(),
record.eventTime().toString(),
record.ingestTime().toString(),
record.rawArchiveUri(),
record.metadata(),
record.payloadJson());
}
}
}

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

@@ -3,8 +3,6 @@ package com.lingniu.ingest.eventhistory;
import com.google.protobuf.InvalidProtocolBufferException;
import com.lingniu.ingest.api.consumer.EnvelopeBatchIngestor;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
import com.lingniu.ingest.tdenginehistory.TdengineEnvelopeRows;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
@@ -16,8 +14,7 @@ import java.util.ArrayList;
import java.util.List;
/**
* Consumer-side entry point that stores one Kafka envelope value into the
* historical detail file store.
* Consumer-side entry point that stores Kafka envelope values into TDengine history tables.
*
* <p>当前 32960 本机运行配置没有启用 Kafka consumer；这个类保留给“其他服务把 envelope
* 写回历史库”的解耦部署方式。失败时返回结构化结果，由 Kafka worker 决定是否进 DLQ。
@@ -26,62 +23,24 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
private static final Logger log = LoggerFactory.getLogger(EventHistoryEnvelopeIngestor.class);
private final EventFileStore store;
private final TelemetryEnvelopeRecordMapper mapper;
private final TdengineHistoryWriter tdengineWriter;
private final boolean telemetryFieldsEnabled;
public EventHistoryEnvelopeIngestor(EventFileStore store, TelemetryEnvelopeRecordMapper mapper) {
this(store, mapper, null);
}
public EventHistoryEnvelopeIngestor(TdengineHistoryWriter tdengineWriter) {
this(tdengineWriter, false);
}
public EventHistoryEnvelopeIngestor(TdengineHistoryWriter tdengineWriter,
boolean telemetryFieldsEnabled) {
this(null, null, tdengineWriter, false, telemetryFieldsEnabled);
}
public EventHistoryEnvelopeIngestor(EventFileStore store,
TelemetryEnvelopeRecordMapper mapper,
TdengineHistoryWriter tdengineWriter) {
this(store, mapper, tdengineWriter, true, false);
}
public EventHistoryEnvelopeIngestor(EventFileStore store,
TelemetryEnvelopeRecordMapper mapper,
TdengineHistoryWriter tdengineWriter,
boolean telemetryFieldsEnabled) {
this(store, mapper, tdengineWriter, true, telemetryFieldsEnabled);
}
private EventHistoryEnvelopeIngestor(EventFileStore store,
TelemetryEnvelopeRecordMapper mapper,
TdengineHistoryWriter tdengineWriter,
boolean requireStore,
boolean telemetryFieldsEnabled) {
if (store == null) {
if (requireStore) {
throw new IllegalArgumentException("store must not be null");
}
if (tdengineWriter == null) {
throw new IllegalArgumentException("tdengineWriter must not be null");
}
if (store != null && mapper == null) {
throw new IllegalArgumentException("mapper must not be null");
}
this.store = store;
this.mapper = mapper;
this.tdengineWriter = tdengineWriter;
this.telemetryFieldsEnabled = telemetryFieldsEnabled;
}
public void ingest(byte[] kafkaValue) throws IOException {
VehicleEnvelope envelope = parse(kafkaValue);
if (store != null) {
EventFileRecord record = mapper.toRecord(envelope);
store.append(record);
}
writeTdengineFacts(List.of(envelope));
}
@@ -101,9 +60,8 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
VehicleEnvelope envelope = null;
try {
envelope = parse(kafkaValue);
EventFileRecord record = store == null ? null : mapper.toRecord(envelope);
results.add(null);
valid.add(new BatchEntry(results.size() - 1, envelope, record));
valid.add(new BatchEntry(results.size() - 1, envelope));
} catch (IllegalArgumentException ex) {
results.add(envelope == null
? EnvelopeIngestResult.invalid(ex.getMessage())
@@ -114,9 +72,6 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
return List.copyOf(results);
}
try {
if (store != null) {
store.appendAll(valid.stream().map(BatchEntry::record).toList());
}
writeTdengineFacts(valid.stream().map(BatchEntry::envelope).toList());
for (BatchEntry entry : valid) {
results.set(entry.index(), EnvelopeIngestResult.stored(
@@ -124,7 +79,7 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
entry.vin()));
}
} catch (IOException ex) {
log.error("event history TDengine/file-store ingest failed batchSize={} validSize={} firstEventId={} firstVin={}",
log.error("event history TDengine ingest failed batchSize={} validSize={} firstEventId={} firstVin={}",
kafkaValues.size(),
valid.size(),
valid.getFirst().envelope().getEventId(),
@@ -140,13 +95,13 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
return List.copyOf(results);
}
private record BatchEntry(int index, VehicleEnvelope envelope, EventFileRecord record) {
private record BatchEntry(int index, VehicleEnvelope envelope) {
private String eventId() {
return record == null ? envelope.getEventId() : record.eventId();
return envelope.getEventId();
}
private String vin() {
return record == null ? envelope.getVin() : record.vin();
return envelope.getVin();
}
}
@@ -162,9 +117,6 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
}
private void writeTdengineFacts(List<VehicleEnvelope> envelopes) throws IOException {
if (tdengineWriter == null) {
return;
}
var rawFrames = envelopes.stream()
.flatMap(envelope -> TdengineEnvelopeRows.rawFrame(envelope).stream())
.toList();

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

@@ -2,10 +2,6 @@ package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.model.Gb32960Message;
import com.lingniu.ingest.protocol.gb32960.model.InfoBlock;
@@ -38,8 +34,8 @@ import java.util.Map;
/**
* GB32960 RAW 历史帧查询和业务快照组装服务。
*
* <p>历史库里保存的是 RAW_ARCHIVE 索引，不保存完整解码后的宽表。查询时先从
* {@link EventFileStore} 找到 archive:// URI，再读取本地 RAW .bin，用当前协议解析器即时解码。
* <p>TDengine raw_frames 保存 archive:// URI。查询时先从 TDengine 找到 RAW 索引，
* 再读取本地 RAW .bin，用当前协议解析器即时解码。
* 这种设计让写入路径保持轻量，也允许后续修复解析器后直接重放历史 RAW 得到新字段。
*
* <p>snapshot 的聚合单位是 {@code vin + eventTime}。同一时刻可能有多个 32960 子包，
@@ -50,37 +46,20 @@ public final class Gb32960DecodedFrameService {
private static final Logger log = LoggerFactory.getLogger(Gb32960DecodedFrameService.class);
private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
private static final int OUTPUT_DOUBLE_SCALE = 6;
private static final String RAW_ARCHIVE_EVENT_TYPE = "RAW_ARCHIVE";
private static final ZoneId DEFAULT_ZONE = ZoneId.of("Asia/Shanghai");
private final EventFileStore store;
private final TdengineHistoryReader tdengineReader;
private final Gb32960MessageDecoder decoder;
private final Path archiveRoot;
private final ObjectMapper objectMapper;
public Gb32960DecodedFrameService(EventFileStore store,
Gb32960MessageDecoder decoder,
Path archiveRoot,
ObjectMapper objectMapper) {
this(store, null, decoder, archiveRoot, objectMapper);
}
public Gb32960DecodedFrameService(TdengineHistoryReader tdengineReader,
Gb32960MessageDecoder decoder,
Path archiveRoot,
ObjectMapper objectMapper) {
this(null, tdengineReader, decoder, archiveRoot, objectMapper);
}
public Gb32960DecodedFrameService(EventFileStore store,
TdengineHistoryReader tdengineReader,
Gb32960MessageDecoder decoder,
Path archiveRoot,
ObjectMapper objectMapper) {
if (store == null && tdengineReader == null) {
if (tdengineReader == null) {
log.info("gb32960 decoded frame service has no history index backend; rawUri direct lookup remains available");
} else if (store == null) {
} else {
log.info("gb32960 decoded frame service using TDengine raw_frames index");
}
if (decoder == null) {
@@ -89,7 +68,6 @@ public final class Gb32960DecodedFrameService {
if (archiveRoot == null) {
throw new IllegalArgumentException("archiveRoot must not be null");
}
this.store = store;
this.tdengineReader = tdengineReader;
this.decoder = decoder;
this.archiveRoot = archiveRoot.toAbsolutePath().normalize();
@@ -98,7 +76,7 @@ public final class Gb32960DecodedFrameService {
public List<DecodedFrame> query(LocalDate dateFrom,
LocalDate dateTo,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -109,39 +87,28 @@ public final class Gb32960DecodedFrameService {
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
int frameLimit = Math.max(1, Math.min(limit, 1000));
List<DecodedFrame> fromEventStore = store == null ? List.of() : queryFromEventStore(
dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, frameLimit, vin, platformAccount);
if (!fromEventStore.isEmpty() || tdengineReader == null) {
return fromEventStore;
if (tdengineReader == null) {
return List.of();
}
return queryFromTdengine(dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, frameLimit, vin, platformAccount);
}
public DecodedFramePage queryPage(QueryTimeRange range,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount,
TdenginePageCursor cursor) throws IOException {
int frameLimit = Math.max(1, Math.min(limit, 1000));
if (tdengineReader != null && vin != null && !vin.isBlank()) {
return queryPageFromTdengine(
range.dateFrom(),
range.dateTo(),
range.eventTimeFrom(),
range.eventTimeTo(),
order,
frameLimit,
vin,
platformAccount,
cursor);
if (tdengineReader == null || vin == null || vin.isBlank()) {
return new DecodedFramePage(List.of(), null);
}
List<DecodedFrame> items = query(
return queryPageFromTdengine(
range.dateFrom(),
range.dateTo(),
range.eventTimeFrom(),
@@ -149,55 +116,15 @@ public final class Gb32960DecodedFrameService {
order,
frameLimit,
vin,
platformAccount);
return new DecodedFramePage(items, null);
}
private List<DecodedFrame> queryFromEventStore(LocalDate dateFrom,
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
EventFileQuery.Order order,
int frameLimit,
String vin,
String platformAccount) throws IOException {
// 只查 RAW_ARCHIVE：REALTIME/LOCATION 等派生事件不再作为 32960 历史查询的数据源。
List<EventFileRecord> records = store.query(new EventFileQuery(
ProtocolId.GB32960,
dateFrom,
dateTo,
eventTimeFrom,
eventTimeTo,
order == null ? EventFileQuery.Order.DESC : order,
frameLimit,
vin,
RAW_ARCHIVE_EVENT_TYPE));
List<DecodedFrame> out = new ArrayList<>();
LinkedHashSet<String> seenUris = new LinkedHashSet<>();
for (EventFileRecord record : records) {
String rawArchiveUri = rawArchiveUri(record);
// 同一个 RAW 可能因为索引重建或事件补偿被看到多次，按 URI 去重保证返回帧唯一。
if (rawArchiveUri == null || rawArchiveUri.isBlank() || !seenUris.add(rawArchiveUri)) {
continue;
}
DecodedFrame frame = decodeIfAvailable(record, rawArchiveUri, platformAccount);
if (frame == null) {
continue;
}
out.add(frame);
if (out.size() >= frameLimit) {
break;
}
}
return out;
platformAccount,
cursor);
}
private List<DecodedFrame> queryFromTdengine(LocalDate dateFrom,
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
EventFileQuery.Order order,
HistoryQueryOrder order,
int frameLimit,
String vin,
String platformAccount) throws IOException {
@@ -229,7 +156,7 @@ public final class Gb32960DecodedFrameService {
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
EventFileQuery.Order order,
HistoryQueryOrder order,
int frameLimit,
String vin,
String platformAccount,
@@ -261,7 +188,7 @@ public final class Gb32960DecodedFrameService {
}
public List<LogicalSnapshot> snapshots(QueryTimeRange range,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -270,7 +197,7 @@ public final class Gb32960DecodedFrameService {
}
public List<FieldSnapshot> snapshotFields(QueryTimeRange range,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount,
@@ -286,7 +213,7 @@ public final class Gb32960DecodedFrameService {
}
public String snapshotFieldsCsv(QueryTimeRange range,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount,
@@ -324,18 +251,12 @@ public final class Gb32960DecodedFrameService {
if (rawArchiveUri == null || rawArchiveUri.isBlank()) {
return null;
}
if (store != null) {
EventFileRecord record = store.findByRawArchiveUri(rawArchiveUri);
if (record != null) {
return decode(record, rawArchiveUri(record), platformAccount);
}
}
return decodeWithoutIndex(rawArchiveUri, platformAccount);
}
public List<LogicalSnapshot> snapshots(LocalDate dateFrom,
LocalDate dateTo,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -349,7 +270,7 @@ public final class Gb32960DecodedFrameService {
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
EventFileQuery.Order order,
HistoryQueryOrder order,
int limit,
String vin,
String platformAccount) throws IOException {
@@ -378,19 +299,6 @@ public final class Gb32960DecodedFrameService {
return out;
}
private DecodedFrame decodeIfAvailable(EventFileRecord record,
String rawArchiveUri,
String platformAccount) throws IOException {
try {
return decode(record, rawArchiveUri, platformAccount);
} catch (NoSuchFileException e) {
// 索引存在但 RAW 文件被人工清理时不中断整次查询，返回仍可用的其他帧。
log.warn("skip gb32960 frame because raw archive is missing eventId={} rawArchiveUri={}",
record.eventId(), rawArchiveUri);
return null;
}
}
private DecodedFrame decodeIfAvailable(TdengineRawFrameRow row, String platformAccount) throws IOException {
try {
return decode(row, platformAccount);
@@ -401,29 +309,6 @@ public final class Gb32960DecodedFrameService {
}
}
private DecodedFrame decode(EventFileRecord record, String rawArchiveUri, String platformAccount) throws IOException {
Path rawPath = archivePath(rawArchiveUri);
byte[] rawBytes = Files.readAllBytes(rawPath);
Gb32960Message message = decoder.decode(ByteBuffer.wrap(rawBytes), platformAccount(record, platformAccount));
List<Map<String, Object>> blocks = new ArrayList<>();
for (InfoBlock block : message.infoBlocks()) {
blocks.add(blockJson(block));
}
return new DecodedFrame(
record.eventId(),
message.header().vin(),
message.header().command().name(),
message.header().responseFlag().name(),
message.header().protocolVersion().name(),
message.header().encryptType().name(),
message.header().dataLength(),
message.header().eventTime() == null ? null : message.header().eventTime().toString(),
record.ingestTime().toString(),
rawArchiveUri,
rawBytes.length,
blocks);
}
private DecodedFrame decode(TdengineRawFrameRow row, String platformAccount) throws IOException {
Path rawPath = archivePath(row.rawUri());
byte[] rawBytes = Files.readAllBytes(rawPath);
@@ -618,26 +503,6 @@ public final class Gb32960DecodedFrameService {
return path;
}
private static String rawArchiveUri(EventFileRecord record) {
if (record.rawArchiveUri() != null && !record.rawArchiveUri().isBlank()) {
return record.rawArchiveUri();
}
if (record.metadata() == null) {
return "";
}
return record.metadata().getOrDefault("rawArchiveUri", "");
}
private static String platformAccount(EventFileRecord record, String override) {
if (override != null && !override.isBlank()) {
return override;
}
if (record.metadata() == null) {
return null;
}
return record.metadata().get("platformAccount");
}
private String platformAccount(TdengineRawFrameRow row, String override) {
if (override != null && !override.isBlank()) {
return override;
@@ -663,8 +528,8 @@ public final class Gb32960DecodedFrameService {
return inclusive.plusMillis(1);
}
private static TdengineQueryOrder tdengineOrder(EventFileQuery.Order order) {
return order == EventFileQuery.Order.ASC ? TdengineQueryOrder.ASC : TdengineQueryOrder.DESC;
private static TdengineQueryOrder tdengineOrder(HistoryQueryOrder order) {
return order == HistoryQueryOrder.ASC ? TdengineQueryOrder.ASC : TdengineQueryOrder.DESC;
}
private static String rawArchiveEventId(String rawArchiveUri) {

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

@@ -1,6 +1,5 @@
package com.lingniu.ingest.eventhistory;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
@@ -23,7 +22,7 @@ import java.util.List;
*
* <p>这个 Controller 只做参数校验、时间范围解析和 Swagger 说明，真正的 RAW 读取、
* 协议解码、snapshot 合并、字段投影都放在 {@link Gb32960DecodedFrameService}。
* 这样可以保证接口层足够薄，后续如果要把查询服务拆出去，只需要迁移 service 和 store。
* 这样可以保证接口层足够薄，后续如果要把查询服务拆出去，只需要迁移 service。
*/
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
@@ -52,7 +51,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
@RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
@RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回帧数量上限。", example = "10")
@RequestParam(defaultValue = "10") int limit,
@Parameter(description = "车辆 VIN；用于缩小到单车查询。", example = "LB9A32A20P0LS1257")
@@ -82,7 +81,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
@RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
@RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回帧数量上限，最大 1000。", example = "100")
@RequestParam(defaultValue = "100") int limit,
@Parameter(description = "车辆 VIN，必填；用于命中 TDengine raw child table。", required = true, example = "LB9A32A20P0LS1257")
@@ -139,7 +138,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
@RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
@RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回 snapshot 数量上限。", example = "10")
@RequestParam(defaultValue = "10") int limit,
@Parameter(description = "车辆 VIN，必填；snapshot 查询只支持单车，避免跨车合并和低效扫描。", required = true, example = "LB9A32A20P0LS1257")
@@ -163,7 +162,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
@RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
@RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回 snapshot 数量上限。", example = "10")
@RequestParam(defaultValue = "10") int limit,
@Parameter(description = "车辆 VIN，必填；字段查询只支持单车。", required = true, example = "LB9A32A20P0LS1257")
@@ -189,7 +188,7 @@ public final class Gb32960FrameController {
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
@RequestParam(defaultValue = "DESC") EventFileQuery.Order order,
@RequestParam(defaultValue = "DESC") HistoryQueryOrder order,
@Parameter(description = "返回 snapshot 数量上限。", example = "1000")
@RequestParam(defaultValue = "1000") int limit,
@Parameter(description = "车辆 VIN，必填；字段导出只支持单车。", required = true, example = "LB9A32A20P0LS1257")

6
modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java Normal file
View File

@@ -0,0 +1,6 @@
package com.lingniu.ingest.eventhistory;
public enum HistoryQueryOrder {
ASC,
DESC
}

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

@@ -1,173 +0,0 @@
package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef;
import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import java.time.Instant;
import java.util.LinkedHashMap;
import java.util.Map;
/**
* Maps Kafka protobuf envelopes into records stored by the historical detail
* file store.
*
* <p>这是跨协议 Kafka envelope 回灌历史库的通用适配器。它保存的是 protobuf
* telemetry_snapshot 的 JSON 视图，不负责 GB32960 原始包解码；32960 全字段 snapshot
* 查询仍以 rawArchiveUri 指向的 .bin 为准。
*/
public final class TelemetryEnvelopeRecordMapper {
private static final JsonFormat.Printer JSON_PRINTER = JsonFormat.printer();
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
public EventFileRecord toRecord(VehicleEnvelope envelope) {
if (envelope == null) {
throw new IllegalArgumentException("envelope must not be null");
}
if (envelope.hasTelemetrySnapshot()) {
return telemetryRecord(envelope);
}
if (envelope.hasRawArchive()) {
return rawArchiveRecord(envelope);
}
throw new IllegalArgumentException("envelope telemetry_snapshot or raw_archive is required");
}
private static EventFileRecord telemetryRecord(VehicleEnvelope envelope) {
ProtocolId protocol = protocol(envelope.getSource());
Map<String, String> metadata = metadata(envelope, protocol);
return new EventFileRecord(
envelope.getEventId(),
protocol,
envelope.getTelemetrySnapshot().getEventType(),
envelope.getVin(),
Instant.ofEpochMilli(envelope.getEventTimeMs()),
Instant.ofEpochMilli(envelope.getIngestTimeMs()),
envelope.getTelemetrySnapshot().getRawArchiveUri(),
metadata,
snapshotJson(envelope)
);
}
private static EventFileRecord rawArchiveRecord(VehicleEnvelope envelope) {
ProtocolId protocol = protocol(envelope.getSource());
RawArchiveRef rawArchive = envelope.getRawArchive();
Map<String, String> metadata = metadata(envelope, protocol);
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);
}
return new EventFileRecord(
envelope.getEventId(),
protocol,
"RAW_ARCHIVE",
envelope.getVin(),
Instant.ofEpochMilli(envelope.getEventTimeMs()),
Instant.ofEpochMilli(envelope.getIngestTimeMs()),
rawArchiveUri,
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()) {
metadata.putIfAbsent("protocolVersion", envelope.getProtocolVersion());
}
if (protocol == ProtocolId.UNKNOWN && !envelope.getSource().isBlank()) {
// 保留未知 source 原文，避免历史库标准化成 UNKNOWN 后丢失排障线索。
metadata.putIfAbsent("originalSource", envelope.getSource());
}
return metadata;
}
private static ProtocolId protocol(String source) {
if (source == null || source.isBlank()) {
return ProtocolId.UNKNOWN;
}
try {
return ProtocolId.valueOf(source);
} catch (IllegalArgumentException ex) {
return ProtocolId.UNKNOWN;
}
}
private static String snapshotJson(VehicleEnvelope envelope) {
try {
return JSON_PRINTER.print(envelope.getTelemetrySnapshot());
} catch (InvalidProtocolBufferException e) {
throw new IllegalArgumentException("failed to serialize telemetry_snapshot", e);
}
}
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());
payload.put("vin", envelope.getVin());
payload.put("protocol", protocol(envelope.getSource()).name());
if (!envelope.getSource().isBlank()) {
payload.put("source", envelope.getSource());
}
payload.put("eventType", "RAW_ARCHIVE");
payload.put("eventTime", Instant.ofEpochMilli(envelope.getEventTimeMs()).toString());
payload.put("ingestTime", Instant.ofEpochMilli(envelope.getIngestTimeMs()).toString());
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());
putParsedJson(payload, rawArchive.getParsedJson());
payload.put("metadata", metadata);
try {
return OBJECT_MAPPER.writeValueAsString(payload);
} catch (JsonProcessingException e) {
throw new IllegalArgumentException("failed to serialize raw_archive", e);
}
}
private static void putParsedJson(Map<String, Object> payload, String parsedJson) {
if (parsedJson == null || parsedJson.isBlank()) {
return;
}
try {
payload.put("parsed", OBJECT_MAPPER.readTree(parsedJson));
} catch (JsonProcessingException ex) {
payload.put("parsed", parsedJson);
payload.put("parsedJsonError", ex.getMessage() == null ? ex.getClass().getSimpleName() : ex.getMessage());
}
}
}

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

@@ -2,8 +2,6 @@ package com.lingniu.ingest.eventhistory.config;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.eventhistory.EventHistoryController;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
@@ -12,7 +10,6 @@ import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
import com.lingniu.ingest.sink.kafka.KafkaSinkAutoConfiguration;
@@ -37,7 +34,7 @@ import java.nio.file.Path;
* <p>该模块有两类入口：
* <ul>
* <li>TDengine HTTP 查询入口：查询位置、RAW 帧和按需解码的 GB32960 帧。
* <li>兼容 EventFileStore 入口：仅在旧索引显式启用时创建旧记录查询和旧写入路径。
* <li>Kafka consumer 入口：消费 envelope 并写入 TDengine 历史表。
* </ul>
*
* <p>当前生产 history app 以 TDengine 为准；Kafka consumer 是否启动还取决于
@@ -52,22 +49,6 @@ import java.nio.file.Path;
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
public class EventHistoryAutoConfiguration {
@Bean
@ConditionalOnBean(EventFileStore.class)
@ConditionalOnMissingBean
public TelemetryEnvelopeRecordMapper telemetryEnvelopeRecordMapper() {
return new TelemetryEnvelopeRecordMapper();
}
@Bean
@ConditionalOnBean(EventFileStore.class)
@ConditionalOnMissingBean(value = {TdengineHistoryWriter.class, EventHistoryEnvelopeIngestor.class})
public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(EventFileStore store,
TelemetryEnvelopeRecordMapper mapper,
ObjectProvider<TdengineHistoryWriter> writer) {
return new EventHistoryEnvelopeIngestor(store, mapper, writer.getIfAvailable());
}
@Bean
@ConditionalOnBean(TdengineHistoryWriter.class)
@ConditionalOnMissingBean
@@ -94,24 +75,14 @@ public class EventHistoryAutoConfiguration {
return new EnvelopeConsumerProcessor("event-history-raw", ingestor, deadLetterSink);
}
@Bean
@ConditionalOnBean(EventFileStore.class)
@ConditionalOnMissingBean
public EventHistoryController eventHistoryController(EventFileStore store) {
return new EventHistoryController(store);
}
@Bean
@ConditionalOnBean(Gb32960MessageDecoder.class)
@ConditionalOnMissingBean
public Gb32960DecodedFrameService gb32960DecodedFrameService(ObjectProvider<EventFileStore> store,
ObjectProvider<TdengineHistoryReader> reader,
public Gb32960DecodedFrameService gb32960DecodedFrameService(ObjectProvider<TdengineHistoryReader> reader,
Gb32960MessageDecoder decoder,
@Value("${lingniu.ingest.event-history.archive-path:${SINK_ARCHIVE_PATH:./archive/}}")
String archivePath) {
// 优先使用 EventFileStore 索引；TDengine-only 高吞吐运行时可直接从 raw_frames 找 rawUri。
return new Gb32960DecodedFrameService(store.getIfAvailable(), reader.getIfAvailable(),
decoder, archiveRoot(archivePath), null);
return new Gb32960DecodedFrameService(reader.getIfAvailable(), decoder, archiveRoot(archivePath), null);
}
@Bean

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

@@ -1,8 +1,7 @@
package com.lingniu.ingest.eventhistory;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import jakarta.servlet.http.HttpServletRequest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -14,8 +13,6 @@ import org.springframework.web.servlet.resource.NoResourceFoundException;
import java.io.IOException;
import java.nio.file.Path;
import java.util.List;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.assertj.core.api.Assertions.assertThat;
@@ -87,18 +84,8 @@ class ApiExceptionHandlerTest {
}
private Gb32960DecodedFrameService service() {
EventFileStore store = new EventFileStore() {
@Override
public void appendAll(List<EventFileRecord> records) {
}
@Override
public List<EventFileRecord> query(com.lingniu.ingest.api.history.EventFileQuery query) {
return List.of();
}
};
return new Gb32960DecodedFrameService(
store,
(TdengineHistoryReader) null,
mock(Gb32960MessageDecoder.class),
archiveRoot,
null);

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

@@ -1,82 +0,0 @@
package com.lingniu.ingest.eventhistory;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
class EventHistoryControllerTest {
@Test
void queriesRecordsByDateRangeAndOrder() throws Exception {
CapturingStore store = new CapturingStore(List.of(record("event-1")));
EventHistoryController controller = new EventHistoryController(store);
List<EventHistoryController.RecordResponse> records = controller.query(
ProtocolId.GB32960,
"2026-06-21",
"2026-06-22",
EventFileQuery.Order.DESC,
50,
"VIN001",
"RAW_ARCHIVE",
"2026-06-22T08:00:00Z",
"2026-06-22T08:00:01Z",
"event-0");
assertThat(records).extracting(EventHistoryController.RecordResponse::eventId).containsExactly("event-1");
assertThat(records.getFirst().eventTime()).isEqualTo("2026-06-22T08:00:00Z");
assertThat(records.getFirst().ingestTime()).isEqualTo("2026-06-22T08:00:01Z");
assertThat(store.lastQuery.protocol()).isEqualTo(ProtocolId.GB32960);
assertThat(store.lastQuery.dateFrom().toString()).isEqualTo("2026-06-21");
assertThat(store.lastQuery.dateTo().toString()).isEqualTo("2026-06-22");
assertThat(store.lastQuery.order()).isEqualTo(EventFileQuery.Order.DESC);
assertThat(store.lastQuery.limit()).isEqualTo(50);
assertThat(store.lastQuery.vin()).isEqualTo("VIN001");
assertThat(store.lastQuery.eventType()).isEqualTo("RAW_ARCHIVE");
assertThat(store.lastQuery.cursorEventTime()).isEqualTo(Instant.parse("2026-06-22T08:00:00Z"));
assertThat(store.lastQuery.cursorIngestTime()).isEqualTo(Instant.parse("2026-06-22T08:00:01Z"));
assertThat(store.lastQuery.cursorEventId()).isEqualTo("event-0");
}
private static EventFileRecord record(String eventId) {
return new EventFileRecord(
eventId,
ProtocolId.GB32960,
"REALTIME",
"VIN001",
Instant.parse("2026-06-22T08:00:00Z"),
Instant.parse("2026-06-22T08:00:01Z"),
"archive://raw/" + eventId + ".bin",
Map.of(),
"{\"fields\":[]}");
}
private static final class CapturingStore implements EventFileStore {
private final List<EventFileRecord> records;
private EventFileQuery lastQuery;
private CapturingStore(List<EventFileRecord> records) {
this.records = records;
}
@Override
public void appendAll(List<EventFileRecord> records) {
throw new UnsupportedOperationException();
}
@Override
public List<EventFileRecord> query(EventFileQuery query) throws IOException {
this.lastQuery = query;
return records;
}
}
}

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

@@ -1,14 +1,7 @@
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.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef;
import com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload;
import com.lingniu.ingest.sink.kafka.proto.TelemetryField;
@@ -22,8 +15,6 @@ import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
@@ -32,105 +23,36 @@ 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();
EventHistoryEnvelopeIngestor ingestor =
new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
assertThat(ingestor).isInstanceOf(EnvelopeIngestor.class);
ingestor.ingest(envelope("event-1").toByteArray());
assertThat(store.records).extracting(EventFileRecord::eventId).containsExactly("event-1");
assertThat(store.records.getFirst().payloadJson()).contains("total_mileage_km");
}
@Test
void rejectsInvalidEnvelopeBytes() {
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor =
new EventHistoryEnvelopeIngestor(new CapturingStore(), new TelemetryEnvelopeRecordMapper());
new EventHistoryEnvelopeIngestor(tdengineWriter);
assertThat(ingestor).isInstanceOf(EnvelopeIngestor.class);
assertThatThrownBy(() -> ingestor.ingest(new byte[]{0x01, 0x02}))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("VehicleEnvelope");
}
@Test
void tryIngestInvalidEnvelopeBytesReturnsInvalidWithoutAppendingRecord() {
CapturingStore store = new CapturingStore();
void tryIngestInvalidEnvelopeBytesReturnsInvalidWithoutWritingTdengine() {
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor =
new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
new EventHistoryEnvelopeIngestor(tdengineWriter);
var result = ingestor.tryIngest(new byte[]{0x01, 0x02});
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.INVALID_ENVELOPE);
assertThat(result.message()).contains("VehicleEnvelope");
assertThat(store.records).isEmpty();
}
@Test
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")
.setEventId("raw-event-1")
.setVin("VINRAW001")
.setSource("GB32960")
.setProtocolVersion("V2016")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.putMetadata(RawArchiveKeys.META_KEY, rawArchiveKey)
.putMetadata(RawArchiveKeys.META_URI, rawArchiveUri)
.putMetadata("command", "0x0002")
.setRawArchive(RawArchiveRef.newBuilder()
.setUri(rawArchiveUri)
.setSizeBytes(128)
.build())
.build();
EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
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.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_112_400_000L));
assertThat(record.rawArchiveUri()).isEqualTo(rawArchiveUri);
assertThat(record.metadata())
.containsEntry("protocolVersion", "V2016")
.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");
assertThat(tdengineWriter.rawFrames).isEmpty();
assertThat(tdengineWriter.locations).isEmpty();
}
@Test
void tryIngestWritesRawAndLocationFactsToTdengineWhenWriterExists() throws Exception {
CapturingStore store = new CapturingStore();
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(
store, new TelemetryEnvelopeRecordMapper(), tdengineWriter);
EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(tdengineWriter);
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
@@ -172,7 +94,6 @@ class EventHistoryEnvelopeIngestorTest {
EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
assertThat(store.records).hasSize(1);
assertThat(tdengineWriter.rawFrames)
.extracting(TdengineRawFrameRow::frameId)
.containsExactly("frame-jt808-1");
@@ -219,10 +140,8 @@ class EventHistoryEnvelopeIngestorTest {
@Test
void tryIngestAllBatchesFileStoreAndTdengineWrites() throws Exception {
CapturingStore store = new CapturingStore();
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(
store, new TelemetryEnvelopeRecordMapper(), tdengineWriter);
EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(tdengineWriter);
VehicleEnvelope first = jt808LocationEnvelope("jt808-location-1", "frame-jt808-1", "013800000001");
VehicleEnvelope second = jt808LocationEnvelope("jt808-location-2", "frame-jt808-2", "013800000002");
@@ -237,9 +156,6 @@ class EventHistoryEnvelopeIngestorTest {
EnvelopeIngestResult.Status.STORED,
EnvelopeIngestResult.Status.INVALID_ENVELOPE,
EnvelopeIngestResult.Status.STORED);
assertThat(store.appendAllCalls).isEqualTo(1);
assertThat(store.records).extracting(EventFileRecord::eventId)
.containsExactly("jt808-location-1", "jt808-location-2");
assertThat(tdengineWriter.rawFrames).extracting(TdengineRawFrameRow::frameId)
.containsExactly("frame-jt808-1", "frame-jt808-2");
assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId)
@@ -247,24 +163,6 @@ class EventHistoryEnvelopeIngestorTest {
assertThat(tdengineWriter.telemetryFields).isEmpty();
}
private static VehicleEnvelope envelope(String eventId) {
return VehicleEnvelope.newBuilder()
.setEventId(eventId)
.setVin("VIN001")
.setSource("GB32960")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
.setEventType("REALTIME")
.addFields(TelemetryField.newBuilder()
.setKey("total_mileage_km")
.setValueType("DOUBLE")
.setValue("12345.6")
.setUnit("km")
.setQuality("GOOD")))
.build();
}
private static VehicleEnvelope jt808LocationEnvelope(String eventId, String frameId, String phone) {
return VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
@@ -304,22 +202,6 @@ class EventHistoryEnvelopeIngestorTest {
.build();
}
private static final class CapturingStore implements EventFileStore {
private final List<EventFileRecord> records = new ArrayList<>();
private int appendAllCalls;
@Override
public void appendAll(List<EventFileRecord> records) {
appendAllCalls++;
this.records.addAll(records);
}
@Override
public List<EventFileRecord> query(EventFileQuery query) throws IOException {
return List.copyOf(records);
}
}
private static final class CapturingTdengineWriter implements TdengineHistoryWriter {
private final List<TdengineRawFrameRow> rawFrames = new ArrayList<>();
private final List<TdengineLocationRow> locations = new ArrayList<>();

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

@@ -1,10 +1,6 @@
package com.lingniu.ingest.eventhistory;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.codec.BccChecksum;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960BodyParser;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.codec.InfoBlockParser;
@@ -53,18 +49,10 @@ class Gb32960DecodedFrameServiceTest {
Path rawFile = archiveRoot.resolve(key);
Files.createDirectories(rawFile.getParent());
Files.write(rawFile, realtimeFrame());
String uri = "archive://" + key;
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of(new EventFileRecord(
"raw-1",
ProtocolId.GB32960,
"RAW_ARCHIVE",
"LNVFC000000000001",
Instant.parse("2026-06-22T10:00:00Z"),
Instant.parse("2026-06-22T10:00:01Z"),
"archive://" + key,
Map.of("platformAccount", "Hyundai"),
"{}"))),
new CapturingTdengineReader(List.of(rawRow("raw-1", uri))),
decoder(),
archiveRoot,
null);
@@ -72,7 +60,7 @@ class Gb32960DecodedFrameServiceTest {
List<Gb32960DecodedFrameService.DecodedFrame> frames = service.query(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
null,
null);
@@ -109,7 +97,7 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of()),
(TdengineHistoryReader) null,
decoder(),
archiveRoot,
null);
@@ -123,42 +111,6 @@ class Gb32960DecodedFrameServiceTest {
.containsExactly("VEHICLE", "GD_FC_DCDC");
}
@Test
void findsSingleFrameDirectlyByRawArchiveUriWhenEventFileStoreIsDisabled() throws Exception {
String key = "2026/06/22/GB32960/LNVFC000000000001/1792527837000000.bin";
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
(EventFileStore) null,
decoder(),
archiveRoot,
null);
Gb32960DecodedFrameService.DecodedFrame frame =
service.findByRawArchiveUri("archive://" + key, "Hyundai");
assertThat(frame.eventId()).isEqualTo("1792527837000000");
assertThat(frame.vin()).isEqualTo("LNVFC000000000001");
}
@Test
void findsSingleFrameByRawArchiveUriIndexWhenAvailable() throws Exception {
String key = "2026/06/22/GB32960/LNVFC000000000001/1792527837000001.bin";
String uri = "archive://" + key;
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of(record("indexed-raw", uri))),
decoder(),
archiveRoot,
null);
Gb32960DecodedFrameService.DecodedFrame frame = service.findByRawArchiveUri(uri, null);
assertThat(frame.eventId()).isEqualTo("indexed-raw");
assertThat(frame.rawArchiveUri()).isEqualTo(uri);
}
@Test
void skipsMissingRawArchiveAndContinues() throws Exception {
String missingKey = "2026/06/22/GB32960/LNVFC000000000001/missing.bin";
@@ -168,9 +120,9 @@ class Gb32960DecodedFrameServiceTest {
Files.write(rawFile, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of(
record("missing", "archive://" + missingKey),
record("available", "archive://" + availableKey))),
new CapturingTdengineReader(List.of(
rawRow("missing", "archive://" + missingKey),
rawRow("available", "archive://" + availableKey))),
decoder(),
archiveRoot,
null);
@@ -178,7 +130,7 @@ class Gb32960DecodedFrameServiceTest {
List<Gb32960DecodedFrameService.DecodedFrame> frames = service.query(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
null,
null);
@@ -189,39 +141,10 @@ class Gb32960DecodedFrameServiceTest {
.containsExactly("VEHICLE", "GD_FC_DCDC");
}
@Test
void queryRequestsOnlyRawArchiveRecordsForOneVin() throws Exception {
String key = "2026/06/22/GB32960/LNVFC000000000001/raw-vin.bin";
writeArchive(key, realtimeFrame());
CapturingStore store = new CapturingStore(List.of(
businessRecord("business-newer", "LNVFC000000000001"),
record("raw-target", "archive://" + key),
record("raw-other-vin", "archive://2026/06/22/GB32960/OTHER/raw.bin", "OTHER000000000000")));
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
store,
decoder(),
archiveRoot,
null);
List<Gb32960DecodedFrameService.DecodedFrame> frames = service.query(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
EventFileQuery.Order.DESC,
1,
"LNVFC000000000001",
null);
assertThat(store.lastQuery.eventType()).isEqualTo("RAW_ARCHIVE");
assertThat(store.lastQuery.vin()).isEqualTo("LNVFC000000000001");
assertThat(frames).hasSize(1);
assertThat(frames.getFirst().eventId()).isEqualTo("raw-target");
}
@Test
void snapshotsRequireVinSoSegmentsAreMergedOnlyWithinOneVehicle() {
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of()),
(TdengineHistoryReader) null,
decoder(),
archiveRoot,
null);
@@ -229,7 +152,7 @@ class Gb32960DecodedFrameServiceTest {
assertThatThrownBy(() -> service.snapshots(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
null,
null))
@@ -248,10 +171,10 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key3, stackFrame(8, 2, false));
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of(
record("segment-3", "archive://" + key3),
record("segment-2", "archive://" + key2),
record("segment-1", "archive://" + key1))),
new CapturingTdengineReader(List.of(
rawRow("segment-3", "archive://" + key3),
rawRow("segment-2", "archive://" + key2),
rawRow("segment-1", "archive://" + key1))),
decoder(),
archiveRoot,
null);
@@ -259,7 +182,7 @@ class Gb32960DecodedFrameServiceTest {
List<Gb32960DecodedFrameService.LogicalSnapshot> snapshots = service.snapshots(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
vin,
null);
@@ -328,7 +251,7 @@ class Gb32960DecodedFrameServiceTest {
List<Gb32960DecodedFrameService.LogicalSnapshot> snapshots = service.snapshots(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
vin,
null);
@@ -376,7 +299,7 @@ class Gb32960DecodedFrameServiceTest {
List<Gb32960DecodedFrameService.DecodedFrame> frames = service.query(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
null,
"Hyundai");
@@ -425,7 +348,7 @@ class Gb32960DecodedFrameServiceTest {
Gb32960DecodedFrameService.DecodedFramePage page = service.queryPage(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
EventFileQuery.Order.ASC,
HistoryQueryOrder.ASC,
10,
vin,
null,
@@ -449,7 +372,7 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, stackFrame(1, 4, true));
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of(record("partial", "archive://" + key))),
new CapturingTdengineReader(List.of(rawRow("partial", "archive://" + key))),
decoder(),
archiveRoot,
null);
@@ -457,7 +380,7 @@ class Gb32960DecodedFrameServiceTest {
List<Gb32960DecodedFrameService.LogicalSnapshot> snapshots = service.snapshots(
LocalDate.parse("2026-06-22"),
LocalDate.parse("2026-06-22"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
vin,
null);
@@ -488,14 +411,14 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of(record("selected", "archive://" + key))),
new CapturingTdengineReader(List.of(rawRow("selected", "archive://" + key))),
decoder(),
archiveRoot,
null);
List<Gb32960DecodedFrameService.FieldSnapshot> snapshots = service.snapshotFields(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
vin,
null,
@@ -515,14 +438,14 @@ class Gb32960DecodedFrameServiceTest {
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of(record("selected-csv", "archive://" + key))),
new CapturingTdengineReader(List.of(rawRow("selected-csv", "archive://" + key))),
decoder(),
archiveRoot,
null);
String csv = service.snapshotFieldsCsv(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
vin,
null,
@@ -536,14 +459,14 @@ class Gb32960DecodedFrameServiceTest {
@Test
void snapshotFieldsRejectsUnknownDictionaryFields() throws Exception {
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
new CapturingStore(List.of()),
(TdengineHistoryReader) null,
decoder(),
archiveRoot,
null);
assertThatThrownBy(() -> service.snapshotFields(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
EventFileQuery.Order.DESC,
HistoryQueryOrder.DESC,
10,
"LNVFC000000000001",
null,
@@ -562,34 +485,30 @@ class Gb32960DecodedFrameServiceTest {
return new Gb32960MessageDecoder(new Gb32960BodyParser(registry, selector));
}
private static EventFileRecord record(String eventId, String rawArchiveUri) {
return record(eventId, rawArchiveUri, "LNVFC000000000001");
private static TdengineRawFrameRow rawRow(String eventId, String rawArchiveUri) {
return rawRow(eventId, rawArchiveUri, "LNVFC000000000001");
}
private static EventFileRecord record(String eventId, String rawArchiveUri, String vin) {
return new EventFileRecord(
eventId,
ProtocolId.GB32960,
"RAW_ARCHIVE",
vin,
private static TdengineRawFrameRow rawRow(String eventId, String rawArchiveUri, String vin) {
return new TdengineRawFrameRow(
Instant.parse("2026-06-22T10:00:00Z"),
Instant.parse("2026-06-22T10:00:01Z"),
rawArchiveUri,
Map.of("platformAccount", "Hyundai"),
"{}");
}
private static EventFileRecord businessRecord(String eventId, String vin) {
return new EventFileRecord(
eventId,
ProtocolId.GB32960,
"REALTIME",
Instant.parse("2026-06-22T10:00:01Z"),
0x02,
0,
Instant.parse("2026-06-22T10:00:00Z"),
rawArchiveUri,
"sha256:test",
64,
"SUCCEEDED",
"",
"127.0.0.1:32960",
"{\"platformAccount\":\"Hyundai\"}",
"",
"GB32960",
vin,
Instant.parse("2026-06-22T10:00:03Z"),
Instant.parse("2026-06-22T10:00:04Z"),
"archive://business/" + eventId + ".bin",
Map.of("platformAccount", "Hyundai"),
"{}");
vin,
"");
}
private static byte[] realtimeFrame() {
@@ -739,38 +658,6 @@ class Gb32960DecodedFrameServiceTest {
os.write((int) (v & 0xFF));
}
private static final class CapturingStore implements EventFileStore {
private final List<EventFileRecord> records;
private EventFileQuery lastQuery;
private CapturingStore(List<EventFileRecord> records) {
this.records = records;
}
@Override
public void appendAll(List<EventFileRecord> records) {
throw new UnsupportedOperationException();
}
@Override
public List<EventFileRecord> query(EventFileQuery query) {
lastQuery = query;
return records.stream()
.filter(record -> query.vin() == null || query.vin().equals(record.vin()))
.filter(record -> query.eventType() == null || query.eventType().equals(record.eventType()))
.limit(query.limit())
.toList();
}
@Override
public EventFileRecord findByRawArchiveUri(String rawArchiveUri) {
return records.stream()
.filter(record -> record.rawArchiveUri().equals(rawArchiveUri))
.findFirst()
.orElse(null);
}
}
private static final class CapturingTdengineReader implements TdengineHistoryReader {
private final List<TdengineRawFrameRow> rows;
private final TdenginePageCursor nextCursor;

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

@@ -1,127 +0,0 @@
package com.lingniu.ingest.eventhistory;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.sink.kafka.proto.RawArchiveRef;
import com.lingniu.ingest.sink.kafka.proto.TelemetryField;
import com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class TelemetryEnvelopeRecordMapperTest {
private final ObjectMapper objectMapper = new ObjectMapper();
@Test
void mapsFullFieldEnvelopeToEventFileRecord() {
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
.setEventId("event-1")
.setTraceId("trace-1")
.setVin("VIN001")
.setSource("GB32960")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.putMetadata("protocolVersion", "V2016")
.setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
.setEventType("REALTIME")
.setRawArchiveUri("archive://raw/event-1.bin")
.addFields(TelemetryField.newBuilder()
.setKey("total_mileage_km")
.setValueType("DOUBLE")
.setValue("12345.6")
.setUnit("km")
.setQuality("GOOD")
.setSourcePath("event.realtime.totalMileageKm"))
.build())
.build();
EventFileRecord record = new TelemetryEnvelopeRecordMapper().toRecord(envelope);
assertThat(record.eventId()).isEqualTo("event-1");
assertThat(record.protocol()).isEqualTo(ProtocolId.GB32960);
assertThat(record.eventType()).isEqualTo("REALTIME");
assertThat(record.vin()).isEqualTo("VIN001");
assertThat(record.rawArchiveUri()).isEqualTo("archive://raw/event-1.bin");
assertThat(record.metadata()).containsEntry("protocolVersion", "V2016");
assertThat(record.payloadJson()).contains("\"fields\"");
assertThat(record.payloadJson()).contains("\"total_mileage_km\"");
}
@Test
void rejectsEnvelopeWithoutTelemetrySnapshot() {
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setEventId("event-1")
.setVin("VIN001")
.setSource("GB32960")
.setEventTimeMs(1)
.setIngestTimeMs(2)
.build();
assertThatThrownBy(() -> new TelemetryEnvelopeRecordMapper().toRecord(envelope))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("telemetry_snapshot");
}
@Test
void unknownEnvelopeSourceIsStoredAsUnknownProtocolWithOriginalSourceMetadata() {
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
.setEventId("event-future-1")
.setTraceId("trace-future-1")
.setVin("VIN-FUTURE-001")
.setSource("FUTURE_OEM")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.putMetadata("tenant", "ln")
.setTelemetrySnapshot(TelemetrySnapshot.newBuilder()
.setEventType("REALTIME")
.addFields(TelemetryField.newBuilder()
.setKey("speed_kmh")
.setValueType("DOUBLE")
.setValue("30.5")
.setQuality("GOOD"))
.build())
.build();
EventFileRecord record = new TelemetryEnvelopeRecordMapper().toRecord(envelope);
assertThat(record.protocol()).isEqualTo(ProtocolId.UNKNOWN);
assertThat(record.metadata())
.containsEntry("tenant", "ln")
.containsEntry("originalSource", "FUTURE_OEM");
}
@Test
void rawArchiveRecordStoresParsedJsonInPayloadOnly() throws Exception {
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
.setEventId("raw-808-1")
.setTraceId("trace-raw-1")
.setVin("VIN808")
.setSource("JT808")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.putMetadata("phone", "13307795425")
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://2026/06/30/JT808/VIN808/raw-808-1.bin")
.setSizeBytes(64)
.setParsedJson("{\"messageId\":\"0x0200\",\"body\":{\"speedKmh\":60.5}}")
.build())
.build();
EventFileRecord record = new TelemetryEnvelopeRecordMapper().toRecord(envelope);
assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
assertThat(record.metadata())
.containsEntry("phone", "13307795425")
.doesNotContainKey("rawArchiveParsedJson");
var payload = objectMapper.readTree(record.payloadJson());
assertThat(payload.path("parsed").path("messageId").asText()).isEqualTo("0x0200");
assertThat(payload.path("parsed").path("body").path("speedKmh").asDouble()).isEqualTo(60.5);
}
}

16
modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
View File

@@ -2,8 +2,6 @@ package com.lingniu.ingest.eventhistory.config;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.eventhistory.EventHistoryController;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
@@ -12,7 +10,6 @@ import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
@@ -30,7 +27,7 @@ class EventHistoryAutoConfigurationTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withBean(EnvelopeDeadLetterSink.class, () -> record -> {})
.withBean(EventFileStore.class, () -> mock(EventFileStore.class));
.withBean(TdengineHistoryWriter.class, () -> mock(TdengineHistoryWriter.class));
@Test
void createsHistoryBeansWhenEnabled() {
@@ -39,13 +36,11 @@ class EventHistoryAutoConfigurationTest {
"lingniu.ingest.event-history.enabled=true",
"lingniu.ingest.sink.kafka.consumer.enabled=true")
.run(context -> {
assertThat(context).hasSingleBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class))
.containsOnlyKeys(
"eventHistoryEnvelopeConsumerProcessor",
"eventHistoryRawEnvelopeConsumerProcessor");
assertThat(context).hasSingleBean(EventHistoryController.class);
assertThat(context).doesNotHaveBean(Gb32960DecodedFrameService.class);
assertThat(context).doesNotHaveBean(Gb32960FrameController.class);
});
@@ -54,9 +49,7 @@ class EventHistoryAutoConfigurationTest {
@Test
void backsOffWhenDisabled() {
contextRunner.run(context -> {
assertThat(context).doesNotHaveBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).doesNotHaveBean(EventHistoryEnvelopeIngestor.class);
assertThat(context).doesNotHaveBean(EventHistoryController.class);
});
}
@@ -75,7 +68,7 @@ class EventHistoryAutoConfigurationTest {
}
@Test
void createsGb32960FrameBeansWithTdengineReaderWhenEventFileStoreIsDisabled() {
void createsGb32960FrameBeansWithTdengineReader() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withPropertyValues(
@@ -85,22 +78,19 @@ class EventHistoryAutoConfigurationTest {
.withBean(Gb32960MessageDecoder.class, () -> mock(Gb32960MessageDecoder.class))
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
.run(context -> {
assertThat(context).doesNotHaveBean(EventFileStore.class);
assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class);
assertThat(context).hasSingleBean(Gb32960FrameController.class);
});
}
@Test
void createsHistoryConsumerWithTdengineWriterWhenEventFileStoreIsDisabled() {
void createsHistoryConsumerWithTdengineWriter() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withBean(EnvelopeDeadLetterSink.class, () -> record -> {})
.withBean(TdengineHistoryWriter.class, () -> mock(TdengineHistoryWriter.class))
.withPropertyValues("lingniu.ingest.event-history.enabled=true")
.run(context -> {
assertThat(context).doesNotHaveBean(EventFileStore.class);
assertThat(context).doesNotHaveBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class))
.containsOnlyKeys(