From 986fc2fac37631a7edc88ef13c712535b71cd5c0 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 11:18:02 +0800 Subject: [PATCH] refactor: remove event file store contracts --- docs/module-data-flow.html | 14 +- .../historyapp/MavenModuleProfileTest.java | 36 ++++ .../VehicleHistoryAppCompositionTest.java | 75 ------- .../ingest/api/event/VehicleEvent.java | 4 +- .../ingest/api/history/EventFileQuery.java | 93 -------- .../ingest/api/history/EventFileRecord.java | 41 ---- .../ingest/api/history/EventFileStore.java | 26 --- .../eventhistory/EventHistoryController.java | 122 ----------- .../EventHistoryEnvelopeIngestor.java | 64 +----- .../Gb32960DecodedFrameService.java | 181 ++-------------- .../eventhistory/Gb32960FrameController.java | 13 +- .../eventhistory/HistoryQueryOrder.java | 6 + .../TelemetryEnvelopeRecordMapper.java | 173 --------------- .../config/EventHistoryAutoConfiguration.java | 35 +-- .../eventhistory/ApiExceptionHandlerTest.java | 17 +- .../EventHistoryControllerTest.java | 82 ------- .../EventHistoryEnvelopeIngestorTest.java | 138 +----------- .../Gb32960DecodedFrameServiceTest.java | 203 ++++-------------- .../TelemetryEnvelopeRecordMapperTest.java | 127 ----------- .../EventHistoryAutoConfigurationTest.java | 16 +- 20 files changed, 151 insertions(+), 1315 deletions(-) delete mode 100644 modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java delete mode 100644 modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java delete mode 100644 modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java delete mode 100644 modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java create mode 100644 modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java delete mode 100644 modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java delete mode 100644 modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java delete mode 100644 modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapperTest.java diff --git a/docs/module-data-flow.html b/docs/module-data-flow.html index 32de4e8c..104eb444 100644 --- a/docs/module-data-flow.html +++ b/docs/module-data-flow.html @@ -317,7 +317,7 @@ flowchart TB end subgraph sink["输出与明细存储 modules/sinks"] - mq["sink-kafka
VehicleEvent 到 Protobuf Envelope 到 Kafka"] + kafka["sink-kafka
VehicleEvent 到 Protobuf Envelope 到 Kafka"] archive["sink-archive
RawArchive 到本地/S3/OSS 冷存"] tdengineStore["tdengine-history-store
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; diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java index 62b0632b..62afc740 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java @@ -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 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++) { diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java index fa4ab4d7..785a920c 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java @@ -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 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 rawFrames = new ArrayList<>(); - - @Override - public void appendRawFrames(List rows) { - rawFrames.addAll(rows); - } - - @Override - public void appendLocations(List rows) { - } - - @Override - public void appendTelemetryFields(List rows) { - } - } } diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java index 26bdaf88..d6ee000e 100644 --- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java +++ b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java @@ -144,8 +144,8 @@ public sealed interface VehicleEvent * 原始报文归档事件：每条成功解码的入站帧由 Dispatcher 产出一条，携带原始字节和归档索引信息。 * 32960 历史全字段查询依赖它对应的 {@code archive://...} 文件能够被回读。 * - *

当前 {@code KafkaEventSink} 默认不发送本类型，{@code EventFileStoreSink} 只保存索引不保存 - * bytes；因此启用新 raw 落盘/转发实现时，必须同时保证 {@link RawArchiveKeys} 的 key 规则一致。 + *

当前 {@code KafkaEventSink} 会发送本类型的索引信息，RAW bytes 由 archive 落盘保存； + * 因此启用 raw 落盘/转发实现时，必须同时保证 {@link RawArchiveKeys} 的 key 规则一致。 * * @param command 协议主命令码（如 32960 0x02/0x03 等），冗余在事件里便于按命令分片归档 * @param infoType 协议子类型（可为 0），同上 diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java deleted file mode 100644 index 71233eb2..00000000 --- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileQuery.java +++ /dev/null @@ -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); - } -} diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java deleted file mode 100644 index 1f93cf64..00000000 --- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileRecord.java +++ /dev/null @@ -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 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; - } -} diff --git a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java b/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java deleted file mode 100644 index 4864d1c8..00000000 --- a/modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/history/EventFileStore.java +++ /dev/null @@ -1,26 +0,0 @@ -package com.lingniu.ingest.api.history; - -import java.io.IOException; -import java.util.List; - -/** - * Historical event index contract. - * - *

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 records) throws IOException; - - List query(EventFileQuery query) throws IOException; - - default EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException { - return null; - } -} diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java deleted file mode 100644 index 6ff8f016..00000000 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryController.java +++ /dev/null @@ -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; - -/** - * 通用历史事件查询接口。 - * - *

这是跨协议的低层记录查询，返回的是 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 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 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 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()); - } - } -} diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java index 00ac451f..c3ca06d8 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java @@ -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. * *

当前 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 envelopes) throws IOException { - if (tdengineWriter == null) { - return; - } var rawFrames = envelopes.stream() .flatMap(envelope -> TdengineEnvelopeRows.rawFrame(envelope).stream()) .toList(); diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java index 77370116..562f21db 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java @@ -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 历史帧查询和业务快照组装服务。 * - *

历史库里保存的是 RAW_ARCHIVE 索引，不保存完整解码后的宽表。查询时先从 - * {@link EventFileStore} 找到 archive:// URI，再读取本地 RAW .bin，用当前协议解析器即时解码。 + *

TDengine raw_frames 保存 archive:// URI。查询时先从 TDengine 找到 RAW 索引， + * 再读取本地 RAW .bin，用当前协议解析器即时解码。 * 这种设计让写入路径保持轻量，也允许后续修复解析器后直接重放历史 RAW 得到新字段。 * *

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_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 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 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 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 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 records = store.query(new EventFileQuery( - ProtocolId.GB32960, - dateFrom, - dateTo, - eventTimeFrom, - eventTimeTo, - order == null ? EventFileQuery.Order.DESC : order, - frameLimit, - vin, - RAW_ARCHIVE_EVENT_TYPE)); - - List out = new ArrayList<>(); - LinkedHashSet 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 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 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 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 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> 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) { diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java index 951a4f67..9daf98a9 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java @@ -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; * *

这个 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") diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java new file mode 100644 index 00000000..8b41219a --- /dev/null +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/HistoryQueryOrder.java @@ -0,0 +1,6 @@ +package com.lingniu.ingest.eventhistory; + +public enum HistoryQueryOrder { + ASC, + DESC +} diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java deleted file mode 100644 index 2c22d708..00000000 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java +++ /dev/null @@ -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. - * - *

这是跨协议 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 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 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 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 metadata(VehicleEnvelope envelope, ProtocolId protocol) { - Map 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 metadata, - String rawArchiveKey, - String rawArchiveUri) { - Map 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 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()); - } - } -} diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java index 47b5df20..1b68e036 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java @@ -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; *

该模块有两类入口： *

    *
  • TDengine HTTP 查询入口：查询位置、RAW 帧和按需解码的 GB32960 帧。 - *
  • 兼容 EventFileStore 入口：仅在旧索引显式启用时创建旧记录查询和旧写入路径。 + *
  • Kafka consumer 入口：消费 envelope 并写入 TDengine 历史表。 *
* *

当前生产 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 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 store, - ObjectProvider reader, + public Gb32960DecodedFrameService gb32960DecodedFrameService(ObjectProvider 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 diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java index 597957ee..9e0adbf1 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/ApiExceptionHandlerTest.java @@ -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 records) { - } - - @Override - public List query(com.lingniu.ingest.api.history.EventFileQuery query) { - return List.of(); - } - }; return new Gb32960DecodedFrameService( - store, + (TdengineHistoryReader) null, mock(Gb32960MessageDecoder.class), archiveRoot, null); diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java deleted file mode 100644 index b4a4794c..00000000 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryControllerTest.java +++ /dev/null @@ -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 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 records; - private EventFileQuery lastQuery; - - private CapturingStore(List records) { - this.records = records; - } - - @Override - public void appendAll(List records) { - throw new UnsupportedOperationException(); - } - - @Override - public List query(EventFileQuery query) throws IOException { - this.lastQuery = query; - return records; - } - } -} diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java index 084f688b..fcd612be 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java @@ -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 records = new ArrayList<>(); - private int appendAllCalls; - - @Override - public void appendAll(List records) { - appendAllCalls++; - this.records.addAll(records); - } - - @Override - public List query(EventFileQuery query) throws IOException { - return List.copyOf(records); - } - } - private static final class CapturingTdengineWriter implements TdengineHistoryWriter { private final List rawFrames = new ArrayList<>(); private final List locations = new ArrayList<>(); diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java index f0590cb8..600df17f 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java @@ -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 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 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 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 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 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 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 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 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 records; - private EventFileQuery lastQuery; - - private CapturingStore(List records) { - this.records = records; - } - - @Override - public void appendAll(List records) { - throw new UnsupportedOperationException(); - } - - @Override - public List 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 rows; private final TdenginePageCursor nextCursor; diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapperTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapperTest.java deleted file mode 100644 index 96538c3a..00000000 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapperTest.java +++ /dev/null @@ -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); - } -} diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java index 40b80ca3..e5fcb08e 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java @@ -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(