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

refactor: simplify jt808 mileage stat pipeline

Browse Source
This commit is contained in:
lingniu
2026-07-01 02:04:58 +08:00
parent 0fc91f512c
commit 7081800b84
26 changed files with 92 additions and 975 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
modules/apps/vehicle-analytics-app/src/main/resources/application.yml
View File

@@ -80,7 +80,7 @@ lingniu:
enabled: ${VEHICLE_STAT_ENABLED:true}
zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai}
jt808:
enabled: ${VEHICLE_STAT_JT808_MILEAGE_ENABLED:false}
enabled: ${VEHICLE_STAT_JT808_MILEAGE_ENABLED:true}
state-store: ${VEHICLE_STAT_JT808_STATE_STORE:redis}
redis-key-prefix: ${VEHICLE_STAT_JT808_REDIS_KEY_PREFIX:vehicle:mileage:jt808:daily:}
state-ttl-days: ${VEHICLE_STAT_JT808_STATE_TTL_DAYS:3}

10
modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java
View File

@@ -6,14 +6,12 @@ import com.lingniu.ingest.sink.mq.KafkaEventSink;
import com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration;
import com.lingniu.ingest.vehiclestate.VehicleStateEnvelopeIngestor;
import com.lingniu.ingest.vehiclestate.config.VehicleStateAutoConfiguration;
import com.lingniu.ingest.vehiclestat.DailyVehicleStatService;
import com.lingniu.ingest.vehiclestat.JdbcVehicleStatMetricRepository;
import com.lingniu.ingest.vehiclestat.VehicleStatController;
import com.lingniu.ingest.vehiclestat.VehicleStatEnvelopeIngestor;
import com.lingniu.ingest.vehiclestat.VehicleStatEventProcessor;
import com.lingniu.ingest.vehiclestat.VehicleStatEventSink;
import com.lingniu.ingest.vehiclestat.VehicleStatRepository;
import com.lingniu.ingest.vehiclestat.config.VehicleStatAutoConfiguration;
import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
@@ -44,16 +42,16 @@ class VehicleAnalyticsAppCompositionTest {
"lingniu.ingest.sink.mq.consumer.enabled=false",
"lingniu.ingest.vehicle-state.enabled=false",
"lingniu.ingest.vehicle-stat.enabled=true",
"lingniu.ingest.vehicle-stat.jt808.enabled=true",
"lingniu.ingest.vehicle-stat.jt808.state-store=memory",
"lingniu.ingest.event-file-store.enabled=false",
"lingniu.ingest.event-history.enabled=false",
"lingniu.ingest.gb32960.enabled=false")
.run(context -> {
assertThat(context).hasSingleBean(VehicleStatRepository.class);
assertThat(context).hasSingleBean(JdbcVehicleStatMetricRepository.class);
assertThat(context).hasSingleBean(DailyVehicleStatService.class);
assertThat(context).hasSingleBean(VehicleStatEventProcessor.class);
assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class);
assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class);
assertThat(context).hasSingleBean(VehicleStatEventSink.class);
assertThat(context).hasSingleBean(VehicleStatController.class);
assertThat(context).hasSingleBean(KafkaEventSink.class);
assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);

3
modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java
View File

