feat: add jt808 tdengine location query api
This commit is contained in:
@@ -0,0 +1,149 @@
|
|||||||
|
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 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;
|
||||||
|
|
||||||
|
@RestController
|
||||||
|
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
|
||||||
|
@ConditionalOnBean(TdengineHistoryReader.class)
|
||||||
|
@RequestMapping("/api/event-history/jt808")
|
||||||
|
@Tag(name = "jt-808-location-controller", description = "JT808 位置历史分页查询接口。")
|
||||||
|
public final class Jt808LocationHistoryController {
|
||||||
|
|
||||||
|
private final TdengineHistoryReader reader;
|
||||||
|
|
||||||
|
public Jt808LocationHistoryController(TdengineHistoryReader reader) {
|
||||||
|
if (reader == null) {
|
||||||
|
throw new IllegalArgumentException("reader must not be null");
|
||||||
|
}
|
||||||
|
this.reader = reader;
|
||||||
|
}
|
||||||
|
|
||||||
|
@GetMapping("/locations")
|
||||||
|
@Operation(
|
||||||
|
summary = "查询 JT808 位置历史",
|
||||||
|
description = "按终端手机号和时间范围查询 TDengine 中的 JT808 位置点。分页使用 cursorTs + cursorId,不使用 offset,适合高并发历史轨迹查询。")
|
||||||
|
public LocationPageResponse locations(
|
||||||
|
@Parameter(description = "JT808 终端手机号或终端标识;内部会映射到 jt808:<phone> 车辆键。", required = true, example = "g7gps")
|
||||||
|
@RequestParam String phone,
|
||||||
|
@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 = "jt808-location-xxx")
|
||||||
|
@RequestParam(required = false) String cursorId) throws IOException {
|
||||||
|
if (phone == null || phone.isBlank()) {
|
||||||
|
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "phone is required for jt808 location query");
|
||||||
|
}
|
||||||
|
QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
|
||||||
|
TdenginePage<TdengineLocationRow> page = reader.queryLocations(new TdengineLocationQuery(
|
||||||
|
"JT808",
|
||||||
|
vehicleKey(phone),
|
||||||
|
range.eventTimeFrom(),
|
||||||
|
range.eventTimeTo().plusMillis(1),
|
||||||
|
order,
|
||||||
|
limit,
|
||||||
|
cursor(cursorTs, cursorId)));
|
||||||
|
return LocationPageResponse.from(page);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String vehicleKey(String phone) {
|
||||||
|
String value = phone.trim();
|
||||||
|
return value.startsWith("jt808:") ? value : "jt808:" + value;
|
||||||
|
}
|
||||||
|
|
||||||
|
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());
|
||||||
|
}
|
||||||
|
|
||||||
|
public record LocationPageResponse(
|
||||||
|
List<LocationResponse> items,
|
||||||
|
CursorResponse nextCursor) {
|
||||||
|
|
||||||
|
private static LocationPageResponse from(TdenginePage<TdengineLocationRow> page) {
|
||||||
|
CursorResponse next = page.nextCursor()
|
||||||
|
.map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker()))
|
||||||
|
.orElse(null);
|
||||||
|
return new LocationPageResponse(
|
||||||
|
page.items().stream().map(LocationResponse::from).toList(),
|
||||||
|
next);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public record CursorResponse(String ts, String id) {
|
||||||
|
}
|
||||||
|
|
||||||
|
public record LocationResponse(
|
||||||
|
String eventTime,
|
||||||
|
String receivedAt,
|
||||||
|
String factId,
|
||||||
|
String frameId,
|
||||||
|
String phone,
|
||||||
|
String vin,
|
||||||
|
String vehicleKey,
|
||||||
|
double longitude,
|
||||||
|
double latitude,
|
||||||
|
double altitudeM,
|
||||||
|
double speedKmh,
|
||||||
|
double directionDeg,
|
||||||
|
long alarmFlag,
|
||||||
|
long statusFlag,
|
||||||
|
Double totalMileageKm,
|
||||||
|
String rawUri,
|
||||||
|
String metadataJson) {
|
||||||
|
|
||||||
|
private static LocationResponse from(TdengineLocationRow row) {
|
||||||
|
return new LocationResponse(
|
||||||
|
row.ts().toString(),
|
||||||
|
row.receivedAt().toString(),
|
||||||
|
row.factId(),
|
||||||
|
row.frameId(),
|
||||||
|
row.phone(),
|
||||||
|
row.vin(),
|
||||||
|
row.vehicleKey(),
|
||||||
|
row.longitude(),
|
||||||
|
row.latitude(),
|
||||||
|
row.altitudeM(),
|
||||||
|
row.speedKmh(),
|
||||||
|
row.directionDeg(),
|
||||||
|
row.alarmFlag(),
|
||||||
|
row.statusFlag(),
|
||||||
|
row.totalMileageKm(),
|
||||||
|
row.rawUri(),
|
||||||
|
row.metadataJson());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -7,11 +7,13 @@ import com.lingniu.ingest.eventhistory.EventHistoryController;
|
|||||||
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
|
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
|
||||||
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
|
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
|
||||||
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
|
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
|
||||||
|
import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
|
||||||
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
|
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
|
||||||
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
|
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
|
||||||
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
|
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
|
||||||
import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration;
|
import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration;
|
||||||
import com.lingniu.ingest.sink.archive.config.SinkArchiveProperties;
|
import com.lingniu.ingest.sink.archive.config.SinkArchiveProperties;
|
||||||
|
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
|
||||||
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
|
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
|
||||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||||
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
||||||
@@ -89,6 +91,13 @@ public class EventHistoryAutoConfiguration {
|
|||||||
return new Gb32960FrameController(service);
|
return new Gb32960FrameController(service);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
@ConditionalOnBean(TdengineHistoryReader.class)
|
||||||
|
@ConditionalOnMissingBean
|
||||||
|
public Jt808LocationHistoryController jt808LocationHistoryController(TdengineHistoryReader reader) {
|
||||||
|
return new Jt808LocationHistoryController(reader);
|
||||||
|
}
|
||||||
|
|
||||||
static Path archiveRoot(String value) {
|
static Path archiveRoot(String value) {
|
||||||
if (value == null || value.isBlank()) {
|
if (value == null || value.isBlank()) {
|
||||||
return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive");
|
return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive");
|
||||||
|
|||||||
@@ -0,0 +1,113 @@
|
|||||||
|
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 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 Jt808LocationHistoryControllerTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void queriesLocationsByPhoneWithKeysetCursor() throws Exception {
|
||||||
|
CapturingReader reader = new CapturingReader(new TdenginePage<>(
|
||||||
|
List.of(row("fact-1", "2026-06-29T05:00:01Z")),
|
||||||
|
Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "fact-1"))));
|
||||||
|
Jt808LocationHistoryController controller = new Jt808LocationHistoryController(reader);
|
||||||
|
|
||||||
|
Jt808LocationHistoryController.LocationPageResponse response = controller.locations(
|
||||||
|
"g7gps",
|
||||||
|
"2026-06-29T13:00:00+08:00",
|
||||||
|
"2026-06-29T13:10:00+08:00",
|
||||||
|
TdengineQueryOrder.DESC,
|
||||||
|
50,
|
||||||
|
"2026-06-29T05:00:00Z",
|
||||||
|
"fact-0");
|
||||||
|
|
||||||
|
assertThat(reader.query.protocol()).isEqualTo("JT808");
|
||||||
|
assertThat(reader.query.vehicleKey()).isEqualTo("jt808:g7gps");
|
||||||
|
assertThat(reader.query.order()).isEqualTo(TdengineQueryOrder.DESC);
|
||||||
|
assertThat(reader.query.limit()).isEqualTo(50);
|
||||||
|
assertThat(reader.query.cursor()).isEqualTo(new TdenginePageCursor(
|
||||||
|
Instant.parse("2026-06-29T05:00:00Z"), "fact-0"));
|
||||||
|
assertThat(response.items())
|
||||||
|
.extracting(Jt808LocationHistoryController.LocationResponse::factId)
|
||||||
|
.containsExactly("fact-1");
|
||||||
|
assertThat(response.items().getFirst().longitude()).isEqualTo(113.12);
|
||||||
|
assertThat(response.nextCursor()).isEqualTo(new Jt808LocationHistoryController.CursorResponse(
|
||||||
|
"2026-06-29T05:00:01Z", "fact-1"));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void rejectsBlankPhone() {
|
||||||
|
Jt808LocationHistoryController controller = new Jt808LocationHistoryController(
|
||||||
|
new CapturingReader(new TdenginePage<>(List.of(), Optional.empty())));
|
||||||
|
|
||||||
|
assertThatThrownBy(() -> controller.locations(
|
||||||
|
" ",
|
||||||
|
"2026-06-29T13:00:00+08:00",
|
||||||
|
"2026-06-29T13:10:00+08:00",
|
||||||
|
TdengineQueryOrder.DESC,
|
||||||
|
50,
|
||||||
|
null,
|
||||||
|
null))
|
||||||
|
.isInstanceOfSatisfying(ResponseStatusException.class, ex ->
|
||||||
|
assertThat(ex.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static TdengineLocationRow row(String factId, String ts) {
|
||||||
|
Instant instant = Instant.parse(ts);
|
||||||
|
return new TdengineLocationRow(
|
||||||
|
instant,
|
||||||
|
factId,
|
||||||
|
"frame-1",
|
||||||
|
instant.plusMillis(500),
|
||||||
|
113.12,
|
||||||
|
23.45,
|
||||||
|
8.0,
|
||||||
|
42.5,
|
||||||
|
90.0,
|
||||||
|
1L,
|
||||||
|
3L,
|
||||||
|
null,
|
||||||
|
"archive://jt808/frame-1.bin",
|
||||||
|
"{\"messageId\":\"512\"}",
|
||||||
|
"JT808",
|
||||||
|
"jt808:g7gps",
|
||||||
|
"",
|
||||||
|
"g7gps");
|
||||||
|
}
|
||||||
|
|
||||||
|
private static final class CapturingReader implements TdengineHistoryReader {
|
||||||
|
private final TdenginePage<TdengineLocationRow> response;
|
||||||
|
private TdengineLocationQuery query;
|
||||||
|
|
||||||
|
private CapturingReader(TdenginePage<TdengineLocationRow> response) {
|
||||||
|
this.response = response;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public TdenginePage<com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow> queryRawFrames(
|
||||||
|
com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery query) {
|
||||||
|
throw new UnsupportedOperationException();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) throws IOException {
|
||||||
|
this.query = query;
|
||||||
|
return response;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -7,8 +7,10 @@ import com.lingniu.ingest.eventhistory.EventHistoryController;
|
|||||||
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
|
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
|
||||||
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
|
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
|
||||||
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
|
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
|
||||||
|
import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
|
||||||
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
|
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
|
||||||
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
|
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
|
||||||
|
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
|
||||||
import com.lingniu.ingest.sink.archive.config.SinkArchiveProperties;
|
import com.lingniu.ingest.sink.archive.config.SinkArchiveProperties;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||||
@@ -63,6 +65,14 @@ class EventHistoryAutoConfigurationTest {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void createsJt808LocationHistoryControllerWhenTdengineReaderExists() {
|
||||||
|
contextRunner
|
||||||
|
.withPropertyValues("lingniu.ingest.event-history.enabled=true")
|
||||||
|
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
|
||||||
|
.run(context -> assertThat(context).hasSingleBean(Jt808LocationHistoryController.class));
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
void parsesFileUriArchiveRoot() {
|
void parsesFileUriArchiveRoot() {
|
||||||
assertThat(EventHistoryAutoConfiguration.archiveRoot("file:///tmp/lingniu-test-archive"))
|
assertThat(EventHistoryAutoConfiguration.archiveRoot("file:///tmp/lingniu-test-archive"))
|
||||||
|
|||||||
Reference in New Issue
Block a user