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

feat: add gb32960 frame cursor pagination

Browse Source
This commit is contained in:
lingniu
2026-06-29 21:00:43 +08:00
parent 12ac1dab3b
commit bdc455dcf7
3 changed files with 185 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<DecodedFrame> items = query(
range.dateFrom(),
range.dateTo(),
range.eventTimeFrom(),
range.eventTimeTo(),
order,
frameLimit,
vin,
platformAccount);
return new DecodedFramePage(items, null);
}
private List<DecodedFrame> 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<TdengineRawFrameRow> page = tdengineReader.queryRawFrames(new TdengineRawFrameQuery(
"GB32960",
vin.trim(),
fromInstant(dateFrom, eventTimeFrom),
toExclusiveInstant(dateTo, eventTimeTo),
tdengineOrder(order),
frameLimit,
cursor));
List<DecodedFrame> out = new ArrayList<>(page.items().size());
LinkedHashSet<String> 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<LogicalSnapshot> snapshots(QueryTimeRange range,
EventFileQuery.Order order,
int limit,
@@ -641,6 +708,14 @@ public final class Gb32960DecodedFrameService {
List<Map<String, Object>> blocks) {
}
public record CursorResponse(String ts, String id) {
}
public record DecodedFramePage(
List<DecodedFrame> items,
CursorResponse nextCursor) {
}
public record SourceFrame(
String eventId,
String ingestTime,

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

@@ -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 完整业务快照",

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

@@ -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<TdengineRawFrameRow> rows;
private final TdenginePageCursor nextCursor;
private TdengineRawFrameQuery lastQuery;
private CapturingTdengineReader(List<TdengineRawFrameRow> rows) {
this(rows, null);
}
private CapturingTdengineReader(List<TdengineRawFrameRow> rows, TdenginePageCursor nextCursor) {
this.rows = rows;
this.nextCursor = nextCursor;
}
@Override
public TdenginePage<TdengineRawFrameRow> queryRawFrames(TdengineRawFrameQuery query) {
lastQuery = query;
return new TdenginePage<>(rows, java.util.Optional.empty());
return new TdenginePage<>(rows, java.util.Optional.ofNullable(nextCursor));
}
@Override