@@ -25,6 +25,9 @@ class VehicleAnalyticsAppDefaultsTest {
.containsEntry("lingniu.ingest.event-file-store.enabled", false)
.containsEntry("lingniu.ingest.event-history.enabled", false)
.containsEntry("lingniu.ingest.vehicle-stat.enabled", "${VEHICLE_STAT_ENABLED:true}")
.containsEntry(
"lingniu.ingest.vehicle-stat.jt808.enabled",
"${VEHICLE_STAT_JT808_MILEAGE_ENABLED:true}")
.containsEntry("lingniu.ingest.vehicle-state.enabled", "${VEHICLE_STATE_ENABLED:false}")
.containsEntry("lingniu.ingest.sink.mq.consumer.enabled", "${KAFKA_CONSUMER_ENABLED:true}")
.containsEntry(

75
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/DailyMileageCalculator.java
View File

@@ -1,75 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.OptionalDouble;
public final class DailyMileageCalculator {
private final ZoneId zoneId;
public DailyMileageCalculator(ZoneId zoneId) {
if (zoneId == null) {
throw new IllegalArgumentException("zoneId must not be null");
}
this.zoneId = zoneId;
}
public OptionalDouble calculate(LocalDate statDate,
DailyMileageStrategy strategy,
List<MileagePoint> points) {
if (statDate == null) {
throw new IllegalArgumentException("statDate must not be null");
}
if (strategy == null) {
throw new IllegalArgumentException("strategy must not be null");
}
if (points == null || points.isEmpty()) {
return OptionalDouble.empty();
}
// 两种策略都只基于同一 VIN 的里程点计算，不能跨车辆 join；
// 上层 repository 负责按 VIN 过滤，避免大车队查询时扩大扫描面。
double value = switch (strategy) {
case CURRENT_LAST_MINUS_PREVIOUS_LAST -> currentLastMinusPreviousLast(statDate, points);
case DAY_MAX_MINUS_DAY_MIN -> dayMaxMinusDayMin(statDate, points);
};
return Double.isFinite(value) && value >= 0 ? OptionalDouble.of(value) : OptionalDouble.empty();
}
private double currentLastMinusPreviousLast(LocalDate statDate, List<MileagePoint> points) {
// 适用于累计里程单调递增的设备：当天最后一帧减前一日最后一帧。
Optional<MileagePoint> previousLast = points.stream()
.filter(point -> localDate(point).isBefore(statDate))
.max(Comparator.comparing(MileagePoint::eventTime));
Optional<MileagePoint> currentLast = points.stream()
.filter(point -> localDate(point).isEqual(statDate))
.max(Comparator.comparing(MileagePoint::eventTime));
if (previousLast.isEmpty() || currentLast.isEmpty()) {
return Double.NaN;
}
return currentLast.get().totalMileageKm() - previousLast.get().totalMileageKm();
}
private double dayMaxMinusDayMin(LocalDate statDate, List<MileagePoint> points) {
// 兜底策略：仅使用当天数据，适合缺少前一日最后点但当天点足够的场景。
List<MileagePoint> currentDay = points.stream()
.filter(point -> localDate(point).isEqual(statDate))
.toList();
if (currentDay.size() < 2) {
return Double.NaN;
}
double min = currentDay.stream().mapToDouble(MileagePoint::totalMileageKm).min().orElse(Double.NaN);
double max = currentDay.stream().mapToDouble(MileagePoint::totalMileageKm).max().orElse(Double.NaN);
return max - min;
}
private LocalDate localDate(MileagePoint point) {
return point.eventTime().atZone(zoneId).toLocalDate();
}
}

49
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/DailyVehicleStatService.java
View File

@@ -1,49 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import java.time.LocalDate;
import java.util.Optional;
import java.util.OptionalDouble;
public final class DailyVehicleStatService {
private final VehicleStatRepository repository;
private final VehicleStatRuleRepository ruleRepository;
private final DailyMileageCalculator mileageCalculator;
public DailyVehicleStatService(VehicleStatRepository repository,
VehicleStatRuleRepository ruleRepository,
DailyMileageCalculator mileageCalculator) {
if (repository == null) {
throw new IllegalArgumentException("repository must not be null");
}
if (ruleRepository == null) {
throw new IllegalArgumentException("ruleRepository must not be null");
}
if (mileageCalculator == null) {
throw new IllegalArgumentException("mileageCalculator must not be null");
}
this.repository = repository;
this.ruleRepository = ruleRepository;
this.mileageCalculator = mileageCalculator;
}
public Optional<VehicleDailyStatResult> calculateAndSave(String vin, LocalDate statDate) {
VehicleStatRule rule = ruleRepository.ruleFor(vin);
// 统计是从里程点二次计算出来的派生数据，不作为 32960 全字段历史查询的数据源。
OptionalDouble dailyMileage = mileageCalculator.calculate(
statDate,
rule.dailyMileageStrategy(),
repository.mileagePoints(vin, statDate));
if (dailyMileage.isEmpty()) {
return Optional.empty();
}
VehicleDailyStatResult result = new VehicleDailyStatResult(
vin,
statDate,
dailyMileage,
rule.dailyMileageStrategy());
repository.saveDailyStat(result);
return Optional.of(result);
}
}

126
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/FileVehicleStatRepository.java
View File

@@ -1,126 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.time.Instant;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.OptionalDouble;
public final class FileVehicleStatRepository implements VehicleStatRepository {
private final Path pointsFile;
private final Path dailyStatsFile;
public FileVehicleStatRepository(Path root) {
if (root == null) {
throw new IllegalArgumentException("root must not be null");
}
Path absoluteRoot = root.toAbsolutePath();
// 当前实现是轻量本地 TSV，适合开发和小规模派生统计；生产历史全字段查询走 DuckDB RAW 索引。
this.pointsFile = absoluteRoot.resolve("mileage-points.tsv");
this.dailyStatsFile = absoluteRoot.resolve("daily-stats.tsv");
}
@Override
public synchronized void appendMileagePoint(String vin, MileagePoint point) {
if (point == null) {
throw new IllegalArgumentException("point must not be null");
}
appendLine(pointsFile, clean(vin) + '\t' + point.eventTime() + '\t' + point.totalMileageKm());
}
@Override
public synchronized List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
String normalizedVin = clean(vin);
List<MileagePoint> out = new ArrayList<>();
for (String line : readLines(pointsFile)) {
String[] parts = line.split("\\t", -1);
if (parts.length != 3 || !normalizedVin.equals(parts[0])) {
continue;
}
try {
// statDate 由计算器按策略判断；这里返回该 VIN 的点，避免仓储层混入业务口径。
out.add(new MileagePoint(Instant.parse(parts[1]), Double.parseDouble(parts[2])));
} catch (RuntimeException ignored) {
// Ignore corrupt rows instead of making the whole statistics API unavailable.
}
}
return out;
}
@Override
public synchronized void saveDailyStat(VehicleDailyStatResult result) {
if (result == null) {
throw new IllegalArgumentException("result must not be null");
}
String mileage = result.dailyMileageKm().isPresent()
? Double.toString(result.dailyMileageKm().getAsDouble())
: "";
appendLine(dailyStatsFile,
clean(result.vin()) + '\t'
+ result.statDate() + '\t'
+ result.dailyMileageStrategy().name() + '\t'
+ mileage);
}
@Override
public synchronized Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
String normalizedVin = clean(vin);
VehicleDailyStatResult latest = null;
for (String line : readLines(dailyStatsFile)) {
String[] parts = line.split("\\t", -1);
if (parts.length != 4 || !normalizedVin.equals(parts[0])) {
continue;
}
try {
LocalDate rowDate = LocalDate.parse(parts[1]);
if (!rowDate.equals(statDate)) {
continue;
}
DailyMileageStrategy strategy = DailyMileageStrategy.valueOf(parts[2]);
OptionalDouble mileage = parts[3].isBlank()
? OptionalDouble.empty()
: OptionalDouble.of(Double.parseDouble(parts[3]));
latest = new VehicleDailyStatResult(normalizedVin, rowDate, mileage, strategy);
} catch (RuntimeException ignored) {
// Ignore corrupt rows; later valid rows can still provide the answer.
}
}
return Optional.ofNullable(latest);
}
private static void appendLine(Path file, String line) {
try {
Files.createDirectories(file.getParent());
Files.writeString(file, line + System.lineSeparator(), StandardCharsets.UTF_8,
StandardOpenOption.CREATE, StandardOpenOption.APPEND);
} catch (IOException e) {
throw new IllegalStateException("write vehicle stat file failed: " + file, e);
}
}
private static List<String> readLines(Path file) {
if (!Files.isRegularFile(file)) {
return List.of();
}
try {
return Files.readAllLines(file, StandardCharsets.UTF_8);
} catch (IOException e) {
throw new IllegalStateException("read vehicle stat file failed: " + file, e);
}
}
private static String clean(String value) {
if (value == null || value.isBlank()) {
throw new IllegalArgumentException("vin must not be blank");
}
// TSV 文件没有转义层，VIN 中的控制字符统一替换，防止破坏行结构。
return value.trim().replace('\t', '_').replace('\n', '_').replace('\r', '_');
}
}

10
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepository.java
View File

