From 68cd3d0d0682cb539de42a0638999c83d16fae9d Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 30 Jun 2026 10:22:05 +0800 Subject: [PATCH] feat: expose generic raw frame query --- .../VehicleHistoryAppCompositionTest.java | 2 + .../RawFrameHistoryController.java | 194 ++++++++++++++++++ .../config/EventHistoryAutoConfiguration.java | 8 + .../RawFrameHistoryControllerTest.java | 132 ++++++++++++ .../EventHistoryAutoConfigurationTest.java | 10 + 5 files changed, 346 insertions(+) create mode 100644 modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/RawFrameHistoryController.java create mode 100644 modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java 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 cc7d9983..0af94db1 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 @@ -11,6 +11,7 @@ import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration; import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController; import com.lingniu.ingest.eventhistory.LocationHistoryController; +import com.lingniu.ingest.eventhistory.RawFrameHistoryController; import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController; import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder; import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration; @@ -83,6 +84,7 @@ class VehicleHistoryAppCompositionTest { assertThat(context).hasSingleBean(Gb32960MessageDecoder.class); assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class); assertThat(context).hasSingleBean(LocationHistoryController.class); + assertThat(context).hasSingleBean(RawFrameHistoryController.class); assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class); assertThat(context).doesNotHaveBean(Gb32960FrameController.class); assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class); diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/RawFrameHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/RawFrameHistoryController.java new file mode 100644 index 00000000..11b5cb55 --- /dev/null +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/RawFrameHistoryController.java @@ -0,0 +1,194 @@ +package com.lingniu.ingest.eventhistory; + +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; +import com.lingniu.ingest.tdenginehistory.TdenginePage; +import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; +import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.tags.Tag; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.http.HttpStatus; +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 org.springframework.web.server.ResponseStatusException; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; +import java.util.Locale; + +@RestController +@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true") +@ConditionalOnBean(TdengineHistoryReader.class) +@RequestMapping("/api/event-history") +@Tag(name = "Raw Frame API", description = "原始帧索引分页查询接口。") +public final class RawFrameHistoryController { + + private final TdengineHistoryReader reader; + + public RawFrameHistoryController(TdengineHistoryReader reader) { + if (reader == null) { + throw new IllegalArgumentException("reader must not be null"); + } + this.reader = reader; + } + + @GetMapping("/raw-frames") + @Operation( + summary = "查询原始帧索引", + description = "按 protocol、vehicleKey、VIN、phone、messageId、parseStatus 和时间范围查询 TDengine raw_frames。" + + "返回 metadataJson 与 rawUri，分页使用 cursorTs + cursorId。") + public RawFramePageResponse rawFrames( + @Parameter(description = "协议类型，例如 GB32960、JT808、MQTT_YUTONG、XINDA_PUSH。", example = "JT808") + @RequestParam String protocol, + @Parameter(description = "内部车辆键；传入后精确过滤 vehicle_key。", example = "jt808:g7gps") + @RequestParam(required = false) String vehicleKey, + @Parameter(description = "VIN。已完成身份反写的 raw 帧可用该字段查询。", example = "LB9A32A24P0LS1270") + @RequestParam(required = false) String vin, + @Parameter(description = "终端手机号或终端标识，主要用于 JT808。", example = "g7gps") + @RequestParam(required = false) String phone, + @Parameter(description = "消息 ID，支持十进制或 0x 十六进制，例如 512 或 0x0200。", example = "0x0200") + @RequestParam(required = false) String messageId, + @Parameter(description = "解析状态，例如 SUCCEEDED、FAILED、NOT_PARSED。", example = "SUCCEEDED") + @RequestParam(required = false) String parseStatus, + @Parameter(description = "开始时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:00:00") + @RequestParam String dateFrom, + @Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:10:00") + @RequestParam String dateTo, + @Parameter(description = "排序方向。", example = "DESC") + @RequestParam(defaultValue = "DESC") TdengineQueryOrder order, + @Parameter(description = "返回原始帧数量上限，最大 1000。", example = "100") + @RequestParam(defaultValue = "100") int limit, + @Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z") + @RequestParam(required = false) String cursorTs, + @Parameter(description = "上一页返回的 nextCursor.id。", example = "frame-xxx") + @RequestParam(required = false) String cursorId) throws IOException { + Integer parsedMessageId = parseMessageId(messageId); + if (allBlank(vehicleKey, vin, phone) && parsedMessageId == null && isBlank(parseStatus)) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, + "vehicleKey, vin, phone, messageId or parseStatus is required for raw frame query"); + } + QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo); + TdenginePage page = reader.queryRawFrames(new TdengineRawFrameQuery( + normalize(protocol).toUpperCase(Locale.ROOT), + normalize(vehicleKey), + normalize(vin), + normalize(phone), + parsedMessageId, + normalize(parseStatus).toUpperCase(Locale.ROOT), + range.eventTimeFrom(), + range.eventTimeTo().plusMillis(1), + order, + limit, + cursor(cursorTs, cursorId))); + return RawFramePageResponse.from(page); + } + + private static Integer parseMessageId(String messageId) { + String value = normalize(messageId); + if (value.isBlank()) { + return null; + } + try { + if (value.startsWith("0x") || value.startsWith("0X")) { + return Integer.parseInt(value.substring(2), 16); + } + return Integer.valueOf(value); + } catch (NumberFormatException ex) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "messageId must be decimal or 0x hex", ex); + } + } + + private static TdenginePageCursor cursor(String cursorTs, String cursorId) { + boolean hasTs = cursorTs != null && !cursorTs.isBlank(); + boolean hasId = cursorId != null && !cursorId.isBlank(); + if (!hasTs && !hasId) { + return null; + } + if (!hasTs || !hasId) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "cursorTs and cursorId must be provided together"); + } + return new TdenginePageCursor(Instant.parse(cursorTs.trim()), cursorId.trim()); + } + + private static boolean allBlank(String... values) { + for (String value : values) { + if (!isBlank(value)) { + return false; + } + } + return true; + } + + private static boolean isBlank(String value) { + return value == null || value.isBlank(); + } + + private static String normalize(String value) { + return value == null ? "" : value.trim(); + } + + public record RawFramePageResponse( + List items, + CursorResponse nextCursor) { + + private static RawFramePageResponse from(TdenginePage page) { + CursorResponse next = page.nextCursor() + .map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker())) + .orElse(null); + return new RawFramePageResponse( + page.items().stream().map(RawFrameResponse::from).toList(), + next); + } + } + + public record CursorResponse(String ts, String id) { + } + + public record RawFrameResponse( + String eventTime, + String receivedAt, + String frameId, + int messageId, + String messageIdHex, + int subType, + String rawUri, + String checksum, + long rawSizeBytes, + String parseStatus, + String parseError, + String peer, + String metadataJson, + String protocol, + String vehicleKey, + String vin, + String phone) { + + private static RawFrameResponse from(TdengineRawFrameRow row) { + return new RawFrameResponse( + row.ts().toString(), + row.receivedAt().toString(), + row.frameId(), + row.messageId(), + "0x%04X".formatted(row.messageId()), + row.subType(), + row.rawUri(), + row.checksum(), + row.rawSizeBytes(), + row.parseStatus(), + row.parseError(), + row.peer(), + row.metadataJson(), + row.protocol(), + row.vehicleKey(), + row.vin(), + row.phone()); + } + } +} 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 efd8fb4c..484964a2 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 @@ -10,6 +10,7 @@ import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController; 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; @@ -109,6 +110,13 @@ public class EventHistoryAutoConfiguration { return new LocationHistoryController(reader); } + @Bean + @ConditionalOnBean(TdengineHistoryReader.class) + @ConditionalOnMissingBean + public RawFrameHistoryController rawFrameHistoryController(TdengineHistoryReader reader) { + return new RawFrameHistoryController(reader); + } + @Bean @ConditionalOnBean(TdengineHistoryReader.class) @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true") diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java new file mode 100644 index 00000000..cc9231d4 --- /dev/null +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/RawFrameHistoryControllerTest.java @@ -0,0 +1,132 @@ +package com.lingniu.ingest.eventhistory; + +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; +import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery; +import com.lingniu.ingest.tdenginehistory.TdengineLocationRow; +import com.lingniu.ingest.tdenginehistory.TdenginePage; +import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; +import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; +import org.springframework.web.server.ResponseStatusException; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class RawFrameHistoryControllerTest { + + @Test + void queriesRawFramesByProtocolHexMessageIdAndCursor() throws Exception { + CapturingReader reader = new CapturingReader(new TdenginePage<>( + List.of(row("frame-1", "2026-06-29T12:00:01Z")), + Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T12:00:01Z"), "frame-1")))); + RawFrameHistoryController controller = new RawFrameHistoryController(reader); + + RawFrameHistoryController.RawFramePageResponse response = controller.rawFrames( + "xinda_push", + null, + "LB9A32A24P0LS1270", + null, + "0x0200", + "not_parsed", + "2026-06-29T00:00:00+08:00", + "2026-06-29T23:59:59+08:00", + TdengineQueryOrder.DESC, + 20, + "2026-06-29T12:00:00Z", + "frame-0"); + + assertThat(reader.query.protocol()).isEqualTo("XINDA_PUSH"); + assertThat(reader.query.vin()).isEqualTo("LB9A32A24P0LS1270"); + assertThat(reader.query.messageId()).isEqualTo(0x0200); + assertThat(reader.query.parseStatus()).isEqualTo("NOT_PARSED"); + assertThat(reader.query.cursor()).isEqualTo(new TdenginePageCursor( + Instant.parse("2026-06-29T12:00:00Z"), "frame-0")); + assertThat(response.items()).hasSize(1); + RawFrameHistoryController.RawFrameResponse item = response.items().getFirst(); + assertThat(item.messageId()).isEqualTo(0x0200); + assertThat(item.messageIdHex()).isEqualTo("0x0200"); + assertThat(item.rawUri()).contains("XINDA_PUSH"); + assertThat(item.metadataJson()).contains("rawArchiveUri"); + assertThat(response.nextCursor()).isEqualTo(new RawFrameHistoryController.CursorResponse( + "2026-06-29T12:00:01Z", "frame-1")); + } + + @Test + void rejectsRawFrameQueryWithoutAnyNarrowingFilter() { + RawFrameHistoryController controller = new RawFrameHistoryController( + new CapturingReader(new TdenginePage<>(List.of(), Optional.empty()))); + + assertThatThrownBy(() -> controller.rawFrames( + "JT808", + null, + null, + null, + null, + null, + "2026-06-29", + "2026-06-30", + TdengineQueryOrder.DESC, + 20, + null, + null)) + .isInstanceOfSatisfying(ResponseStatusException.class, ex -> + assertThat(ex.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST)); + } + + private static TdengineRawFrameRow row(String frameId, String ts) { + Instant instant = Instant.parse(ts); + return new TdengineRawFrameRow( + instant, + frameId, + instant.plusMillis(500), + 0x0200, + 0, + instant, + "archive://2026/06/29/XINDA_PUSH/unknown/" + frameId + ".bin", + "sha256:" + frameId, + 72, + "NOT_PARSED", + "", + "xinda", + "{\"rawArchiveUri\":\"archive://2026/06/29/XINDA_PUSH/unknown/" + frameId + ".bin\"}", + "XINDA_PUSH", + "LB9A32A24P0LS1270", + "LB9A32A24P0LS1270", + ""); + } + + private static final class CapturingReader implements TdengineHistoryReader { + private final TdenginePage response; + private TdengineRawFrameQuery query; + + private CapturingReader(TdenginePage response) { + this.response = response; + } + + @Override + public TdenginePage queryRawFrames(TdengineRawFrameQuery query) throws IOException { + this.query = query; + return response; + } + + @Override + public TdenginePage queryLocations(TdengineLocationQuery query) { + throw new UnsupportedOperationException(); + } + + @Override + public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) { + throw new UnsupportedOperationException(); + } + } +} 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 24bc4030..9bfbe7de 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 @@ -10,6 +10,7 @@ import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController; 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; @@ -104,12 +105,21 @@ class EventHistoryAutoConfigurationTest { .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class)) .run(context -> { assertThat(context).hasSingleBean(LocationHistoryController.class); + assertThat(context).hasSingleBean(RawFrameHistoryController.class); assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class); assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class); assertThat(context).doesNotHaveBean(Jt808RawFrameHistoryController.class); }); } + @Test + void createsGenericRawFrameHistoryControllerWhenTdengineReaderExists() { + contextRunner + .withPropertyValues("lingniu.ingest.event-history.enabled=true") + .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class)) + .run(context -> assertThat(context).hasSingleBean(RawFrameHistoryController.class)); + } + @Test void createsJt808RawFrameHistoryControllerWhenTdengineReaderExists() { contextRunner