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 4a0dba6d..aeea25b0 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 @@ -11,6 +11,7 @@ import com.lingniu.ingest.protocol.gb32960.model.Gb32960Message; import com.lingniu.ingest.protocol.gb32960.model.InfoBlock; 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; @@ -121,6 +122,37 @@ public final class Gb32960DecodedFrameService { return queryFromTdengine(dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, frameLimit, vin, platformAccount); } + public DecodedFramePage queryPage(QueryTimeRange range, + EventFileQuery.Order 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); + } + List items = query( + range.dateFrom(), + range.dateTo(), + range.eventTimeFrom(), + range.eventTimeTo(), + order, + frameLimit, + vin, + platformAccount); + return new DecodedFramePage(items, null); + } + private List queryFromEventStore(LocalDate dateFrom, LocalDate dateTo, java.time.Instant eventTimeFrom, @@ -195,6 +227,41 @@ public final class Gb32960DecodedFrameService { return out; } + private DecodedFramePage queryPageFromTdengine(LocalDate dateFrom, + LocalDate dateTo, + java.time.Instant eventTimeFrom, + java.time.Instant eventTimeTo, + EventFileQuery.Order order, + int frameLimit, + String vin, + String platformAccount, + TdenginePageCursor cursor) throws IOException { + TdenginePage page = tdengineReader.queryRawFrames(new TdengineRawFrameQuery( + "GB32960", + vin.trim(), + fromInstant(dateFrom, eventTimeFrom), + toExclusiveInstant(dateTo, eventTimeTo), + tdengineOrder(order), + frameLimit, + cursor)); + List out = new ArrayList<>(page.items().size()); + LinkedHashSet seenUris = new LinkedHashSet<>(); + for (TdengineRawFrameRow row : page.items()) { + String rawArchiveUri = row.rawUri(); + if (rawArchiveUri == null || rawArchiveUri.isBlank() || !seenUris.add(rawArchiveUri)) { + continue; + } + DecodedFrame frame = decodeIfAvailable(row, platformAccount); + if (frame != null) { + out.add(frame); + } + } + CursorResponse next = page.nextCursor() + .map(value -> new CursorResponse(value.ts().toString(), value.tieBreaker())) + .orElse(null); + return new DecodedFramePage(out, next); + } + public List snapshots(QueryTimeRange range, EventFileQuery.Order order, int limit, @@ -641,6 +708,14 @@ public final class Gb32960DecodedFrameService { List> blocks) { } + public record CursorResponse(String ts, String id) { + } + + public record DecodedFramePage( + List items, + CursorResponse nextCursor) { + } + public record SourceFrame( String eventId, String ingestTime, 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 9f6e971a..a6528476 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,12 +1,13 @@ package com.lingniu.ingest.eventhistory; import com.lingniu.ingest.eventfilestore.EventFileQuery; +import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; 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.http.HttpStatus; 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; @@ -14,6 +15,7 @@ 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; /** @@ -68,6 +70,40 @@ public final class Gb32960FrameController { platformAccount); } + @GetMapping("/frames/page") + @Operation( + summary = "分页查询 GB32960 原始帧解析结果", + description = "按 VIN 和时间范围分页查询 TDengine raw_frames，并回读 archive 原始包即时解析。" + + "分页使用 cursorTs + cursorId，不使用 offset；适合生产热历史连续翻页。") + public Gb32960DecodedFrameService.DecodedFramePage framesPage( + @Parameter(description = "开始时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:48:00") + @RequestParam String dateFrom, + @Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-23T10:49:00") + @RequestParam String dateTo, + @Parameter(description = "排序方向。", example = "DESC") + @RequestParam(defaultValue = "DESC") EventFileQuery.Order order, + @Parameter(description = "返回帧数量上限，最大 1000。", example = "100") + @RequestParam(defaultValue = "100") int limit, + @Parameter(description = "车辆 VIN，必填；用于命中 TDengine raw child table。", required = true, example = "LB9A32A20P0LS1257") + @RequestParam String vin, + @Parameter(description = "平台账号；用于选择对应厂商扩展解析规则。", example = "Hyundai") + @RequestParam(required = false) String platformAccount, + @Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z") + @RequestParam(required = false) String cursorTs, + @Parameter(description = "上一页返回的 nextCursor.id。", example = "gb32960-frame-xxx") + @RequestParam(required = false) String cursorId) throws IOException { + if (vin == null || vin.isBlank()) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "vin is required for gb32960 frame page query"); + } + return service.queryPage( + QueryTimeRange.parse(dateFrom, dateTo), + order, + limit, + vin, + platformAccount, + cursor(cursorTs, cursorId)); + } + @GetMapping("/frame") @Operation( summary = "按 rawArchiveUri 查询单个 GB32960 原始帧", @@ -80,6 +116,18 @@ public final class Gb32960FrameController { return service.findByRawArchiveUri(rawArchiveUri, platformAccount); } + 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()); + } + @GetMapping("/snapshots") @Operation( summary = "查询 GB32960 完整业务快照", 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 e5f9d07e..8c3440a5 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 @@ -18,6 +18,8 @@ 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; @@ -339,6 +341,58 @@ class Gb32960DecodedFrameServiceTest { .containsExactly("VEHICLE", "GD_FC_DCDC"); } + @Test + void framePageUsesTdengineCursorAndReturnsNextCursor() throws Exception { + String vin = "LNVFC000000000001"; + String key = "2026/06/22/GB32960/" + vin + "/tdengine-page.bin"; + String uri = "archive://" + key; + writeArchive(key, realtimeFrame()); + TdenginePageCursor nextCursor = new TdenginePageCursor( + Instant.parse("2026-06-22T10:00:00Z"), "tdengine-page"); + CapturingTdengineReader reader = new CapturingTdengineReader(List.of(new TdengineRawFrameRow( + Instant.parse("2026-06-22T10:00:00Z"), + "tdengine-page", + Instant.parse("2026-06-22T10:00:01Z"), + 0x02, + 0, + Instant.parse("2026-06-22T10:00:00Z"), + uri, + "sha256:test", + realtimeFrame().length, + "SUCCEEDED", + "", + "127.0.0.1:32960", + "{\"platformAccount\":\"Hyundai\"}", + "GB32960", + vin, + vin, + "")), nextCursor); + + Gb32960DecodedFrameService service = new Gb32960DecodedFrameService( + reader, + decoder(), + archiveRoot, + null); + + Gb32960DecodedFrameService.DecodedFramePage page = service.queryPage( + QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"), + EventFileQuery.Order.ASC, + 10, + vin, + null, + new TdenginePageCursor(Instant.parse("2026-06-22T09:59:00Z"), "previous")); + + assertThat(reader.lastQuery.protocol()).isEqualTo("GB32960"); + assertThat(reader.lastQuery.vehicleKey()).isEqualTo(vin); + assertThat(reader.lastQuery.order()).isEqualTo(TdengineQueryOrder.ASC); + assertThat(reader.lastQuery.cursor()).isEqualTo(new TdenginePageCursor( + Instant.parse("2026-06-22T09:59:00Z"), "previous")); + assertThat(page.items()).hasSize(1); + assertThat(page.items().getFirst().rawArchiveUri()).isEqualTo(uri); + assertThat(page.nextCursor()).isEqualTo(new Gb32960DecodedFrameService.CursorResponse( + "2026-06-22T10:00:00Z", "tdengine-page")); + } + @Test void snapshotsKeepMissingStackCellsAsNullsUpToCellCount() throws Exception { String vin = "LNVFC000000000001"; @@ -637,16 +691,22 @@ class Gb32960DecodedFrameServiceTest { private static final class CapturingTdengineReader implements TdengineHistoryReader { private final List rows; + private final TdenginePageCursor nextCursor; private TdengineRawFrameQuery lastQuery; private CapturingTdengineReader(List rows) { + this(rows, null); + } + + private CapturingTdengineReader(List rows, TdenginePageCursor nextCursor) { this.rows = rows; + this.nextCursor = nextCursor; } @Override public TdenginePage queryRawFrames(TdengineRawFrameQuery query) { lastQuery = query; - return new TdenginePage<>(rows, java.util.Optional.empty()); + return new TdenginePage<>(rows, java.util.Optional.ofNullable(nextCursor)); } @Override