@@ -24,16 +24,6 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
ensureSchema();
}
@Override
public void appendMileagePoint(String vin, MileagePoint point) {
// Production metrics are written as final metric rows; raw source points stay in RAW/locations history.
}
@Override
public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return List.of();
}
@Override
public void saveDailyStat(VehicleDailyStatResult result) {
if (result == null || result.dailyMileageKm().isEmpty()) {

15
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/MileagePoint.java
View File

@@ -1,15 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import java.time.Instant;
public record MileagePoint(Instant eventTime, double totalMileageKm) {
public MileagePoint {
if (eventTime == null) {
throw new IllegalArgumentException("eventTime must not be null");
}
if (!Double.isFinite(totalMileageKm)) {
throw new IllegalArgumentException("totalMileageKm must be finite");
}
}
}

18
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatController.java
View File

@@ -17,33 +17,25 @@ import java.util.Map;
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat", name = "enabled", havingValue = "true")
@ConditionalOnBean(DailyVehicleStatService.class)
@ConditionalOnBean(VehicleStatRepository.class)
@RequestMapping(path = "/api/vehicle-stat", produces = MediaType.APPLICATION_JSON_VALUE)
public final class VehicleStatController {
private final VehicleStatRepository repository;
private final DailyVehicleStatService dailyStatService;
public VehicleStatController(VehicleStatRepository repository,
DailyVehicleStatService dailyStatService) {
public VehicleStatController(VehicleStatRepository repository) {
if (repository == null) {
throw new IllegalArgumentException("repository must not be null");
}
if (dailyStatService == null) {
throw new IllegalArgumentException("dailyStatService must not be null");
}
this.repository = repository;
this.dailyStatService = dailyStatService;
}
@GetMapping("/{vin}/daily")
public ResponseEntity<Map<String, Object>> daily(
@PathVariable String vin,
@RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date,
@RequestParam(defaultValue = "true") boolean calculateIfMissing) {
var result = repository.findDailyStat(vin, date)
.or(() -> calculateIfMissing ? dailyStatService.calculateAndSave(vin, date) : java.util.Optional.empty());
return result.map(stat -> ResponseEntity.ok(toJson(stat)))
@RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date) {
return repository.findDailyStat(vin, date)
.map(stat -> ResponseEntity.ok(toJson(stat)))
.orElseGet(() -> ResponseEntity.notFound().build());
}

33
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java
View File

@@ -8,19 +8,12 @@ import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor;
public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor {
private final VehicleStatEventProcessor processor;
private final Jt808MileageStreamProcessor jt808MileageProcessor;
public VehicleStatEnvelopeIngestor(VehicleStatEventProcessor processor) {
this(processor, null);
}
public VehicleStatEnvelopeIngestor(VehicleStatEventProcessor processor,
Jt808MileageStreamProcessor jt808MileageProcessor) {
if (processor == null) {
throw new IllegalArgumentException("processor must not be null");
public VehicleStatEnvelopeIngestor(Jt808MileageStreamProcessor jt808MileageProcessor) {
if (jt808MileageProcessor == null) {
throw new IllegalArgumentException("jt808MileageProcessor must not be null");
}
this.processor = processor;
this.jt808MileageProcessor = jt808MileageProcessor;
}
@@ -29,10 +22,9 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor {
if (!envelope.hasTelemetrySnapshot()) {
return;
}
if (processJt808Mileage(envelope)) {
return;
if ("JT808".equalsIgnoreCase(envelope.getSource())) {
jt808MileageProcessor.process(envelope);
}
processor.process(envelope);
}
@Override
@@ -44,11 +36,12 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor {
return EnvelopeIngestResult.skipped(
envelope.getEventId(), envelope.getVin(), "envelope telemetry_snapshot is required");
}
if (processJt808Mileage(envelope)) {
if ("JT808".equalsIgnoreCase(envelope.getSource())) {
jt808MileageProcessor.process(envelope);
return EnvelopeIngestResult.processed(envelope.getEventId(), envelope.getVin());
}
processor.process(envelope);
return EnvelopeIngestResult.processed(envelope.getEventId(), envelope.getVin());
return EnvelopeIngestResult.skipped(
envelope.getEventId(), envelope.getVin(), "vehicle-stat only accepts JT808 telemetry");
} catch (IllegalArgumentException ex) {
return envelope == null
? EnvelopeIngestResult.invalid(ex.getMessage())
@@ -56,14 +49,6 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor {
}
}
private boolean processJt808Mileage(VehicleEnvelope envelope) {
if (jt808MileageProcessor != null && "JT808".equalsIgnoreCase(envelope.getSource())) {
jt808MileageProcessor.process(envelope);
return true;
}
return false;
}
private static VehicleEnvelope parse(byte[] kafkaValue) {
if (kafkaValue == null || kafkaValue.length == 0) {
throw new IllegalArgumentException("VehicleEnvelope bytes must not be empty");

54
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEventProcessor.java
View File

@@ -1,54 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import java.time.Instant;
import java.util.OptionalDouble;
public final class VehicleStatEventProcessor {
private static final String TOTAL_MILEAGE_KEY = "total_mileage_km";
private final VehicleStatRepository repository;
public VehicleStatEventProcessor(VehicleStatRepository repository) {
if (repository == null) {
throw new IllegalArgumentException("repository must not be null");
}
this.repository = repository;
}
public void process(VehicleEnvelope envelope) {
if (envelope == null) {
throw new IllegalArgumentException("envelope must not be null");
}
if (!envelope.hasTelemetrySnapshot()) {
return;
}
// 统计消费的是已经标准化后的 telemetry_snapshot，不重新解析 RAW .bin。
OptionalDouble totalMileage = totalMileage(envelope);
if (totalMileage.isEmpty()) {
return;
}
repository.appendMileagePoint(
envelope.getVin(),
new MileagePoint(Instant.ofEpochMilli(envelope.getEventTimeMs()), totalMileage.getAsDouble()));
}
private static OptionalDouble totalMileage(VehicleEnvelope envelope) {
for (TelemetryField field : envelope.getTelemetrySnapshot().getFieldsList()) {
if (!TOTAL_MILEAGE_KEY.equals(field.getKey())) {
continue;
}
try {
double value = Double.parseDouble(field.getValue());
return Double.isFinite(value) ? OptionalDouble.of(value) : OptionalDouble.empty();
} catch (NumberFormatException ex) {
return OptionalDouble.empty();
}
}
return OptionalDouble.empty();
}
}

62
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEventSink.java
View File

@@ -1,62 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.sink.EventSink;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.concurrent.CompletableFuture;
public final class VehicleStatEventSink implements EventSink {
private final VehicleStatRepository repository;
private final DailyVehicleStatService dailyStatService;
private final ZoneId zoneId;
public VehicleStatEventSink(VehicleStatRepository repository,
DailyVehicleStatService dailyStatService,
ZoneId zoneId) {
if (repository == null) {
throw new IllegalArgumentException("repository must not be null");
}
if (dailyStatService == null) {
throw new IllegalArgumentException("dailyStatService must not be null");
}
this.repository = repository;
this.dailyStatService = dailyStatService;
this.zoneId = zoneId == null ? ZoneId.of("Asia/Shanghai") : zoneId;
}
@Override
public String name() {
return "vehicle-stat";
}
@Override
public boolean accepts(VehicleEvent event) {
// 直接 EventSink 路径只服务老的 Realtime 派生统计；32960 RAW 历史链路不依赖它。
return event instanceof VehicleEvent.Realtime realtime
&& realtime.payload() != null
&& realtime.payload().totalMileageKm() != null;
}
@Override
public CompletableFuture<Void> publish(VehicleEvent event) {
if (!accepts(event)) {
return CompletableFuture.completedFuture(null);
}
CompletableFuture<Void> future = new CompletableFuture<>();
try {
VehicleEvent.Realtime realtime = (VehicleEvent.Realtime) event;
repository.appendMileagePoint(
realtime.vin(),
new MileagePoint(realtime.eventTime(), realtime.payload().totalMileageKm()));
LocalDate statDate = LocalDate.ofInstant(realtime.eventTime(), zoneId);
dailyStatService.calculateAndSave(realtime.vin(), statDate);
future.complete(null);
} catch (RuntimeException e) {
future.completeExceptionally(e);
}
return future;
}
}

5
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatRepository.java
View File

@@ -1,15 +1,10 @@
package com.lingniu.ingest.vehiclestat;
import java.time.LocalDate;
import java.util.List;
import java.util.Optional;
public interface VehicleStatRepository {
void appendMileagePoint(String vin, MileagePoint point);
List<MileagePoint> mileagePoints(String vin, LocalDate statDate);
void saveDailyStat(VehicleDailyStatResult result);
Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate);

13
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatRule.java
View File

@@ -1,13 +0,0 @@
package com.lingniu.ingest.vehiclestat;
public record VehicleStatRule(String vin, DailyMileageStrategy dailyMileageStrategy) {
public VehicleStatRule {
if (vin == null || vin.isBlank()) {
throw new IllegalArgumentException("vin must not be blank");
}
if (dailyMileageStrategy == null) {
dailyMileageStrategy = DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST;
}
}
}

6
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatRuleRepository.java
View File

@@ -1,6 +0,0 @@
package com.lingniu.ingest.vehiclestat;
public interface VehicleStatRuleRepository {
VehicleStatRule ruleFor(String vin);
}

62
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java
View File

@@ -2,17 +2,10 @@ package com.lingniu.ingest.vehiclestat.config;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.vehiclestat.DailyMileageCalculator;
import com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
import com.lingniu.ingest.vehiclestat.DailyVehicleStatService;
import com.lingniu.ingest.vehiclestat.JdbcVehicleStatMetricRepository;
import com.lingniu.ingest.vehiclestat.VehicleStatController;
import com.lingniu.ingest.vehiclestat.VehicleStatEnvelopeIngestor;
import com.lingniu.ingest.vehiclestat.VehicleStatEventSink;
import com.lingniu.ingest.vehiclestat.VehicleStatEventProcessor;
import com.lingniu.ingest.vehiclestat.VehicleStatRepository;
import com.lingniu.ingest.vehiclestat.VehicleStatRule;
import com.lingniu.ingest.vehiclestat.VehicleStatRuleRepository;
import com.lingniu.ingest.vehiclestat.jt808.InMemoryJt808MileageStateStore;
import com.lingniu.ingest.vehiclestat.jt808.Jt808LocationPointExtractor;
import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStateStore;
@@ -27,7 +20,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
@@ -51,43 +43,6 @@ public class VehicleStatAutoConfiguration {
return new JdbcVehicleStatMetricRepository(jdbcTemplate);
}
@Bean
@ConditionalOnMissingBean
public VehicleStatRuleRepository vehicleStatRuleRepository() {
return vin -> new VehicleStatRule(vin, DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST);
}
@Bean
@ConditionalOnMissingBean
public DailyMileageCalculator dailyMileageCalculator(VehicleStatProperties props) {
return new DailyMileageCalculator(ZoneId.of(props.getZoneId()));
}
@Bean
@ConditionalOnBean(VehicleStatRepository.class)
@ConditionalOnMissingBean
public DailyVehicleStatService dailyVehicleStatService(VehicleStatRepository repository,
VehicleStatRuleRepository ruleRepository,
DailyMileageCalculator mileageCalculator) {
return new DailyVehicleStatService(repository, ruleRepository, mileageCalculator);
}
@Bean
@ConditionalOnBean(VehicleStatRepository.class)
@ConditionalOnMissingBean
public VehicleStatEventProcessor vehicleStatEventProcessor(VehicleStatRepository repository) {
return new VehicleStatEventProcessor(repository);
}
@Bean
@ConditionalOnBean(VehicleStatEventProcessor.class)
@ConditionalOnMissingBean
public VehicleStatEnvelopeIngestor vehicleStatEnvelopeIngestor(
VehicleStatEventProcessor processor,
ObjectProvider<Jt808MileageStreamProcessor> jt808MileageProcessor) {
return new VehicleStatEnvelopeIngestor(processor, jt808MileageProcessor.getIfAvailable());
}
@Bean
@ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true")
@ConditionalOnMissingBean
@@ -110,6 +65,7 @@ public class VehicleStatAutoConfiguration {
@Bean
@ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true")
@ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "state-store", havingValue = "memory")
@ConditionalOnMissingBean(Jt808MileageStateStore.class)
public Jt808MileageStateStore inMemoryJt808MileageStateStore() {
return new InMemoryJt808MileageStateStore();
@@ -128,20 +84,18 @@ public class VehicleStatAutoConfiguration {
}
@Bean
@ConditionalOnBean({VehicleStatRepository.class, DailyVehicleStatService.class})
@ConditionalOnBean(Jt808MileageStreamProcessor.class)
@ConditionalOnMissingBean
public VehicleStatEventSink vehicleStatEventSink(VehicleStatRepository repository,
DailyVehicleStatService dailyStatService,
VehicleStatProperties props) {
return new VehicleStatEventSink(repository, dailyStatService, ZoneId.of(props.getZoneId()));
public VehicleStatEnvelopeIngestor vehicleStatEnvelopeIngestor(
Jt808MileageStreamProcessor jt808MileageProcessor) {
return new VehicleStatEnvelopeIngestor(jt808MileageProcessor);
}
@Bean
@ConditionalOnBean(DailyVehicleStatService.class)
@ConditionalOnBean(VehicleStatRepository.class)
@ConditionalOnMissingBean
public VehicleStatController vehicleStatController(VehicleStatRepository repository,
DailyVehicleStatService dailyStatService) {
return new VehicleStatController(repository, dailyStatService);
public VehicleStatController vehicleStatController(VehicleStatRepository repository) {
return new VehicleStatController(repository);
}
@Bean

2
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java
View File

@@ -28,7 +28,7 @@ public class VehicleStatProperties {
public static class Jt808 {
private boolean enabled;
private String stateStore = "memory";
private String stateStore = "redis";
private String redisKeyPrefix = "vehicle:mileage:jt808:daily:";
private long stateTtlDays = 3;

59
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/DailyMileageCalculatorTest.java
View File

@@ -1,59 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.List;
import java.util.OptionalDouble;
import static org.assertj.core.api.Assertions.assertThat;
class DailyMileageCalculatorTest {
private static final ZoneId ZONE = ZoneId.of("Asia/Shanghai");
private final DailyMileageCalculator calculator = new DailyMileageCalculator(ZONE);
@Test
void calculatesCurrentLastMinusPreviousLast() {
OptionalDouble mileage = calculator.calculate(
LocalDate.of(2026, 6, 22),
DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST,
List.of(
point("2026-06-21T15:59:00Z", 1_000.0),
point("2026-06-22T00:10:00Z", 1_002.0),
point("2026-06-22T10:00:00Z", 1_035.5)));
assertThat(mileage).hasValue(35.5);
}
@Test
void calculatesDayMaxMinusDayMin() {
OptionalDouble mileage = calculator.calculate(
LocalDate.of(2026, 6, 22),
DailyMileageStrategy.DAY_MAX_MINUS_DAY_MIN,
List.of(
point("2026-06-22T00:10:00Z", 1_002.0),
point("2026-06-22T10:00:00Z", 1_035.5),
point("2026-06-22T11:00:00Z", 1_030.0)));
assertThat(mileage).hasValue(33.5);
}
@Test
void ignoresNegativeMileageFromCounterReset() {
OptionalDouble mileage = calculator.calculate(
LocalDate.of(2026, 6, 22),
DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST,
List.of(
point("2026-06-21T15:59:00Z", 1_000.0),
point("2026-06-22T10:00:00Z", 10.0)));
assertThat(mileage).isEmpty();
}
private static MileagePoint point(String instant, double totalMileageKm) {
return new MileagePoint(Instant.parse(instant), totalMileageKm);
}
}

90
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/DailyVehicleStatServiceTest.java
View File

@@ -1,90 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.List;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
class DailyVehicleStatServiceTest {
private final InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
private final VehicleStatRuleRepository rules = vin -> new VehicleStatRule(
vin,
"VIN-DAYMAX".equals(vin)
? DailyMileageStrategy.DAY_MAX_MINUS_DAY_MIN
: DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST);
private final DailyVehicleStatService service = new DailyVehicleStatService(
repository,
rules,
new DailyMileageCalculator(ZoneId.of("Asia/Shanghai")));
@Test
void calculatesAndSavesDailyMileageWithVehicleRule() {
repository.appendMileagePoint("VIN001", point("2026-06-21T15:59:00Z", 100.0));
repository.appendMileagePoint("VIN001", point("2026-06-22T10:00:00Z", 135.5));
Optional<VehicleDailyStatResult> result = service.calculateAndSave("VIN001", LocalDate.of(2026, 6, 22));
assertThat(result).hasValueSatisfying(stat -> {
assertThat(stat.vin()).isEqualTo("VIN001");
assertThat(stat.statDate()).isEqualTo(LocalDate.of(2026, 6, 22));
assertThat(stat.dailyMileageKm()).hasValue(35.5);
});
assertThat(repository.findDailyStat("VIN001", LocalDate.of(2026, 6, 22))).isEqualTo(result);
}
@Test
void usesPerVehicleDayMaxStrategy() {
repository.appendMileagePoint("VIN-DAYMAX", point("2026-06-22T00:10:00Z", 100.0));
repository.appendMileagePoint("VIN-DAYMAX", point("2026-06-22T10:00:00Z", 118.0));
repository.appendMileagePoint("VIN-DAYMAX", point("2026-06-22T11:00:00Z", 115.0));
Optional<VehicleDailyStatResult> result =
service.calculateAndSave("VIN-DAYMAX", LocalDate.of(2026, 6, 22));
assertThat(result).hasValueSatisfying(stat -> assertThat(stat.dailyMileageKm()).hasValue(18.0));
}
@Test
void returnsEmptyWhenMileageCannotBeCalculated() {
repository.appendMileagePoint("VIN001", point("2026-06-22T10:00:00Z", 135.5));
Optional<VehicleDailyStatResult> result = service.calculateAndSave("VIN001", LocalDate.of(2026, 6, 22));
assertThat(result).isEmpty();
}
private static MileagePoint point(String instant, double totalMileageKm) {
return new MileagePoint(Instant.parse(instant), totalMileageKm);
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
private final java.util.Map<String, java.util.List<MileagePoint>> points = new java.util.HashMap<>();
private final java.util.Map<String, VehicleDailyStatResult> results = new java.util.HashMap<>();
@Override
public void appendMileagePoint(String vin, MileagePoint point) {
points.computeIfAbsent(vin, ignored -> new java.util.ArrayList<>()).add(point);
}
@Override
public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return points.getOrDefault(vin, List.of());
}
@Override
public void saveDailyStat(VehicleDailyStatResult result) {
results.put(result.vin() + ":" + result.statDate(), result);
}
@Override
public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
return Optional.ofNullable(results.get(vin + ":" + statDate));
}
}
}

38
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/FileVehicleStatRepositoryTest.java
View File

@@ -1,38 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Path;
import java.time.Instant;
import java.time.LocalDate;
import java.util.OptionalDouble;
import static org.assertj.core.api.Assertions.assertThat;
class FileVehicleStatRepositoryTest {
@TempDir
Path tempDir;
@Test
void persistsMileagePointsAndLatestDailyStat() {
FileVehicleStatRepository repository = new FileVehicleStatRepository(tempDir);
repository.appendMileagePoint("VIN001", new MileagePoint(Instant.parse("2026-06-21T15:59:00Z"), 100.0));
repository.appendMileagePoint("VIN001", new MileagePoint(Instant.parse("2026-06-22T10:00:00Z"), 135.5));
repository.saveDailyStat(new VehicleDailyStatResult(
"VIN001",
LocalDate.parse("2026-06-22"),
OptionalDouble.of(35.5),
DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST));
FileVehicleStatRepository reopened = new FileVehicleStatRepository(tempDir);
assertThat(reopened.mileagePoints("VIN001", LocalDate.parse("2026-06-22")))
.extracting(MileagePoint::totalMileageKm)
.containsExactly(100.0, 135.5);
assertThat(reopened.findDailyStat("VIN001", LocalDate.parse("2026-06-22")))
.hasValueSatisfying(stat -> assertThat(stat.dailyMileageKm()).hasValue(35.5));
}
}

33
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatControllerTest.java
View File

@@ -3,9 +3,9 @@ package com.lingniu.ingest.vehiclestat;
import org.junit.jupiter.api.Test;
import org.springframework.test.web.servlet.MockMvc;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.Optional;
import java.util.OptionalDouble;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
@@ -17,14 +17,12 @@ class VehicleStatControllerTest {
@Test
void returnsDailyStatJson() throws Exception {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
DailyVehicleStatService service = new DailyVehicleStatService(
repository,
vin -> new VehicleStatRule(vin, DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST),
new DailyMileageCalculator(ZoneId.of("Asia/Shanghai")));
repository.appendMileagePoint("VIN001", new MileagePoint(Instant.parse("2026-06-21T15:59:00Z"), 100.0));
repository.appendMileagePoint("VIN001", new MileagePoint(Instant.parse("2026-06-22T10:00:00Z"), 135.5));
service.calculateAndSave("VIN001", LocalDate.parse("2026-06-22"));
MockMvc mvc = standaloneSetup(new VehicleStatController(repository, service)).build();
repository.saveDailyStat(new VehicleDailyStatResult(
"VIN001",
LocalDate.parse("2026-06-22"),
OptionalDouble.of(35.5),
DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST));
MockMvc mvc = standaloneSetup(new VehicleStatController(repository)).build();
mvc.perform(get("/api/vehicle-stat/VIN001/daily").param("date", "2026-06-22"))
.andExpect(status().isOk())
@@ -34,27 +32,16 @@ class VehicleStatControllerTest {
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
private final java.util.Map<String, java.util.List<MileagePoint>> points = new java.util.HashMap<>();
private final java.util.Map<String, VehicleDailyStatResult> results = new java.util.HashMap<>();
@Override
public void appendMileagePoint(String vin, MileagePoint point) {
points.computeIfAbsent(vin, ignored -> new java.util.ArrayList<>()).add(point);
}
@Override
public java.util.List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return points.getOrDefault(vin, java.util.List.of());
}
@Override
public void saveDailyStat(VehicleDailyStatResult result) {
results.put(result.vin() + ":" + result.statDate(), result);
}
@Override
public java.util.Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
return java.util.Optional.ofNullable(results.get(vin + ":" + statDate));
public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
return Optional.ofNullable(results.get(vin + ":" + statDate));
}
}
}

98
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java
View File

@@ -9,68 +9,46 @@ import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor;
import org.junit.jupiter.api.Test;
import java.time.LocalDate;
import java.util.List;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
class VehicleStatEnvelopeIngestorTest {
@Test
void parsesEnvelopeBytesAndProcessesMileagePoint() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
VehicleStatEnvelopeIngestor ingestor =
new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(repository));
assertThat(ingestor).isInstanceOf(EnvelopeIngestor.class);
ingestor.ingest(envelope().toByteArray());
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22)))
.extracting(MileagePoint::totalMileageKm)
.containsExactly(123.45);
}
@Test
void doesNotFanOutNonJt808TelemetryToJt808MileageProcessor() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
void ignoresNonJt808Telemetry() {
Jt808MileageStreamProcessor jt808Processor = mock(Jt808MileageStreamProcessor.class);
VehicleEnvelope envelope = envelope("GB32960");
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(
new VehicleStatEventProcessor(repository),
jt808Processor);
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(jt808Processor);
ingestor.ingest(envelope.toByteArray());
var result = ingestor.tryIngest(envelope.toByteArray());
verifyNoInteractions(jt808Processor);
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22)))
.extracting(MileagePoint::totalMileageKm)
.containsExactly(123.45);
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED);
assertThat(result.message()).contains("only accepts JT808");
}
@Test
void routesJt808TelemetryOnlyToJt808MileageProcessorWhenAvailable() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
Jt808MileageStreamProcessor jt808Processor = mock(Jt808MileageStreamProcessor.class);
VehicleEnvelope envelope = envelope("JT808");
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(
new VehicleStatEventProcessor(repository),
jt808Processor);
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(jt808Processor);
assertThat(ingestor).isInstanceOf(EnvelopeIngestor.class);
ingestor.ingest(envelope.toByteArray());
var result = ingestor.tryIngest(envelope.toByteArray());
verify(jt808Processor).process(envelope);
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
verify(jt808Processor, times(2)).process(envelope);
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.PROCESSED);
}
@Test
void rejectsInvalidEnvelopeBytes() {
VehicleStatEnvelopeIngestor ingestor =
new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(new InMemoryVehicleStatRepository()));
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(mock(Jt808MileageStreamProcessor.class));
assertThatThrownBy(() -> ingestor.ingest(new byte[]{0x01, 0x02}))
.isInstanceOf(IllegalArgumentException.class)
@@ -78,23 +56,21 @@ class VehicleStatEnvelopeIngestorTest {
}
@Test
void tryIngestInvalidEnvelopeBytesReturnsInvalidWithoutAppendingMileagePoint() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
VehicleStatEnvelopeIngestor ingestor =
new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(repository));
void tryIngestInvalidEnvelopeBytesReturnsInvalidWithoutProcessing() {
Jt808MileageStreamProcessor jt808Processor = mock(Jt808MileageStreamProcessor.class);
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(jt808Processor);
var result = ingestor.tryIngest(new byte[]{0x01, 0x02});
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.INVALID_ENVELOPE);
assertThat(result.message()).contains("VehicleEnvelope");
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
verifyNoInteractions(jt808Processor);
}
@Test
void tryIngestEnvelopeWithoutTelemetrySnapshotReturnsSkipped() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
VehicleStatEnvelopeIngestor ingestor =
new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(repository));
Jt808MileageStreamProcessor jt808Processor = mock(Jt808MileageStreamProcessor.class);
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(jt808Processor);
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setEventId("event-login-1")
.setVin("VIN001")
@@ -108,14 +84,13 @@ class VehicleStatEnvelopeIngestorTest {
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED);
assertThat(result.eventId()).isEqualTo("event-login-1");
assertThat(result.message()).contains("telemetry_snapshot");
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
verifyNoInteractions(jt808Processor);
}
@Test
void rawArchiveEnvelopeIsIgnoredWithoutAppendingMileagePoint() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
VehicleStatEnvelopeIngestor ingestor =
new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(repository));
void rawArchiveEnvelopeIsIgnoredWithoutProcessing() {
Jt808MileageStreamProcessor jt808Processor = mock(Jt808MileageStreamProcessor.class);
VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor(jt808Processor);
VehicleEnvelope envelope = rawArchiveEnvelope();
ingestor.ingest(envelope.toByteArray());
@@ -125,11 +100,7 @@ class VehicleStatEnvelopeIngestorTest {
assertThat(result.eventId()).isEqualTo("raw-event-1");
assertThat(result.vin()).isEqualTo("VIN001");
assertThat(result.message()).contains("telemetry_snapshot");
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
}
private static VehicleEnvelope envelope() {
return envelope("GB32960");
verifyNoInteractions(jt808Processor);
}
private static VehicleEnvelope envelope(String source) {
@@ -162,27 +133,4 @@ class VehicleStatEnvelopeIngestorTest {
.build())
.build();
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
private final java.util.Map<String, java.util.List<MileagePoint>> points = new java.util.HashMap<>();
@Override
public void appendMileagePoint(String vin, MileagePoint point) {
points.computeIfAbsent(vin, ignored -> new java.util.ArrayList<>()).add(point);
}
@Override
public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return points.getOrDefault(vin, List.of());
}
@Override
public void saveDailyStat(VehicleDailyStatResult result) {
}
@Override
public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
return Optional.empty();
}
}
}

74
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEventProcessorTest.java
View File

@@ -1,74 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import org.junit.jupiter.api.Test;
import java.time.LocalDate;
import java.util.List;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
class VehicleStatEventProcessorTest {
private final InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
private final VehicleStatEventProcessor processor = new VehicleStatEventProcessor(repository);
@Test
void appendsMileagePointFromInternalField() {
processor.process(envelope("total_mileage_km", "123.45"));
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22)))
.extracting(MileagePoint::totalMileageKm)
.containsExactly(123.45);
}
@Test
void ignoresEventsWithoutMileageField() {
processor.process(envelope("speed_kmh", "80.5"));
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
}
private static VehicleEnvelope envelope(String key, String value) {
return VehicleEnvelope.newBuilder()
.setEventId("event-1")
.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(key)
.setValueType("DOUBLE")
.setValue(value)
.setQuality("GOOD")))
.build();
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
private final java.util.Map<String, java.util.List<MileagePoint>> points = new java.util.HashMap<>();
@Override
public void appendMileagePoint(String vin, MileagePoint point) {
points.computeIfAbsent(vin, ignored -> new java.util.ArrayList<>()).add(point);
}
@Override
public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return points.getOrDefault(vin, List.of());
}
@Override
public void saveDailyStat(VehicleDailyStatResult result) {
}
@Override
public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
return Optional.empty();
}
}
}

54
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEventSinkTest.java
View File

@@ -1,54 +0,0 @@
package com.lingniu.ingest.vehiclestat;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.RealtimePayload;
import com.lingniu.ingest.api.event.VehicleEvent;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Path;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
class VehicleStatEventSinkTest {
@TempDir
Path tempDir;
@Test
void consumesRealtimeEventsAndCalculatesDailyMileage() {
FileVehicleStatRepository repository = new FileVehicleStatRepository(tempDir);
DailyVehicleStatService service = new DailyVehicleStatService(
repository,
vin -> new VehicleStatRule(vin, DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST),
new DailyMileageCalculator(ZoneId.of("Asia/Shanghai")));
VehicleStatEventSink sink = new VehicleStatEventSink(repository, service, ZoneId.of("Asia/Shanghai"));
sink.publish(realtime("previous", Instant.parse("2026-06-21T15:59:00Z"), 100.0)).join();
sink.publish(realtime("current", Instant.parse("2026-06-22T10:00:00Z"), 135.5)).join();
assertThat(repository.findDailyStat("VIN001", LocalDate.parse("2026-06-22")))
.hasValueSatisfying(stat -> assertThat(stat.dailyMileageKm()).hasValue(35.5));
}
private static VehicleEvent.Realtime realtime(String id, Instant eventTime, double totalMileageKm) {
return new VehicleEvent.Realtime(
id,
"VIN001",
ProtocolId.GB32960,
eventTime,
eventTime.plusMillis(100),
"trace-" + id,
Map.of(),
new RealtimePayload(
42.1, totalMileageKm, 87.0, null, null,
null, null, null, 8.3, null, null,
RealtimePayload.VehicleState.STARTED,
null, null, null, null, null,
113.12, 23.45, null, null, null, null));
}
}

66
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java
View File

@@ -2,18 +2,10 @@ package com.lingniu.ingest.vehiclestat.config;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.vehiclestat.DailyMileageCalculator;
import com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
import com.lingniu.ingest.vehiclestat.DailyVehicleStatService;
import com.lingniu.ingest.vehiclestat.JdbcVehicleStatMetricRepository;
import com.lingniu.ingest.vehiclestat.MileagePoint;
import com.lingniu.ingest.vehiclestat.VehicleStatController;
import com.lingniu.ingest.vehiclestat.VehicleDailyStatResult;
import com.lingniu.ingest.vehiclestat.VehicleStatEnvelopeIngestor;
import com.lingniu.ingest.vehiclestat.VehicleStatEventSink;
import com.lingniu.ingest.vehiclestat.VehicleStatEventProcessor;
import com.lingniu.ingest.vehiclestat.VehicleStatRepository;
import com.lingniu.ingest.vehiclestat.VehicleStatRuleRepository;
import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStateStore;
import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor;
import com.lingniu.ingest.vehiclestat.jt808.RedisJt808MileageStateStore;
@@ -27,9 +19,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import javax.sql.DataSource;
import java.time.LocalDate;
import java.util.List;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
@@ -38,33 +27,27 @@ class VehicleStatAutoConfigurationTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(VehicleStatAutoConfiguration.class))
.withBean(EnvelopeDeadLetterSink.class, () -> record -> {})
.withBean(VehicleStatRepository.class, InMemoryVehicleStatRepository::new);
.withBean(EnvelopeDeadLetterSink.class, () -> record -> {});
@Test
void createsStatBeansWhenEnabledAndRepositoryExists() {
void createsOnlyMetricRepositoryAndReadControllerWhenJt808MileageDisabled() {
contextRunner
.withBean(JdbcTemplate.class, () -> mock(JdbcTemplate.class))
.withPropertyValues("lingniu.ingest.vehicle-stat.enabled=true")
.run(context -> {
assertThat(context).hasSingleBean(VehicleStatRuleRepository.class);
assertThat(context).hasSingleBean(DailyMileageCalculator.class);
assertThat(context).hasSingleBean(DailyVehicleStatService.class);
assertThat(context).hasSingleBean(VehicleStatEventProcessor.class);
assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class);
assertThat(context).hasSingleBean(VehicleStatEventSink.class);
assertThat(context).hasSingleBean(VehicleStatRepository.class);
assertThat(context).hasSingleBean(JdbcVehicleStatMetricRepository.class);
assertThat(context).hasSingleBean(VehicleStatController.class);
assertThat(context).hasSingleBean(EnvelopeConsumerProcessor.class);
assertThat(context.getBean(VehicleStatRuleRepository.class)
.ruleFor("VIN001").dailyMileageStrategy())
.isEqualTo(DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST);
assertThat(context).doesNotHaveBean(Jt808MileageStreamProcessor.class);
assertThat(context).doesNotHaveBean(VehicleStatEnvelopeIngestor.class);
assertThat(context).doesNotHaveBean(EnvelopeConsumerProcessor.class);
});
}
@Test
void backsOffWhenDisabled() {
contextRunner.run(context -> {
assertThat(context).doesNotHaveBean(DailyVehicleStatService.class);
assertThat(context).doesNotHaveBean(VehicleStatEventProcessor.class);
assertThat(context).doesNotHaveBean(VehicleStatRepository.class);
assertThat(context).doesNotHaveBean(VehicleStatEnvelopeIngestor.class);
});
}
@@ -76,8 +59,7 @@ class VehicleStatAutoConfigurationTest {
.withPropertyValues("lingniu.ingest.vehicle-stat.enabled=true")
.run(context -> {
assertThat(context).doesNotHaveBean(VehicleStatRepository.class);
assertThat(context).doesNotHaveBean(DailyVehicleStatService.class);
assertThat(context).doesNotHaveBean(VehicleStatEventProcessor.class);
assertThat(context).doesNotHaveBean(VehicleStatController.class);
});
}
@@ -96,11 +78,26 @@ class VehicleStatAutoConfigurationTest {
assertThat(context).hasSingleBean(RedisJt808MileageStateStore.class);
assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class);
assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class);
assertThat(context).hasSingleBean(EnvelopeConsumerProcessor.class);
});
}
@Test
void fallsBackToInMemoryJt808StateStoreWhenRedisIsNotSelected() {
void doesNotFallbackToMemoryJt808StateStoreWhenRedisIsDefaultButMissing() {
contextRunner
.withBean(JdbcTemplate.class, () -> mock(JdbcTemplate.class))
.withPropertyValues(
"lingniu.ingest.vehicle-stat.enabled=true",
"lingniu.ingest.vehicle-stat.jt808.enabled=true")
.run(context -> {
assertThat(context).doesNotHaveBean(Jt808MileageStateStore.class);
assertThat(context).doesNotHaveBean(Jt808MileageStreamProcessor.class);
assertThat(context).doesNotHaveBean(VehicleStatEnvelopeIngestor.class);
});
}
@Test
void createsInMemoryJt808StateStoreOnlyWhenExplicitlySelected() {
contextRunner
.withBean(JdbcTemplate.class, () -> mock(JdbcTemplate.class))
.withPropertyValues(
@@ -111,6 +108,7 @@ class VehicleStatAutoConfigurationTest {
assertThat(context).hasSingleBean(Jt808MileageStateStore.class);
assertThat(context).doesNotHaveBean(RedisJt808MileageStateStore.class);
assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class);
assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class);
});
}
@@ -134,6 +132,7 @@ class VehicleStatAutoConfigurationTest {
assertThat(context).hasSingleBean(JdbcVehicleStatMetricRepository.class);
assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class);
assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class);
assertThat(context).hasSingleBean(EnvelopeConsumerProcessor.class);
});
}
@@ -155,13 +154,4 @@ class VehicleStatAutoConfigurationTest {
assertThat(context).hasSingleBean(JdbcVehicleStatMetricRepository.class);
});
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
@Override public void appendMileagePoint(String vin, MileagePoint point) {}
@Override public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) { return List.of(); }
@Override public void saveDailyStat(VehicleDailyStatResult result) {}
@Override public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
return Optional.empty();
}
}
}

10
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessorTest.java
View File

@@ -4,7 +4,6 @@ import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
import com.lingniu.ingest.vehiclestat.MileagePoint;
import com.lingniu.ingest.vehiclestat.VehicleDailyStatResult;
import com.lingniu.ingest.vehiclestat.VehicleStatRepository;
import org.junit.jupiter.api.Test;
@@ -106,15 +105,6 @@ class Jt808MileageStreamProcessorTest {
private static final class CapturingRepository implements VehicleStatRepository {
private final List<VehicleDailyStatResult> results = new ArrayList<>();
@Override
public void appendMileagePoint(String vin, MileagePoint point) {
}
@Override
public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return List.of();
}
@Override
public void saveDailyStat(VehicleDailyStatResult result) {
results.add(result);