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

Merge branch 'codex/gb32960-service-split'

Browse Source
This commit is contained in:
lingniu
2026-06-24 14:32:59 +08:00
parent b983f77c23 620821b624
commit a70f8f39ad
48 changed files with 4166 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
modules/apps/gb32960-ingest-app/pom.xml Normal file
View File

@@ -0,0 +1,66 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lingniu.ingest</groupId>
<artifactId>lingniu-vehicle-ingest</artifactId>
<version>0.1.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
<artifactId>gb32960-ingest-app</artifactId>
<name>gb32960-ingest-app</name>
<description>GB32960 protocol ingress runtime: TCP decode, auth, Kafka production, and ACK.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-core</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>session-core</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>observability</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>protocol-gb32960</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-mq</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>gb32960-ingest-app</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
</configuration>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

21
modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java Normal file
View File

@@ -0,0 +1,21 @@
package com.lingniu.ingest.gb32960app;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
public class Gb32960IngestApplication {
private static final Logger log = LoggerFactory.getLogger(Gb32960IngestApplication.class);
public static void main(String[] args) {
try {
SpringApplication.run(Gb32960IngestApplication.class, args);
} catch (Throwable t) {
log.error("GB32960 ingest application failed to start, forcing JVM exit", t);
System.exit(1);
}
}
}

110
modules/apps/gb32960-ingest-app/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,110 @@
spring:
application:
name: gb32960-ingest-app
threads:
virtual:
enabled: true
server:
port: ${HTTP_PORT:20100}
lingniu:
ingest:
gb32960:
enabled: true
server:
enabled: true
port: ${GB32960_PORT:32960}
boss-threads: ${GB32960_BOSS_THREADS:1}
worker-threads: ${GB32960_WORKER_THREADS:0}
auth:
enabled: ${GB32960_AUTH_ENABLED:false}
case-sensitive: false
whitelist: []
platforms:
- username: ${GB32960_PLATFORM_USER_HYUNDAI:Hyundai}
password: ${GB32960_PLATFORM_PWD_HYUNDAI:f2e3445d7cda409fb4f278f6fb890734}
allowed-ips:
- ${GB32960_PLATFORM_IP_HYUNDAI:8.134.95.166}
description: 外部下级平台 - Hyundai
tls:
enabled: ${GB32960_TLS_ENABLED:false}
cert-path: ${GB32960_TLS_CERT:}
key-path: ${GB32960_TLS_KEY:}
trust-cert-path: ${GB32960_TLS_CA:}
require-client-auth: true
vendor-extensions:
- name: guangdong-fc
match:
platform-accounts:
- Hyundai
vin-prefixes: []
vins: []
parse:
lenient-block-failure: true
diagnostics:
max-logged-frame-keys-per-channel: ${GB32960_DIAGNOSTICS_MAX_LOGGED_FRAME_KEYS_PER_CHANNEL:4096}
pipeline:
disruptor:
ring-buffer-size: ${PIPELINE_RING_BUFFER_SIZE:131072}
wait-strategy: ${PIPELINE_WAIT_STRATEGY:yielding}
producer-type: multi
dedup:
enabled: true
cache-size: 200000
ttl-seconds: 600
rate-limit:
per-vin-qps: 50
session:
store: ${SESSION_STORE:memory}
ttl: ${SESSION_TTL:30m}
identity:
store: ${VEHICLE_IDENTITY_STORE:file}
file:
path: ${VEHICLE_IDENTITY_FILE:./data/vehicle-identity.jsonl}
sink:
mq:
enabled: ${KAFKA_ENABLED:true}
type: kafka
bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
compression-type: zstd
linger-ms: 20
batch-size: 65536
acks: all
enable-idempotence: true
node-id: ${KAFKA_NODE_ID:gb32960-ingest-local}
topics:
realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
location: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
alarm: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
session: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
media-meta: ${KAFKA_TOPIC_MEDIA_META:vehicle.media.meta.v1}
raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
consumer:
enabled: false
event-file-store:
enabled: false
event-history:
enabled: false
vehicle-state:
enabled: false
vehicle-stat:
enabled: false
management:
endpoints:
web:
exposure:
include: health,info,metrics,prometheus,env
health:
redis:
enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
metrics:
tags:
application: gb32960-ingest-app
logging:
level:
root: INFO
com.lingniu.ingest: INFO

68
modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java Normal file
View File

@@ -0,0 +1,68 @@
package com.lingniu.ingest.gb32960app;
import com.lingniu.ingest.core.config.IngestCoreAutoConfiguration;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
import com.lingniu.ingest.session.config.SessionCoreAutoConfiguration;
import com.lingniu.ingest.sink.mq.KafkaEventSink;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink;
import com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.util.ClassUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
class Gb32960IngestAppCompositionTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
IngestCoreAutoConfiguration.class,
SessionCoreAutoConfiguration.class,
SinkMqAutoConfiguration.class,
Gb32960AutoConfiguration.class))
.withAllowBeanDefinitionOverriding(true)
.withBean("kafkaProducer", KafkaProducer.class, Gb32960IngestAppCompositionTest::kafkaProducer)
.withPropertyValues(
"lingniu.ingest.gb32960.enabled=true",
"lingniu.ingest.gb32960.server.enabled=true",
"lingniu.ingest.gb32960.port=0",
"lingniu.ingest.session.store=memory",
"lingniu.ingest.sink.mq.enabled=true",
"lingniu.ingest.sink.mq.type=kafka",
"lingniu.ingest.sink.mq.bootstrap-servers=localhost:9092",
"lingniu.ingest.sink.mq.consumer.enabled=false",
"lingniu.ingest.event-file-store.enabled=false",
"lingniu.ingest.event-history.enabled=false",
"lingniu.ingest.vehicle-state.enabled=false",
"lingniu.ingest.vehicle-stat.enabled=false");
@Test
void createsGb32960ListenerWithoutHistoryStorageBoundaries() {
contextRunner.run(context -> {
assertThat(context).hasSingleBean(Gb32960MessageDecoder.class);
assertThat(context).hasSingleBean(Gb32960NettyServer.class);
assertThat(context).hasSingleBean(KafkaEventSink.class);
assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
assertTypeNotPresent(context, "com.lingniu.ingest.sink.archive.ArchiveStore");
assertTypeNotPresent(context, "com.lingniu.ingest.eventfilestore.EventFileStore");
});
}
@SuppressWarnings("unchecked")
private static KafkaProducer<String, byte[]> kafkaProducer() {
return mock(KafkaProducer.class);
}
private static void assertTypeNotPresent(ApplicationContext context, String className) {
assertThat(ClassUtils.isPresent(className, context.getClassLoader()))
.as("%s must stay outside the GB32960 ingest runtime", className)
.isFalse();
}
}

39
modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java Normal file
View File

@@ -0,0 +1,39 @@
package com.lingniu.ingest.gb32960app;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;
import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
class Gb32960IngestAppDefaultsTest {
@Test
void applicationDefaultsKeepGb32960IngestAsTcpKafkaProducerOnly() {
Properties properties = applicationProperties();
assertThat(properties)
.containsEntry("spring.application.name", "gb32960-ingest-app")
.containsEntry("lingniu.ingest.gb32960.enabled", true)
.containsEntry("lingniu.ingest.gb32960.server.enabled", true)
.containsEntry("lingniu.ingest.gb32960.port", "${GB32960_PORT:32960}")
.containsEntry("lingniu.ingest.sink.mq.enabled", "${KAFKA_ENABLED:true}")
.containsEntry("lingniu.ingest.sink.mq.consumer.enabled", false)
.containsEntry("lingniu.ingest.event-file-store.enabled", false)
.containsEntry("lingniu.ingest.event-history.enabled", false)
.containsEntry("lingniu.ingest.vehicle-state.enabled", false)
.containsEntry("lingniu.ingest.vehicle-stat.enabled", false);
}
private static Properties applicationProperties() {
YamlPropertiesFactoryBean factory = new YamlPropertiesFactoryBean();
factory.setResources(new ClassPathResource("application.yml"));
Properties properties = factory.getObject();
assertThat(properties).isNotNull();
return properties;
}
}

66
modules/apps/vehicle-analytics-app/pom.xml Normal file
View File

@@ -0,0 +1,66 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lingniu.ingest</groupId>
<artifactId>lingniu-vehicle-ingest</artifactId>
<version>0.1.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
<artifactId>vehicle-analytics-app</artifactId>
<name>vehicle-analytics-app</name>
<description>Vehicle state, daily statistics, alarms, and analytics runtime.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-api</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>observability</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-mq</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>vehicle-state-service</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>vehicle-stat-service</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>vehicle-analytics-app</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
</configuration>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

21
modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java Normal file
View File

@@ -0,0 +1,21 @@
package com.lingniu.ingest.analyticsapp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
public class VehicleAnalyticsApplication {
private static final Logger log = LoggerFactory.getLogger(VehicleAnalyticsApplication.class);
public static void main(String[] args) {
try {
SpringApplication.run(VehicleAnalyticsApplication.class, args);
} catch (Throwable t) {
log.error("Vehicle analytics application failed to start, forcing JVM exit", t);
System.exit(1);
}
}
}

62
modules/apps/vehicle-analytics-app/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,62 @@
spring:
application:
name: vehicle-analytics-app
threads:
virtual:
enabled: true
server:
port: ${HTTP_PORT:20300}
lingniu:
ingest:
gb32960:
enabled: false
sink:
mq:
enabled: ${KAFKA_ENABLED:true}
type: kafka
bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
topics:
realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
consumer:
enabled: ${KAFKA_CONSUMER_ENABLED:true}
client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-analytics}
auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest}
max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500}
bindings:
vehicleStateEnvelopeConsumerProcessor:
enabled: ${VEHICLE_STATE_ENABLED:false}
group-id: ${KAFKA_GROUP_STATE:vehicle-state}
topics:
- ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
vehicleStatEnvelopeConsumerProcessor:
enabled: ${VEHICLE_STAT_ENABLED:true}
group-id: ${KAFKA_GROUP_STAT:vehicle-stat}
topics:
- ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
archive:
enabled: false
event-file-store:
enabled: false
event-history:
enabled: false
vehicle-state:
enabled: ${VEHICLE_STATE_ENABLED:false}
vehicle-stat:
enabled: ${VEHICLE_STAT_ENABLED:true}
file-path: ${VEHICLE_STAT_FILE_PATH:./target/vehicle-stat/}
zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai}
management:
endpoints:
web:
exposure:
include: health,info,metrics,prometheus,env
health:
redis:
enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
metrics:
tags:
application: vehicle-analytics-app

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

@@ -0,0 +1,85 @@
package com.lingniu.ingest.analyticsapp;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink;
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.FileVehicleStatRepository;
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 org.apache.kafka.clients.producer.KafkaProducer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.util.ClassUtils;
import java.nio.file.Path;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
class VehicleAnalyticsAppCompositionTest {
@TempDir
Path tempDir;
@Test
void createsStatsBeansWithoutProtocolListenerOrEventFileStore() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
SinkMqAutoConfiguration.class,
VehicleStateAutoConfiguration.class,
VehicleStatAutoConfiguration.class))
.withAllowBeanDefinitionOverriding(true)
.withBean("kafkaProducer", KafkaProducer.class, VehicleAnalyticsAppCompositionTest::kafkaProducer)
.withPropertyValues(
"lingniu.ingest.sink.mq.enabled=true",
"lingniu.ingest.sink.mq.type=kafka",
"lingniu.ingest.sink.mq.bootstrap-servers=localhost:9092",
"lingniu.ingest.sink.mq.consumer.enabled=false",
"lingniu.ingest.vehicle-state.enabled=false",
"lingniu.ingest.vehicle-stat.enabled=true",
"lingniu.ingest.vehicle-stat.file-path=" + tempDir.resolve("vehicle-stat"),
"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(FileVehicleStatRepository.class);
assertThat(context).hasSingleBean(DailyVehicleStatService.class);
assertThat(context).hasSingleBean(VehicleStatEventProcessor.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);
assertThat(context).hasBean("vehicleStatEnvelopeConsumerProcessor");
assertThat(context.getBean("vehicleStatEnvelopeConsumerProcessor"))
.isInstanceOf(EnvelopeConsumerProcessor.class);
assertThat(context).doesNotHaveBean(VehicleStateEnvelopeIngestor.class);
assertTypeNotPresent(context, "com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer");
assertTypeNotPresent(context, "com.lingniu.ingest.eventfilestore.EventFileStore");
});
}
@SuppressWarnings("unchecked")
private static KafkaProducer<String, byte[]> kafkaProducer() {
return mock(KafkaProducer.class);
}
private static void assertTypeNotPresent(ApplicationContext context, String className) {
assertThat(ClassUtils.isPresent(className, context.getClassLoader()))
.as("%s must stay outside the vehicle analytics runtime", className)
.isFalse();
}
}

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

@@ -0,0 +1,49 @@
package com.lingniu.ingest.analyticsapp;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;
import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
class VehicleAnalyticsAppDefaultsTest {
@Test
void applicationDefaultsKeepAnalyticsAsStatConsumerRuntime() {
Properties properties = applicationProperties();
assertThat(properties)
.containsEntry("spring.application.name", "vehicle-analytics-app")
.containsEntry("lingniu.ingest.gb32960.enabled", false)
.containsEntry("lingniu.ingest.sink.archive.enabled", false)
.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-state.enabled", "${VEHICLE_STATE_ENABLED:false}")
.containsEntry("lingniu.ingest.sink.mq.consumer.enabled", "${KAFKA_CONSUMER_ENABLED:true}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.vehicleStatEnvelopeConsumerProcessor.enabled",
"${VEHICLE_STAT_ENABLED:true}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.vehicleStateEnvelopeConsumerProcessor.enabled",
"${VEHICLE_STATE_ENABLED:false}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.vehicleStatEnvelopeConsumerProcessor.group-id",
"${KAFKA_GROUP_STAT:vehicle-stat}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.vehicleStateEnvelopeConsumerProcessor.group-id",
"${KAFKA_GROUP_STATE:vehicle-state}");
}
private static Properties applicationProperties() {
YamlPropertiesFactoryBean factory = new YamlPropertiesFactoryBean();
factory.setResources(new ClassPathResource("application.yml"));
Properties properties = factory.getObject();
assertThat(properties).isNotNull();
return properties;
}
}

74
modules/apps/vehicle-history-app/pom.xml Normal file
View File

@@ -0,0 +1,74 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lingniu.ingest</groupId>
<artifactId>lingniu-vehicle-ingest</artifactId>
<version>0.1.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
<artifactId>vehicle-history-app</artifactId>
<name>vehicle-history-app</name>
<description>Vehicle raw archive, event history, and GB32960 frame query runtime.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-api</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>observability</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-mq</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>sink-archive</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>event-file-store</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>event-history-service</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>protocol-gb32960</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>vehicle-history-app</finalName>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<jvmArguments>--sun-misc-unsafe-memory-access=allow</jvmArguments>
</configuration>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

21
modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java Normal file
View File

@@ -0,0 +1,21 @@
package com.lingniu.ingest.historyapp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
public class VehicleHistoryApplication {
private static final Logger log = LoggerFactory.getLogger(VehicleHistoryApplication.class);
public static void main(String[] args) {
try {
SpringApplication.run(VehicleHistoryApplication.class, args);
} catch (Throwable t) {
log.error("Vehicle history application failed to start, forcing JVM exit", t);
System.exit(1);
}
}
}

65
modules/apps/vehicle-history-app/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,65 @@
spring:
application:
name: vehicle-history-app
threads:
virtual:
enabled: true
server:
port: ${HTTP_PORT:20200}
lingniu:
ingest:
gb32960:
enabled: true
server:
enabled: false
sink:
mq:
enabled: ${KAFKA_ENABLED:true}
type: kafka
bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
topics:
realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
consumer:
enabled: ${KAFKA_CONSUMER_ENABLED:true}
client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-history}
auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest}
max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500}
bindings:
eventHistoryEnvelopeConsumerProcessor:
enabled: true
group-id: ${KAFKA_GROUP_HISTORY:vehicle-history}
topics:
- ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
- ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
archive:
enabled: ${SINK_ARCHIVE_ENABLED:true}
type: local
path: ${SINK_ARCHIVE_PATH:./archive/}
event-file-store:
enabled: ${EVENT_FILE_STORE_ENABLED:true}
path: ${EVENT_FILE_STORE_PATH:./target/event-store/}
zone-id: ${EVENT_FILE_STORE_ZONE_ID:Asia/Shanghai}
batch-size: ${EVENT_FILE_STORE_BATCH_SIZE:500}
flush-interval-millis: ${EVENT_FILE_STORE_FLUSH_INTERVAL_MILLIS:1000}
event-history:
enabled: true
vehicle-state:
enabled: false
vehicle-stat:
enabled: false
management:
endpoints:
web:
exposure:
include: health,info,metrics,prometheus,env
health:
redis:
enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
metrics:
tags:
application: vehicle-history-app

83
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java Normal file
View File

@@ -0,0 +1,83 @@
package com.lingniu.ingest.historyapp;
import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.eventfilestore.EventFileStoreSink;
import com.lingniu.ingest.eventfilestore.config.EventFileStoreAutoConfiguration;
import com.lingniu.ingest.eventhistory.EventHistoryController;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
import com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
import com.lingniu.ingest.sink.archive.ArchiveStore;
import com.lingniu.ingest.sink.archive.RawArchiveEventSink;
import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink;
import com.lingniu.ingest.sink.mq.KafkaEventSink;
import com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import java.nio.file.Path;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
class VehicleHistoryAppCompositionTest {
@TempDir
Path tempDir;
@Test
void createsHistoryStorageAndQueryBeansWithoutGb32960TcpServer() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
SinkArchiveAutoConfiguration.class,
EventFileStoreAutoConfiguration.class,
SinkMqAutoConfiguration.class,
Gb32960AutoConfiguration.class,
EventHistoryAutoConfiguration.class))
.withAllowBeanDefinitionOverriding(true)
.withBean("kafkaProducer", KafkaProducer.class, VehicleHistoryAppCompositionTest::kafkaProducer)
.withPropertyValues(
"lingniu.ingest.sink.archive.enabled=true",
"lingniu.ingest.sink.archive.type=local",
"lingniu.ingest.sink.archive.path=" + tempDir.resolve("archive"),
"lingniu.ingest.event-file-store.enabled=true",
"lingniu.ingest.event-file-store.path=" + tempDir.resolve("event-store"),
"lingniu.ingest.event-history.enabled=true",
"lingniu.ingest.gb32960.enabled=true",
"lingniu.ingest.gb32960.server.enabled=false",
"lingniu.ingest.sink.mq.enabled=true",
"lingniu.ingest.sink.mq.type=kafka",
"lingniu.ingest.sink.mq.bootstrap-servers=localhost:9092",
"lingniu.ingest.sink.mq.consumer.enabled=false",
"lingniu.ingest.vehicle-state.enabled=false",
"lingniu.ingest.vehicle-stat.enabled=false")
.run(context -> {
assertThat(context).hasSingleBean(ArchiveStore.class);
assertThat(context).hasSingleBean(RawArchiveEventSink.class);
assertThat(context).hasSingleBean(EventFileStore.class);
assertThat(context).hasSingleBean(EventFileStoreSink.class);
assertThat(context).hasSingleBean(KafkaEventSink.class);
assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context).hasSingleBean(EventHistoryController.class);
assertThat(context).hasSingleBean(Gb32960MessageDecoder.class);
assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class);
assertThat(context).hasSingleBean(Gb32960FrameController.class);
assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
});
}
@SuppressWarnings("unchecked")
private static KafkaProducer<String, byte[]> kafkaProducer() {
return mock(KafkaProducer.class);
}
}

50
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java Normal file
View File

@@ -0,0 +1,50 @@
package com.lingniu.ingest.historyapp;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;
import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
class VehicleHistoryAppDefaultsTest {
@Test
void applicationDefaultsKeepHistoryAsDecoderConsumerAndStorageRuntime() {
Properties properties = applicationProperties();
assertThat(properties)
.containsEntry("spring.application.name", "vehicle-history-app")
.containsEntry("lingniu.ingest.gb32960.enabled", true)
.containsEntry("lingniu.ingest.gb32960.server.enabled", false)
.containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}")
.containsEntry("lingniu.ingest.event-file-store.enabled", "${EVENT_FILE_STORE_ENABLED:true}")
.containsEntry("lingniu.ingest.event-history.enabled", true)
.containsEntry("lingniu.ingest.vehicle-state.enabled", false)
.containsEntry("lingniu.ingest.vehicle-stat.enabled", false)
.containsEntry("lingniu.ingest.sink.mq.consumer.enabled", "${KAFKA_CONSUMER_ENABLED:true}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.enabled",
true)
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.group-id",
"${KAFKA_GROUP_HISTORY:vehicle-history}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.topics[1]",
"${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}");
}
private static Properties applicationProperties() {
YamlPropertiesFactoryBean factory = new YamlPropertiesFactoryBean();
factory.setResources(new ClassPathResource("application.yml"));
Properties properties = factory.getObject();
assertThat(properties).isNotNull();
return properties;
}
}

62
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java
View File

@@ -13,7 +13,12 @@ import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;
@@ -28,10 +33,14 @@ public final class DisruptorEventBus implements AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(DisruptorEventBus.class);
private final Disruptor<VehicleEventSlot> disruptor;
private final List<EventSink> sinks;
private final ExecutorService awaitExecutor;
private final AtomicLong published = new AtomicLong();
private final AtomicLong failed = new AtomicLong();
public DisruptorEventBus(int ringBufferSize, String waitStrategyName, List<EventSink> sinks) {
this.sinks = List.copyOf(sinks);
this.awaitExecutor = Executors.newVirtualThreadPerTaskExecutor();
ThreadFactory tf = Thread.ofVirtual().name("ingest-bus-", 0).factory();
this.disruptor = new Disruptor<>(
VehicleEventSlot::new,
@@ -40,14 +49,14 @@ public final class DisruptorEventBus implements AutoCloseable {
ProducerType.MULTI,
waitStrategy(waitStrategyName));
EventHandler<VehicleEventSlot>[] handlers = sinks.stream()
EventHandler<VehicleEventSlot>[] handlers = this.sinks.stream()
.map(this::toHandler)
.toArray(EventHandler[]::new);
// 每个 sink 一个独立 handler，事件按扇出模式同时写 Kafka、event-file-store 等目标。
disruptor.handleEventsWith(handlers);
disruptor.start();
log.info("DisruptorEventBus started ringBuffer={} wait={} sinks={}",
ringBufferSize, waitStrategyName, sinks.size());
ringBufferSize, waitStrategyName, this.sinks.size());
}
public void publish(VehicleEvent event) {
@@ -55,9 +64,36 @@ public final class DisruptorEventBus implements AutoCloseable {
published.incrementAndGet();
}
public CompletableFuture<Void> publishAndAwait(VehicleEvent event, String requiredSinkName) {
String requiredSink = Objects.requireNonNull(requiredSinkName, "requiredSinkName");
published.incrementAndGet();
List<EventSink> acceptingSinks = sinks.stream()
.filter(sink -> sink.accepts(event))
.toList();
List<EventSink> requiredSinks = acceptingSinks.stream()
.filter(sink -> requiredSink.equals(sink.name()))
.toList();
if (requiredSinks.isEmpty()) {
return CompletableFuture.failedFuture(new IllegalStateException(
"required sink '" + requiredSink + "' did not accept event " + event.eventId()));
}
List<CompletableFuture<Void>> requiredFutures = new ArrayList<>();
for (EventSink sink : acceptingSinks) {
CompletableFuture<Void> future = CompletableFuture
.supplyAsync(() -> publishToSinkAndTrack(sink, event), awaitExecutor)
.thenCompose(f -> f);
if (requiredSink.equals(sink.name())) {
requiredFutures.add(future);
}
}
return CompletableFuture.allOf(requiredFutures.toArray(CompletableFuture[]::new));
}
@Override
public void close() {
disruptor.shutdown();
awaitExecutor.shutdown();
log.info("DisruptorEventBus stopped published={} failed={}", published.get(), failed.get());
}
@@ -66,18 +102,28 @@ public final class DisruptorEventBus implements AutoCloseable {
VehicleEvent e = slot.event;
if (e == null || !sink.accepts(e)) return;
try {
sink.publish(e).exceptionally(ex -> {
// sink 异步失败只计数和打日志，不反向阻塞入站连接；生产侧靠 sink 自身重试/DLQ 保证可观测。
failed.incrementAndGet();
log.warn("sink {} publish failed", sink.name(), ex);
return null;
});
publishToSinkAndTrack(sink, e);
} finally {
if (endOfBatch) slot.clear();
}
};
}
private CompletableFuture<Void> publishToSinkAndTrack(EventSink sink, VehicleEvent event) {
try {
return sink.publish(event).whenComplete((ignored, ex) -> {
if (ex == null) return;
// sink 异步失败只计数和打日志；await 调用方仍会收到异常并决定是否 ACK。
failed.incrementAndGet();
log.warn("sink {} publish failed", sink.name(), ex);
});
} catch (Throwable t) {
failed.incrementAndGet();
log.warn("sink {} publish failed", sink.name(), t);
return CompletableFuture.failedFuture(t);
}
}
private static WaitStrategy waitStrategy(String name) {
return switch (name == null ? "yielding" : name.toLowerCase()) {
case "blocking" -> new BlockingWaitStrategy();

97
modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java
View File

@@ -10,11 +10,16 @@ import com.lingniu.ingest.core.pipeline.InterceptorChain;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
@@ -30,6 +35,7 @@ public final class Dispatcher {
private static final Logger log = LoggerFactory.getLogger(Dispatcher.class);
private static final AtomicLong RAW_ARCHIVE_SEQUENCE = new AtomicLong();
private static final String REQUIRED_DURABLE_SINK = "kafka";
private final HandlerRegistry registry;
private final InterceptorChain interceptors;
@@ -52,13 +58,67 @@ public final class Dispatcher {
public void dispatch(RawFrame frame) {
IngestContext ctx = new IngestContext(UUID.randomUUID().toString());
try {
// 在 interceptor 之前发 RawArchive，保证原始字节被无条件落盘（dedup/rate-limit
// 不会过滤它），满足"原始可回放"目标。archive 的写盘靠下游 ArchiveEventSink 消费。
RawArchiveLookup rawArchive = emitRawArchive(frame, ctx);
DispatchPlan plan = mapFrameToEvents(frame, ctx, false);
for (VehicleEvent e : plan.events()) {
eventBus.publish(e);
}
if (plan.failure() != null) {
throw plan.failure();
}
} catch (Throwable t) {
log.error("dispatch failure traceId={}", ctx.traceId(), t);
interceptors.onError(t, ctx);
}
}
public CompletableFuture<Void> dispatchAndAwait(RawFrame frame) {
IngestContext ctx = new IngestContext(UUID.randomUUID().toString());
try {
DispatchPlan plan = mapFrameToEvents(frame, ctx, true);
if (plan.events().isEmpty() && plan.failure() == null) {
return CompletableFuture.failedFuture(new IllegalStateException(
"awaited dispatch produced no events; required sink '" + REQUIRED_DURABLE_SINK + "' was not reached"));
}
CompletableFuture<?>[] futures = plan.events().stream()
.map(event -> eventBus.publishAndAwait(event, REQUIRED_DURABLE_SINK))
.toArray(CompletableFuture[]::new);
CompletableFuture<Void> boundary = CompletableFuture.allOf(futures);
if (plan.failure() == null) {
return boundary;
}
return boundary.handle((ignored, publishFailure) -> {
log.error("dispatch failure traceId={}", ctx.traceId(), plan.failure());
interceptors.onError(plan.failure(), ctx);
if (publishFailure != null) {
throw new CompletionException(publishFailure);
}
throw new CompletionException(plan.failure());
});
} catch (Throwable t) {
log.error("dispatch failure traceId={}", ctx.traceId(), t);
interceptors.onError(t, ctx);
return CompletableFuture.failedFuture(t);
}
}
public CompletableFuture<Void> dispatchAndAwait(RawFrame frame, Duration timeout) {
CompletableFuture<Void> future = dispatchAndAwait(frame);
if (timeout == null || timeout.isZero() || timeout.isNegative()) {
return future;
}
return future.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS);
}
private DispatchPlan mapFrameToEvents(RawFrame frame, IngestContext ctx, boolean awaitDurability) {
List<VehicleEvent> out = new ArrayList<>();
// 在 interceptor 之前发 RawArchive，保证原始字节被无条件落盘（dedup/rate-limit
// 不会过滤它），满足"原始可回放"目标。archive 的写盘靠下游 Sink 消费。
RawArchiveLookup rawArchive = appendRawArchive(frame, ctx, out);
try {
if (!interceptors.before(frame, ctx)) {
log.debug("frame aborted: {}", ctx.abortReason());
return;
return new DispatchPlan(out, null);
}
List<HandlerDefinition> handlers = registry.resolve(
@@ -67,11 +127,16 @@ public final class Dispatcher {
log.debug("no handler for {} cmd=0x{} info=0x{}",
frame.protocolId(), Integer.toHexString(frame.command()),
Integer.toHexString(frame.infoType()));
return;
return new DispatchPlan(out, null);
}
for (HandlerDefinition def : handlers) {
if (def.asyncBatch() != null) {
if (awaitDurability) {
throw new AwaitedAsyncBatchUnsupportedException(
"@AsyncBatch handler cannot be used with dispatchAndAwait until batch completion is awaited: "
+ def.method());
}
batchExecutor.submit(def, frame.payload(), event -> enrichWithRawArchive(event, rawArchive));
continue;
}
@@ -79,12 +144,15 @@ public final class Dispatcher {
for (VehicleEvent e : events) {
e = enrichWithRawArchive(e, rawArchive);
interceptors.after(e, ctx);
eventBus.publish(e);
out.add(e);
}
}
return new DispatchPlan(out, null);
} catch (Throwable t) {
log.error("dispatch failure traceId={}", ctx.traceId(), t);
interceptors.onError(t, ctx);
if (t instanceof AwaitedAsyncBatchUnsupportedException) {
return new DispatchPlan(List.of(), t);
}
return new DispatchPlan(out, t);
}
}
@@ -95,7 +163,7 @@ public final class Dispatcher {
* <p>VIN 取自 sourceMeta 里的 {@code vin} key（由各入站适配器负责填充）；缺失时
* 留空字符串，archive sink 会用 "unknown-vin" 占位保证 key 可解析。
*/
private RawArchiveLookup emitRawArchive(RawFrame frame, IngestContext ctx) {
private RawArchiveLookup appendRawArchive(RawFrame frame, IngestContext ctx, List<VehicleEvent> out) {
byte[] bytes = frame.rawBytes();
if (bytes == null || bytes.length == 0) return RawArchiveLookup.empty();
@@ -117,7 +185,7 @@ public final class Dispatcher {
frame.command(),
frame.infoType(),
bytes);
eventBus.publish(raw);
out.add(raw);
return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key));
}
@@ -182,4 +250,13 @@ public final class Dispatcher {
return key == null || key.isBlank();
}
}
private record DispatchPlan(List<VehicleEvent> events, Throwable failure) {
}
private static final class AwaitedAsyncBatchUnsupportedException extends IllegalStateException {
private AwaitedAsyncBatchUnsupportedException(String message) {
super(message);
}
}
}

81
modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/concurrency/DisruptorEventBusAwaitTest.java Normal file
View File

@@ -0,0 +1,81 @@
package com.lingniu.ingest.core.concurrency;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.sink.EventSink;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class DisruptorEventBusAwaitTest {
@Test
void publishAndAwaitDoesNotInvokeSinkPublishOnCallerThread() throws Exception {
long callerThreadId = Thread.currentThread().threadId();
CapturingSink sink = new CapturingSink("kafka");
try (DisruptorEventBus bus = new DisruptorEventBus(1024, "blocking", java.util.List.of(sink))) {
bus.publishAndAwait(event(), "kafka").get(3, TimeUnit.SECONDS);
}
assertThat(sink.publishThreadId.get()).isNotEqualTo(callerThreadId);
}
@Test
void publishAndAwaitDoesNotInvokeOptionalSinksWhenRequiredSinkIsAbsent() throws Exception {
CapturingSink optionalSink = new CapturingSink("event-file-store");
try (DisruptorEventBus bus = new DisruptorEventBus(1024, "blocking", List.of(optionalSink))) {
assertThatThrownBy(() -> bus.publishAndAwait(event(), "kafka").join())
.isInstanceOf(CompletionException.class)
.hasRootCauseInstanceOf(IllegalStateException.class)
.hasMessageContaining("kafka");
assertThat(optionalSink.published.await(200, TimeUnit.MILLISECONDS)).isFalse();
}
}
private static VehicleEvent.Heartbeat event() {
Instant now = Instant.parse("2026-06-23T10:00:00Z");
return new VehicleEvent.Heartbeat(
"heartbeat-1",
"VIN001",
ProtocolId.GB32960,
now,
now,
"trace-1",
Map.of());
}
private static final class CapturingSink implements EventSink {
private final String name;
private final AtomicLong publishThreadId = new AtomicLong(-1);
private final CountDownLatch published = new CountDownLatch(1);
private CapturingSink(String name) {
this.name = name;
}
@Override
public String name() {
return name;
}
@Override
public CompletableFuture<Void> publish(VehicleEvent event) {
publishThreadId.set(Thread.currentThread().threadId());
published.countDown();
return CompletableFuture.completedFuture(null);
}
}
}

217
modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherDurableAckBoundaryTest.java Normal file
View File

@@ -0,0 +1,217 @@
package com.lingniu.ingest.core.dispatcher;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.annotation.AsyncBatch;
import com.lingniu.ingest.api.annotation.MessageMapping;
import com.lingniu.ingest.api.annotation.ProtocolHandler;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.pipeline.IngestContext;
import com.lingniu.ingest.api.pipeline.IngestInterceptor;
import com.lingniu.ingest.api.pipeline.RawFrame;
import com.lingniu.ingest.api.sink.EventSink;
import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor;
import com.lingniu.ingest.core.concurrency.DisruptorEventBus;
import com.lingniu.ingest.core.pipeline.InterceptorChain;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Method;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class DispatcherDurableAckBoundaryTest {
@Test
void dispatchAndAwaitFailsWhenNoKafkaSinkAcceptsEvent() throws Exception {
EventSink fileSink = new NamedSink("event-file-store", true);
try (Harness harness = new Harness(registryWithHeartbeatHandler(), List.of(fileSink), List.of())) {
CompletionStage<Void> future = harness.dispatcher.dispatchAndAwait(frame(0x07, "payload"));
assertThatThrownBy(() -> future.toCompletableFuture().join())
.isInstanceOf(CompletionException.class)
.hasRootCauseInstanceOf(IllegalStateException.class)
.hasMessageContaining("kafka");
}
}
@Test
void dispatchAndAwaitNotifiesInterceptorWhenPlanCapturesFailure() throws Exception {
CapturingInterceptor interceptor = new CapturingInterceptor();
try (Harness harness = new Harness(registryWithThrowingHandler(), List.of(new NamedSink("kafka", true)),
List.of(interceptor))) {
CompletionStage<Void> future = harness.dispatcher.dispatchAndAwait(frame(0x08, "payload"));
assertThatThrownBy(() -> future.toCompletableFuture().join())
.isInstanceOf(CompletionException.class)
.hasRootCauseInstanceOf(IllegalStateException.class)
.rootCause()
.hasMessageContaining("handler failed");
assertThat(interceptor.error.get())
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("Handler threw exception");
}
}
@Test
void dispatchAndAwaitFailsFastForAsyncBatchHandler() throws Exception {
NamedSink kafka = new NamedSink("kafka", true);
try (Harness harness = new Harness(registryWithAsyncBatchHandler(), List.of(kafka), List.of())) {
CompletionStage<Void> future = harness.dispatcher.dispatchAndAwait(
frame(0x09, "payload", new byte[]{0x23, 0x23, 0x09}));
assertThatThrownBy(() -> future.toCompletableFuture().join())
.isInstanceOf(CompletionException.class)
.hasRootCauseInstanceOf(IllegalStateException.class)
.hasMessageContaining("@AsyncBatch");
assertThat(kafka.published.get()).isZero();
}
}
private static HandlerRegistry registryWithHeartbeatHandler() throws NoSuchMethodException {
HeartbeatHandler bean = new HeartbeatHandler();
Method method = HeartbeatHandler.class.getDeclaredMethod("onHeartbeat", Object.class);
return registry(bean, method, 0x07, method.getAnnotation(AsyncBatch.class));
}
private static HandlerRegistry registryWithThrowingHandler() throws NoSuchMethodException {
ThrowingHandler bean = new ThrowingHandler();
Method method = ThrowingHandler.class.getDeclaredMethod("onFrame", Object.class);
return registry(bean, method, 0x08, method.getAnnotation(AsyncBatch.class));
}
private static HandlerRegistry registryWithAsyncBatchHandler() throws NoSuchMethodException {
AsyncHandler bean = new AsyncHandler();
Method method = AsyncHandler.class.getDeclaredMethod("onBatch", List.class);
return registry(bean, method, 0x09, method.getAnnotation(AsyncBatch.class));
}
private static HandlerRegistry registry(Object bean, Method method, int command, AsyncBatch asyncBatch) {
HandlerRegistry registry = new HandlerRegistry();
registry.register(new HandlerDefinition(
ProtocolId.GB32960,
command,
0,
method.getName(),
bean,
method,
Object.class,
null,
null,
asyncBatch));
return registry;
}
private static RawFrame frame(int command, Object payload) {
return frame(command, payload, null);
}
private static RawFrame frame(int command, Object payload, byte[] rawBytes) {
return new RawFrame(
ProtocolId.GB32960,
command,
0,
payload,
rawBytes,
Map.of("vin", "VIN001"),
Instant.parse("2026-06-23T10:00:00Z"));
}
private static VehicleEvent.Heartbeat event(String id) {
Instant now = Instant.parse("2026-06-23T10:00:00Z");
return new VehicleEvent.Heartbeat(id, "VIN001", ProtocolId.GB32960, now, now, "trace-" + id, Map.of());
}
private static final class Harness implements AutoCloseable {
private final DisruptorEventBus eventBus;
private final AsyncBatchExecutor batchExecutor;
private final Dispatcher dispatcher;
private Harness(HandlerRegistry registry, List<EventSink> sinks, List<IngestInterceptor> interceptors) {
this.eventBus = new DisruptorEventBus(1024, "blocking", sinks);
this.batchExecutor = new AsyncBatchExecutor(eventBus::publish);
this.dispatcher = new Dispatcher(
registry,
new InterceptorChain(interceptors),
new HandlerInvoker(),
eventBus,
batchExecutor);
}
@Override
public void close() {
batchExecutor.close();
eventBus.close();
}
}
private static final class NamedSink implements EventSink {
private final String name;
private final boolean accepts;
private final AtomicInteger published = new AtomicInteger();
private NamedSink(String name, boolean accepts) {
this.name = name;
this.accepts = accepts;
}
@Override
public String name() {
return name;
}
@Override
public CompletableFuture<Void> publish(VehicleEvent event) {
published.incrementAndGet();
return CompletableFuture.completedFuture(null);
}
@Override
public boolean accepts(VehicleEvent event) {
return accepts;
}
}
private static final class CapturingInterceptor implements IngestInterceptor {
private final AtomicReference<Throwable> error = new AtomicReference<>();
@Override
public void onError(Throwable error, IngestContext ctx) {
this.error.set(error);
}
}
@ProtocolHandler(protocol = ProtocolId.GB32960)
static final class HeartbeatHandler {
@MessageMapping(command = 0x07)
VehicleEvent.Heartbeat onHeartbeat(Object ignored) {
return event("heartbeat-1");
}
}
@ProtocolHandler(protocol = ProtocolId.GB32960)
static final class ThrowingHandler {
@MessageMapping(command = 0x08)
VehicleEvent.Heartbeat onFrame(Object ignored) {
throw new IllegalStateException("handler failed");
}
}
@ProtocolHandler(protocol = ProtocolId.GB32960)
public static final class AsyncHandler {
@MessageMapping(command = 0x09)
@AsyncBatch(size = 2, waitMs = 10, poolSize = 1)
public List<VehicleEvent> onBatch(List<Object> ignored) {
return List.of(event("async-1"));
}
}
}

5
modules/protocols/protocol-gb32960/pom.xml
View File

@@ -55,5 +55,10 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

6
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java
View File

@@ -59,8 +59,8 @@ import java.util.List;
* <p>每个 Parser Bean 都会被 {@link InfoBlockParserRegistry} 按
* {@code (ProtocolVersion, typeCode)} 注册。2016/2025 可同时加载。
*
* <p>只有 {@code lingniu.ingest.gb32960.enabled=true} 时才会启动 32960 TCP 服务。
* 解析、鉴权、ACK、RAW 入库都从 {@link Gb32960NettyServer} 建立的 Netty pipeline 进入。
* <p>只有 {@code lingniu.ingest.gb32960.enabled=true} 时才会装配 32960 协议解析组件。
* TCP 监听由 {@code lingniu.ingest.gb32960.server.enabled} 单独控制。
*/
@AutoConfiguration
@EnableConfigurationProperties(Gb32960Properties.class)
@@ -222,6 +222,8 @@ public class Gb32960AutoConfiguration {
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "lingniu.ingest.gb32960.server", name = "enabled",
havingValue = "true", matchIfMissing = true)
public Gb32960NettyServer gb32960NettyServer(Gb32960Properties props,
Gb32960MessageDecoder decoder,
Dispatcher dispatcher,

13
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960Properties.java
View File

@@ -10,8 +10,10 @@ import java.util.Set;
@ConfigurationProperties(prefix = "lingniu.ingest.gb32960")
public class Gb32960Properties {
/** 是否启动 32960 TCP 监听。生产 32960 线路打开后，此开关决定端口是否真正 bind。 */
/** 是否装配 32960 协议解析组件。TCP 监听由 {@link #server} 单独控制。 */
private boolean enabled = false;
/** 32960 TCP 监听开关。默认跟随协议启用后启动，查询类应用可单独关闭。 */
private Server server = new Server();
/** 32960 TCP 监听端口；当前生产验证线使用 32960。 */
private int port = 9000;
/** Netty worker 线程数；0 表示按 Netty/CPU 默认策略分配。 */
@@ -63,6 +65,8 @@ public class Gb32960Properties {
public boolean isEnabled() { return enabled; }
public void setEnabled(boolean enabled) { this.enabled = enabled; }
public Server getServer() { return server; }
public void setServer(Server server) { this.server = server; }
public int getPort() { return port; }
public void setPort(int port) { this.port = port; }
public int getWorkerThreads() { return workerThreads; }
@@ -82,6 +86,13 @@ public class Gb32960Properties {
this.vendorExtensions = vendorExtensions;
}
public static class Server {
private boolean enabled = true;
public boolean isEnabled() { return enabled; }
public void setEnabled(boolean enabled) { this.enabled = enabled; }
}
/**
* VIN 白名单认证。
*

173
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java
View File

@@ -20,6 +20,7 @@ import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.time.Instant;
import java.util.concurrent.CompletableFuture;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -55,6 +56,9 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
public static final AttributeKey<String> PLATFORM_ACCOUNT_ATTR =
Gb32960AccessService.PLATFORM_ACCOUNT_ATTR;
private static final AttributeKey<CompletableFuture<Void>> DURABLE_ACK_CHAIN_ATTR =
AttributeKey.valueOf("gb32960.durableAckChain");
private final Gb32960MessageDecoder decoder;
private final Dispatcher dispatcher;
private final Gb32960AccessService accessService;
@@ -122,41 +126,137 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
return;
}
// 鉴权通过：按命令类型回 ACK，写成功后统一 dispatch 到通用管线。
// 鉴权通过：先完成命令侧状态变更/日志，再统一 dispatch 到通用管线。
// 需要成功 ACK 的已接收帧，等待配置的 Kafka dispatch 边界完成后再回 ACK。
// 平台登入鉴权失败会短路 return，不进 dispatch。
switch (cmd) {
case VEHICLE_LOGIN -> handleVehicleLogin(ctx, msg, rawVin);
case VEHICLE_LOGOUT -> handleVehicleLogout(ctx, msg, rawVin);
case VEHICLE_LOGIN -> logVehicleLogin(ctx, msg);
case VEHICLE_LOGOUT -> logVehicleLogout(ctx, msg);
case PLATFORM_LOGIN -> {
if (!handlePlatformLogin(ctx, msg, rawVin)) return;
}
case PLATFORM_LOGOUT -> handlePlatformLogout(ctx, msg, rawVin);
case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> handleReportOrHeartbeat(ctx, msg, rawVin);
case TIME_CALIBRATION -> handleTimeCalibration(ctx, msg, rawVin);
case PLATFORM_LOGOUT -> logPlatformLogout(ctx);
case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> logReportOrHeartbeat(ctx, msg);
case TIME_CALIBRATION -> { }
default -> logOtherFrame(ctx, msg);
}
RawFrame rf = rawFrame(ctx, msg, frame);
if (!requiresDurableAck(cmd)) {
dispatcher.dispatch(rf);
return;
}
enqueueDurableAck(ctx, msg, rawVin, vin, cmd, dispatcher.dispatchAndAwait(rf));
}
private void enqueueDurableAck(ChannelHandlerContext ctx,
Gb32960Message msg,
byte[] rawVin,
String vin,
CommandType cmd,
CompletableFuture<Void> dispatchFuture) {
CompletableFuture<Void> previous = ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).get();
if (previous == null) {
previous = CompletableFuture.completedFuture(null);
}
CompletableFuture<Void> next = previous
.handle((ignored, previousFailure) -> null)
.thenCompose(ignored -> dispatchFuture.handle((ok, failure) -> failure))
.thenCompose(failure -> runOnEventLoop(ctx,
() -> handleDurableAckResult(ctx, msg, rawVin, vin, cmd, failure)));
ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).set(next);
}
private CompletableFuture<Void> runOnEventLoop(ChannelHandlerContext ctx, Runnable task) {
CompletableFuture<Void> done = new CompletableFuture<>();
Runnable wrapped = () -> {
try {
task.run();
done.complete(null);
} catch (Throwable t) {
done.completeExceptionally(t);
}
};
if (ctx.executor().inEventLoop()) {
wrapped.run();
} else {
ctx.executor().execute(wrapped);
}
return done;
}
private void handleDurableAckResult(ChannelHandlerContext ctx,
Gb32960Message msg,
byte[] rawVin,
String vin,
CommandType cmd,
Throwable failure) {
if (!ctx.channel().isActive()) {
return;
}
if (failure != null) {
log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}",
addr(ctx), vin, Integer.toHexString(cmd.code()), failure);
ctx.close();
return;
}
writeAckForCommand(ctx, msg, rawVin);
}
private RawFrame rawFrame(ChannelHandlerContext ctx, Gb32960Message msg, byte[] frame) {
Map<String, String> sourceMeta = new HashMap<>(4);
sourceMeta.put("vin", vin);
sourceMeta.put("vin", msg.header().vin());
sourceMeta.put("peer", addr(ctx));
String platformAccount = accessService.platformAccount(ctx.channel());
if (platformAccount != null && !platformAccount.isBlank()) {
sourceMeta.put("platformAccount", platformAccount);
}
// RawFrame.rawBytes 是后续 raw archive 和按 rawArchiveUri 回查的源头；不要在这里裁剪或重编码。
RawFrame rf = new RawFrame(
return new RawFrame(
ProtocolId.GB32960,
cmd.code(),
msg.header().command().code(),
0,
msg,
frame,
sourceMeta,
Instant.now());
dispatcher.dispatch(rf);
}
private static boolean requiresDurableAck(CommandType cmd) {
return switch (cmd) {
case VEHICLE_LOGIN,
VEHICLE_LOGOUT,
PLATFORM_LOGIN,
PLATFORM_LOGOUT,
REALTIME_REPORT,
RESEND_REPORT,
HEARTBEAT,
TIME_CALIBRATION -> true;
default -> false;
};
}
private void writeAckForCommand(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
switch (msg.header().command()) {
case VEHICLE_LOGIN -> writeVehicleLoginAck(ctx, msg, rawVin);
case VEHICLE_LOGOUT -> writeVehicleLogoutAck(ctx, msg, rawVin);
case PLATFORM_LOGIN -> writePlatformLoginAck(ctx, msg, rawVin);
case PLATFORM_LOGOUT -> writePlatformLogoutAck(ctx, msg, rawVin);
case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> writeReportOrHeartbeatAck(ctx, msg, rawVin);
case TIME_CALIBRATION -> writeTimeCalibrationAck(ctx, msg, rawVin);
default -> { }
}
}
/** 0x01 车辆登入：按 §6.3.2 带原采集时间回成功 ACK + INFO 日志。 */
private void handleVehicleLogin(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
private void logVehicleLogin(ChannelHandlerContext ctx, Gb32960Message msg) {
log.info("[gb32960] vehicle login peer={} vin={} protocolVersion={}",
addr(ctx), msg.header().vin(), msg.header().protocolVersion());
}
private void writeVehicleLoginAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
ackService.writeResponse(
ctx,
msg.header().protocolVersion(),
@@ -166,13 +266,14 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
eventOrNow(msg),
null,
"vehicle-login-ack");
log.info("[gb32960] vehicle login peer={} vin={} protocolVersion={}",
addr(ctx), msg.header().vin(), msg.header().protocolVersion());
}
/** 0x04 车辆登出：INFO 日志 + 带原采集时间回成功 ACK。 */
private void handleVehicleLogout(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
private void logVehicleLogout(ChannelHandlerContext ctx, Gb32960Message msg) {
log.info("[gb32960] vehicle logout peer={} vin={}", addr(ctx), msg.header().vin());
}
private void writeVehicleLogoutAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
ackService.writeResponse(ctx,
msg.header().protocolVersion(),
CommandType.VEHICLE_LOGOUT,
@@ -189,8 +290,8 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
* <ul>
* <li>消息体解析缺失：关闭连接，返回 false 短路
* <li>鉴权失败：写 {@code 0x02 OTHER_ERROR} NACK + 关闭连接，返回 false 短路
* <li>鉴权通过：把 username 钉到 channel attribute（供后续 vendor profile 路由），
* 写成功 ACK + INFO 日志，返回 true 继续 dispatch
* <li>鉴权通过：把 username 钉到 channel attribute（供后续 vendor profile 路由），INFO 日志，
* 返回 true 继续 dispatch；成功 ACK 等 dispatch 持久化边界完成后再写
* </ul>
*
* @return true 表示鉴权通过应继续 dispatch；false 表示调用方应 short-circuit return
@@ -221,6 +322,10 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
addr(ctx), pl.username(), pl.encryptRule(), pl.serialNo(), pl.loginTime(), result);
// 把 username 钉到 channel attribute，供后续 0x02/0x03 帧的 vendor profile 路由
accessService.bindPlatformAccount(ctx.channel(), pl.username());
return true;
}
private void writePlatformLoginAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
ackService.writeResponse(
ctx,
msg.header().protocolVersion(),
@@ -230,12 +335,14 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
eventOrNow(msg),
null,
"platform-login-ack");
return true;
}
/** 0x06 平台登出：INFO 日志 + 带原采集时间回成功 ACK。 */
private void handlePlatformLogout(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
private void logPlatformLogout(ChannelHandlerContext ctx) {
log.info("[gb32960] platform logout peer={}", addr(ctx));
}
private void writePlatformLogoutAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
ackService.writeResponse(ctx,
msg.header().protocolVersion(),
CommandType.PLATFORM_LOGOUT,
@@ -247,11 +354,11 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
}
/**
* 0x02/0x03/0x07 实时 / 补发 / 心跳：按 §6.3.2 回应答帧，保留原帧采集时间。
* 0x02/0x03/0x07 实时 / 补发 / 心跳：记录帧诊断；成功 ACK 在配置的 Kafka dispatch 边界后写回。
* 部分车端实现严格按规范，没收到应答会停止后续上报或重连。ack-tag 按 cmd.code 区分便于日志筛选。
*
* <p>注意：应答成功不代表下游已完成持久化，只表示平台已接收并通过 Dispatcher 投递。
* 生产排查存储问题时，应继续检查 raw archive sink、EventFileStoreSink 和 Kafka sink 的日志。
* <p>注意：应答成功表示当前配置的 Kafka sink 已完成该帧映射事件的 dispatch 边界。
* 生产排查其他存储问题时，应继续检查 raw archive sink、EventFileStoreSink 等可选 sink 的日志。
*
* <p>日志策略（按 (channel, vin, rawTypeSignature) 去重，避免高频帧刷屏）：
* <ul>
@@ -264,17 +371,9 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
* </ul>
* 帧内字段全量 JSON 仍保留 DEBUG。
*/
private void handleReportOrHeartbeat(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
private void logReportOrHeartbeat(ChannelHandlerContext ctx, Gb32960Message msg) {
CommandType cmd = msg.header().command();
String platformAccount = accessService.platformAccount(ctx.channel());
ackService.writeResponse(ctx,
msg.header().protocolVersion(),
cmd,
ResponseFlag.SUCCESS,
rawVin,
eventOrNow(msg),
null,
"report-ack-0x" + Integer.toHexString(cmd.code()));
if (cmd == CommandType.HEARTBEAT) {
if (log.isDebugEnabled()) {
log.debug("[gb32960] HEARTBEAT peer={} vin={} platformAccount={}",
@@ -304,8 +403,20 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
}
}
private void writeReportOrHeartbeatAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
CommandType cmd = msg.header().command();
ackService.writeResponse(ctx,
msg.header().protocolVersion(),
cmd,
ResponseFlag.SUCCESS,
rawVin,
eventOrNow(msg),
null,
"report-ack-0x" + Integer.toHexString(cmd.code()));
}
/** 0x08 终端校时：平台应答 data 段为平台当前时间 6B（用 Instant.now()）。 */
private void handleTimeCalibration(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
private void writeTimeCalibrationAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
ackService.writeResponse(ctx,
msg.header().protocolVersion(),
CommandType.TIME_CALIBRATION,

37
modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfigurationTest.java Normal file
View File

@@ -0,0 +1,37 @@
package com.lingniu.ingest.protocol.gb32960.config;
import com.lingniu.ingest.core.dispatcher.Dispatcher;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import static org.assertj.core.api.Assertions.assertThat;
class Gb32960AutoConfigurationTest {
private final ApplicationContextRunner runner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(Gb32960AutoConfiguration.class))
.withBean(Dispatcher.class, () -> new Dispatcher(null, null, null, null, null))
.withPropertyValues(
"lingniu.ingest.gb32960.enabled=true",
"lingniu.ingest.gb32960.port=0");
@Test
void createsDecoderWithoutTcpServerWhenServerDisabled() {
runner.withPropertyValues("lingniu.ingest.gb32960.server.enabled=false")
.run(context -> {
assertThat(context).hasSingleBean(Gb32960MessageDecoder.class);
assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
});
}
@Test
void createsTcpServerByDefaultWhenProtocolEnabled() {
runner.run(context -> {
assertThat(context).hasSingleBean(Gb32960MessageDecoder.class);
assertThat(context).hasSingleBean(Gb32960NettyServer.class);
});
}
}

212
modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java Normal file
View File

@@ -0,0 +1,212 @@
package com.lingniu.ingest.protocol.gb32960.inbound;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.sink.EventSink;
import com.lingniu.ingest.codec.BccChecksum;
import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor;
import com.lingniu.ingest.core.concurrency.DisruptorEventBus;
import com.lingniu.ingest.core.dispatcher.Dispatcher;
import com.lingniu.ingest.core.dispatcher.HandlerInvoker;
import com.lingniu.ingest.core.dispatcher.HandlerRegistry;
import com.lingniu.ingest.core.pipeline.InterceptorChain;
import com.lingniu.ingest.protocol.gb32960.auth.Gb32960PlatformAuthorizer;
import com.lingniu.ingest.protocol.gb32960.auth.Gb32960VinAuthorizer;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960BodyParser;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960DecoderTest;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.codec.InfoBlockParserRegistry;
import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.PositionV2016BlockParser;
import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.VehicleV2016BlockParser;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960Properties;
import io.netty.buffer.ByteBuf;
import io.netty.channel.embedded.EmbeddedChannel;
import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import static org.assertj.core.api.Assertions.assertThat;
class Gb32960ChannelHandlerAckBoundaryTest {
@Test
void realtimeReportAckWaitsForDispatchFutureCompletion() {
try (DispatchHarness dispatch = new DispatchHarness()) {
EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher()));
assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse();
assertThat((Object) channel.readOutbound()).isNull();
dispatch.awaitPublishCount(1);
dispatch.completeDurability();
channel.runPendingTasks();
ByteBuf ack = channel.readOutbound();
assertThat(ack).isNotNull();
assertThat(ack.getUnsignedByte(2)).isEqualTo((short) 0x02);
assertThat(ack.getUnsignedByte(3)).isEqualTo((short) 0x01);
ack.release();
channel.finishAndReleaseAll();
}
}
@Test
void realtimeReportAckIsNotWrittenWhenDispatchFutureFails() {
try (DispatchHarness dispatch = new DispatchHarness()) {
EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher()));
assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse();
dispatch.awaitPublishCount(1);
dispatch.failDurability();
channel.runPendingTasks();
assertThat((Object) channel.readOutbound()).isNull();
assertThat(channel.isActive()).isFalse();
channel.finishAndReleaseAll();
}
}
@Test
void durableAcksAreWrittenInInboundOrderWhenLaterDispatchCompletesFirst() {
try (DispatchHarness dispatch = new DispatchHarness()) {
EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher()));
assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse();
assertThat(channel.writeInbound(buildHeartbeatFrame("LTEST000000000001"))).isFalse();
assertThat((Object) channel.readOutbound()).isNull();
dispatch.awaitPublishCount(2);
dispatch.completeDurability(1);
channel.runPendingTasks();
assertThat((Object) channel.readOutbound()).isNull();
dispatch.completeDurability(0);
channel.runPendingTasks();
ByteBuf firstAck = channel.readOutbound();
ByteBuf secondAck = channel.readOutbound();
assertThat(firstAck).isNotNull();
assertThat(secondAck).isNotNull();
assertThat(firstAck.getUnsignedByte(2)).isEqualTo((short) 0x02);
assertThat(secondAck.getUnsignedByte(2)).isEqualTo((short) 0x07);
firstAck.release();
secondAck.release();
channel.finishAndReleaseAll();
}
}
private static Gb32960ChannelHandler handlerWith(Dispatcher dispatcher) {
return new Gb32960ChannelHandler(
decoder(),
dispatcher,
accessService(),
new Gb32960AckService(),
new Gb32960FrameDiagnostics(16));
}
private static Gb32960MessageDecoder decoder() {
return new Gb32960MessageDecoder(new Gb32960BodyParser(new InfoBlockParserRegistry(List.of(
new VehicleV2016BlockParser(),
new PositionV2016BlockParser()))));
}
private static Gb32960AccessService accessService() {
Gb32960Properties.Auth auth = new Gb32960Properties.Auth();
auth.setEnabled(false);
return new Gb32960AccessService(
new Gb32960VinAuthorizer(auth),
new Gb32960PlatformAuthorizer(auth.getPlatforms()));
}
private static byte[] buildHeartbeatFrame(String vin) {
ByteArrayOutputStream out = new ByteArrayOutputStream();
out.write(0x23);
out.write(0x23);
out.write(0x07);
out.write(0xFE);
byte[] vinBytes = vin.getBytes(StandardCharsets.US_ASCII);
out.write(vinBytes, 0, 17);
out.write(0x01);
out.write(0x00);
out.write(0x00);
byte[] almost = out.toByteArray();
out.write(BccChecksum.compute(almost, 2, almost.length - 2) & 0xFF);
return out.toByteArray();
}
private static final class DispatchHarness implements AutoCloseable {
private final ControlledSink sink = new ControlledSink();
private final DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink));
private final AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
private final Dispatcher dispatcher = new Dispatcher(
new HandlerRegistry(),
new InterceptorChain(List.of()),
new HandlerInvoker(),
eventBus,
batchExecutor);
Dispatcher dispatcher() {
return dispatcher;
}
void completeDurability() {
completeDurability(0);
}
void completeDurability(int index) {
sink.future(index).complete(null);
}
void failDurability() {
sink.future(0).completeExceptionally(new IllegalStateException("durability failed"));
}
void awaitPublishCount(int count) {
long deadline = System.currentTimeMillis() + 3000;
while (System.currentTimeMillis() < deadline) {
synchronized (sink.futures) {
if (sink.futures.size() >= count) {
return;
}
}
Thread.onSpinWait();
}
throw new AssertionError("expected " + count + " sink publishes, actual=" + sink.futures.size());
}
@Override
public void close() {
batchExecutor.close();
eventBus.close();
}
}
private static final class ControlledSink implements EventSink {
private final List<CompletableFuture<Void>> futures = new ArrayList<>();
@Override
public String name() {
return "kafka";
}
@Override
public CompletableFuture<Void> publish(VehicleEvent event) {
CompletableFuture<Void> future = new CompletableFuture<>();
synchronized (futures) {
futures.add(future);
}
return future;
}
private CompletableFuture<Void> future(int index) {
synchronized (futures) {
return futures.get(index);
}
}
}
}

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

@@ -1,7 +1,11 @@
package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.eventfilestore.EventFileRecord;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
@@ -21,20 +25,24 @@ import java.util.Map;
public final class TelemetryEnvelopeRecordMapper {
private static final JsonFormat.Printer JSON_PRINTER = JsonFormat.printer();
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
public EventFileRecord toRecord(VehicleEnvelope envelope) {
if (envelope == null) {
throw new IllegalArgumentException("envelope must not be null");
}
if (!envelope.hasTelemetrySnapshot()) {
throw new IllegalArgumentException("envelope telemetry_snapshot is required");
if (envelope.hasTelemetrySnapshot()) {
return telemetryRecord(envelope);
}
if (envelope.hasRawArchive()) {
return rawArchiveRecord(envelope);
}
throw new IllegalArgumentException("envelope telemetry_snapshot or raw_archive is required");
}
private static EventFileRecord telemetryRecord(VehicleEnvelope envelope) {
ProtocolId protocol = protocol(envelope.getSource());
Map<String, String> metadata = new LinkedHashMap<>(envelope.getMetadataMap());
if (protocol == ProtocolId.UNKNOWN && !envelope.getSource().isBlank()) {
// 保留未知 source 原文，避免历史库标准化成 UNKNOWN 后丢失排障线索。
metadata.putIfAbsent("originalSource", envelope.getSource());
}
Map<String, String> metadata = metadata(envelope, protocol);
return new EventFileRecord(
envelope.getEventId(),
protocol,
@@ -48,6 +56,57 @@ public final class TelemetryEnvelopeRecordMapper {
);
}
private static EventFileRecord rawArchiveRecord(VehicleEnvelope envelope) {
ProtocolId protocol = protocol(envelope.getSource());
RawArchiveRef rawArchive = envelope.getRawArchive();
Map<String, String> metadata = metadata(envelope, protocol);
String rawArchiveKey = rawArchiveKey(rawArchive.getUri(), metadata);
String rawArchiveUri = rawArchive.getUri().isBlank() && !rawArchiveKey.isBlank()
? RawArchiveKeys.logicalUri(rawArchiveKey)
: rawArchive.getUri();
metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, envelope.getEventId());
if (!rawArchiveKey.isBlank()) {
metadata.putIfAbsent(RawArchiveKeys.META_KEY, rawArchiveKey);
}
if (!rawArchiveUri.isBlank()) {
metadata.putIfAbsent(RawArchiveKeys.META_URI, rawArchiveUri);
}
return new EventFileRecord(
envelope.getEventId(),
protocol,
"RAW_ARCHIVE",
envelope.getVin(),
Instant.ofEpochMilli(envelope.getEventTimeMs()),
Instant.ofEpochMilli(envelope.getIngestTimeMs()),
rawArchiveUri,
metadata,
rawArchiveJson(envelope, metadata, rawArchiveKey, rawArchiveUri)
);
}
private static String rawArchiveKey(String rawArchiveUri, Map<String, String> metadata) {
String key = metadata.getOrDefault(RawArchiveKeys.META_KEY, "");
if (!key.isBlank()) {
return key;
}
if (rawArchiveUri != null && rawArchiveUri.startsWith("archive://")) {
return rawArchiveUri.substring("archive://".length());
}
return "";
}
private static Map<String, String> metadata(VehicleEnvelope envelope, ProtocolId protocol) {
Map<String, String> metadata = new LinkedHashMap<>(envelope.getMetadataMap());
if (!envelope.getProtocolVersion().isBlank()) {
metadata.putIfAbsent("protocolVersion", envelope.getProtocolVersion());
}
if (protocol == ProtocolId.UNKNOWN && !envelope.getSource().isBlank()) {
// 保留未知 source 原文，避免历史库标准化成 UNKNOWN 后丢失排障线索。
metadata.putIfAbsent("originalSource", envelope.getSource());
}
return metadata;
}
private static ProtocolId protocol(String source) {
if (source == null || source.isBlank()) {
return ProtocolId.UNKNOWN;
@@ -66,4 +125,36 @@ public final class TelemetryEnvelopeRecordMapper {
throw new IllegalArgumentException("failed to serialize telemetry_snapshot", e);
}
}
private static String rawArchiveJson(VehicleEnvelope envelope,
Map<String, String> metadata,
String rawArchiveKey,
String rawArchiveUri) {
Map<String, Object> payload = new LinkedHashMap<>();
RawArchiveRef rawArchive = envelope.getRawArchive();
payload.put("eventId", envelope.getEventId());
payload.put("vin", envelope.getVin());
payload.put("protocol", protocol(envelope.getSource()).name());
if (!envelope.getSource().isBlank()) {
payload.put("source", envelope.getSource());
}
payload.put("eventType", "RAW_ARCHIVE");
payload.put("eventTime", Instant.ofEpochMilli(envelope.getEventTimeMs()).toString());
payload.put("ingestTime", Instant.ofEpochMilli(envelope.getIngestTimeMs()).toString());
payload.put("rawArchiveUri", rawArchiveUri);
payload.put("rawArchiveKey", rawArchiveKey);
if (metadata.containsKey("command")) {
payload.put("command", metadata.get("command"));
}
if (metadata.containsKey("infoType")) {
payload.put("infoType", metadata.get("infoType"));
}
payload.put("rawSizeBytes", rawArchive.getSizeBytes());
payload.put("metadata", metadata);
try {
return OBJECT_MAPPER.writeValueAsString(payload);
} catch (JsonProcessingException e) {
throw new IllegalArgumentException("failed to serialize raw_archive", e);
}
}
}

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

@@ -1,16 +1,22 @@
package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.eventfilestore.EventFileQuery;
import com.lingniu.ingest.eventfilestore.EventFileRecord;
import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
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.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
@@ -19,6 +25,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
class EventHistoryEnvelopeIngestorTest {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
@Test
void parsesKafkaEnvelopeBytesAndAppendsRecord() throws Exception {
CapturingStore store = new CapturingStore();
@@ -55,6 +63,61 @@ class EventHistoryEnvelopeIngestorTest {
assertThat(store.records).isEmpty();
}
@Test
void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() throws Exception {
CapturingStore store = new CapturingStore();
EventHistoryEnvelopeIngestor ingestor =
new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
String rawArchiveKey = "2026/06/23/GB32960/VINRAW001/raw-event-1.bin";
String rawArchiveUri = "archive://" + rawArchiveKey;
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
.setEventId("raw-event-1")
.setVin("VINRAW001")
.setSource("GB32960")
.setProtocolVersion("V2016")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.putMetadata(RawArchiveKeys.META_KEY, rawArchiveKey)
.putMetadata(RawArchiveKeys.META_URI, rawArchiveUri)
.putMetadata("command", "0x0002")
.setRawArchive(RawArchiveRef.newBuilder()
.setUri(rawArchiveUri)
.setSizeBytes(128)
.build())
.build();
EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
assertThat(store.records).hasSize(1);
EventFileRecord record = store.records.getFirst();
assertThat(record.eventId()).isEqualTo("raw-event-1");
assertThat(record.protocol()).isEqualTo(ProtocolId.GB32960);
assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
assertThat(record.vin()).isEqualTo("VINRAW001");
assertThat(record.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_112_400_000L));
assertThat(record.rawArchiveUri()).isEqualTo(rawArchiveUri);
assertThat(record.metadata())
.containsEntry("protocolVersion", "V2016")
.containsEntry(RawArchiveKeys.META_EVENT_ID, "raw-event-1")
.containsEntry(RawArchiveKeys.META_KEY, rawArchiveKey)
.containsEntry(RawArchiveKeys.META_URI, rawArchiveUri)
.containsEntry("command", "0x0002");
JsonNode payload = OBJECT_MAPPER.readTree(record.payloadJson());
assertThat(payload.get("eventId").asText()).isEqualTo("raw-event-1");
assertThat(payload.get("protocol").asText()).isEqualTo("GB32960");
assertThat(payload.get("eventType").asText()).isEqualTo("RAW_ARCHIVE");
assertThat(payload.get("vin").asText()).isEqualTo("VINRAW001");
assertThat(payload.get("eventTime").asText()).isEqualTo("2026-06-22T07:13:20Z");
assertThat(payload.get("rawArchiveUri").asText()).isEqualTo(rawArchiveUri);
assertThat(payload.get("rawArchiveKey").asText()).isEqualTo(rawArchiveKey);
assertThat(payload.get("rawSizeBytes").asLong()).isEqualTo(128);
assertThat(payload.get("metadata").get("command").asText()).isEqualTo("0x0002");
assertThat(payload.toString()).doesNotContain("rawBytes");
}
private static VehicleEnvelope envelope(String eventId) {
return VehicleEnvelope.newBuilder()
.setEventId(eventId)

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

@@ -17,7 +17,11 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor {
}
public void ingest(byte[] kafkaValue) {
processor.process(parse(kafkaValue));
VehicleEnvelope envelope = parse(kafkaValue);
if (!envelope.hasTelemetrySnapshot()) {
return;
}
processor.process(envelope);
}
@Override

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

@@ -2,6 +2,7 @@ package com.lingniu.ingest.vehiclestat;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
@@ -74,6 +75,23 @@ class VehicleStatEnvelopeIngestorTest {
assertThat(repository.mileagePoints("VIN001", LocalDate.of(2026, 6, 22))).isEmpty();
}
@Test
void rawArchiveEnvelopeIsIgnoredWithoutAppendingMileagePoint() {
InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository();
VehicleStatEnvelopeIngestor ingestor =
new VehicleStatEnvelopeIngestor(new VehicleStatEventProcessor(repository));
VehicleEnvelope envelope = rawArchiveEnvelope();
ingestor.ingest(envelope.toByteArray());
var result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED);
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 VehicleEnvelope.newBuilder()
.setEventId("event-1")
@@ -91,6 +109,20 @@ class VehicleStatEnvelopeIngestorTest {
.build();
}
private static VehicleEnvelope rawArchiveEnvelope() {
return VehicleEnvelope.newBuilder()
.setEventId("raw-event-1")
.setVin("VIN001")
.setSource("GB32960")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://2026/06/23/GB32960/VIN001/raw-event-1.bin")
.setSizeBytes(128)
.build())
.build();
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
private final java.util.Map<String, java.util.List<MileagePoint>> points = new java.util.HashMap<>();

10
modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java
View File

@@ -17,7 +17,11 @@ public final class VehicleStateEnvelopeIngestor implements EnvelopeIngestor {
}
public void ingest(byte[] kafkaValue) {
updater.update(parse(kafkaValue));
VehicleEnvelope envelope = parse(kafkaValue);
if (!envelope.hasTelemetrySnapshot()) {
return;
}
updater.update(envelope);
}
@Override
@@ -25,6 +29,10 @@ public final class VehicleStateEnvelopeIngestor implements EnvelopeIngestor {
VehicleEnvelope envelope = null;
try {
envelope = parse(kafkaValue);
if (!envelope.hasTelemetrySnapshot()) {
return EnvelopeIngestResult.skipped(
envelope.getEventId(), envelope.getVin(), "envelope telemetry_snapshot is required");
}
// Kafka 消费路径只接受标准 VehicleEnvelope；坏消息返回明确结果给处理器写 DLQ。
updater.update(envelope);
return EnvelopeIngestResult.processed(envelope.getEventId(), envelope.getVin());

34
modules/services/vehicle-state-service/src/test/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestorTest.java
View File

@@ -2,6 +2,7 @@ package com.lingniu.ingest.vehiclestate;
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
@@ -47,6 +48,25 @@ class VehicleStateEnvelopeIngestorTest {
assertThat(repository.getState("VIN001")).isEmpty();
}
@Test
void rawArchiveEnvelopeIsIgnoredWithoutUpdatingRepository() {
InMemoryVehicleStateRepository repository = new InMemoryVehicleStateRepository();
VehicleStateEnvelopeIngestor ingestor = new VehicleStateEnvelopeIngestor(new VehicleStateUpdater(repository));
VehicleEnvelope envelope = rawArchiveEnvelope();
ingestor.ingest(envelope.toByteArray());
var result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.SKIPPED);
assertThat(result.eventId()).isEqualTo("raw-event-1");
assertThat(result.vin()).isEqualTo("VIN001");
assertThat(result.message()).contains("telemetry_snapshot");
assertThat(repository.getState("VIN001")).isEmpty();
assertThat(repository.getLocation("VIN001")).isEmpty();
assertThat(repository.getSafety("VIN001")).isEmpty();
assertThat(repository.getLastEvent("VIN001")).isEmpty();
}
private static VehicleEnvelope envelope() {
return VehicleEnvelope.newBuilder()
.setEventId("event-1")
@@ -64,6 +84,20 @@ class VehicleStateEnvelopeIngestorTest {
.build();
}
private static VehicleEnvelope rawArchiveEnvelope() {
return VehicleEnvelope.newBuilder()
.setEventId("raw-event-1")
.setVin("VIN001")
.setSource("GB32960")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://2026/06/23/GB32960/VIN001/raw-event-1.bin")
.setSizeBytes(128)
.build())
.build();
}
private static final class EmptyRepository implements VehicleStateRepository {
@Override public void putState(String vin, String json) {}
@Override public void putLocation(String vin, String json) {}

17
modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveStore.java Normal file
View File

@@ -0,0 +1,17 @@
package com.lingniu.ingest.sink.archive;
import java.io.IOException;
import java.io.InputStream;
public interface ArchiveStore {
String put(String key, InputStream data, long length) throws IOException;
String append(String key, byte[] chunk) throws IOException;
InputStream get(String key) throws IOException;
boolean exists(String key);
long size(String key) throws IOException;
}

150
modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/LocalArchiveStore.java Normal file
View File

@@ -0,0 +1,150 @@
package com.lingniu.ingest.sink.archive;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public final class LocalArchiveStore implements ArchiveStore {
private final Path root;
private final ConcurrentMap<Path, Object> appendLocks = new ConcurrentHashMap<>();
public LocalArchiveStore(String root) {
this(rootPath(root));
}
public LocalArchiveStore(Path root) {
this.root = root.toAbsolutePath().normalize();
}
@Override
public String put(String key, InputStream data, long length) throws IOException {
if (data == null) {
throw new IllegalArgumentException("data must not be null");
}
Path target = resolve(key);
createDirectoriesInsideRoot(target.getParent());
rejectSymlinkPath(target);
try (OutputStream out = Files.newOutputStream(
target,
StandardOpenOption.CREATE,
StandardOpenOption.TRUNCATE_EXISTING,
StandardOpenOption.WRITE,
LinkOption.NOFOLLOW_LINKS)) {
data.transferTo(out);
}
return target.toUri().toString();
}
@Override
public String append(String key, byte[] chunk) throws IOException {
Path target = resolve(key);
synchronized (appendLocks.computeIfAbsent(target, ignored -> new Object())) {
createDirectoriesInsideRoot(target.getParent());
rejectSymlinkPath(target);
Files.write(
target,
chunk == null ? new byte[0] : chunk,
StandardOpenOption.CREATE,
StandardOpenOption.APPEND,
StandardOpenOption.WRITE,
LinkOption.NOFOLLOW_LINKS);
}
return target.toUri().toString();
}
@Override
public InputStream get(String key) throws IOException {
Path target = resolveExisting(key);
return Files.newInputStream(target, StandardOpenOption.READ, LinkOption.NOFOLLOW_LINKS);
}
@Override
public boolean exists(String key) {
try {
return Files.exists(resolveExisting(key), LinkOption.NOFOLLOW_LINKS);
} catch (IOException | IllegalArgumentException e) {
return false;
}
}
@Override
public long size(String key) throws IOException {
return Files.size(resolveExisting(key));
}
private Path resolve(String key) {
if (key == null || key.isBlank()) {
throw new IllegalArgumentException("archive key must not be blank");
}
Path target = root.resolve(key).normalize();
if (!target.startsWith(root)) {
throw new IllegalArgumentException("archive key escapes root: " + key);
}
return target;
}
private Path resolveExisting(String key) throws IOException {
Path target = resolve(key);
rejectSymlinkPath(target);
return target;
}
private void createDirectoriesInsideRoot(Path directory) throws IOException {
rejectEscapedPath(directory);
if (Files.notExists(root, LinkOption.NOFOLLOW_LINKS)) {
Files.createDirectories(root);
}
Path current = root;
Path relative = root.relativize(directory);
for (Path segment : relative) {
current = current.resolve(segment);
if (Files.isSymbolicLink(current)) {
throw new IOException("archive path traverses symlink: " + current);
}
if (Files.notExists(current, LinkOption.NOFOLLOW_LINKS)) {
Files.createDirectory(current);
} else if (!Files.isDirectory(current, LinkOption.NOFOLLOW_LINKS)) {
throw new IOException("archive path segment is not a directory: " + current);
}
}
}
private void rejectSymlinkPath(Path target) throws IOException {
rejectEscapedPath(target);
Path current = root;
Path relative = root.relativize(target);
for (Path segment : relative) {
current = current.resolve(segment);
if (Files.isSymbolicLink(current)) {
throw new IOException("archive path traverses symlink: " + current);
}
if (Files.notExists(current, LinkOption.NOFOLLOW_LINKS)) {
return;
}
}
}
private void rejectEscapedPath(Path path) {
if (!path.normalize().startsWith(root)) {
throw new IllegalArgumentException("archive path escapes root: " + path);
}
}
private static Path rootPath(String value) {
if (value == null || value.isBlank()) {
return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive");
}
if (value.startsWith("file://")) {
return Path.of(URI.create(value));
}
return Path.of(value);
}
}

44
modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java Normal file
View File

@@ -0,0 +1,44 @@
package com.lingniu.ingest.sink.archive;
import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.sink.EventSink;
import java.io.ByteArrayInputStream;
import java.util.concurrent.CompletableFuture;
public final class RawArchiveEventSink implements EventSink {
private final ArchiveStore store;
public RawArchiveEventSink(ArchiveStore store) {
this.store = store;
}
@Override
public String name() {
return "raw-archive";
}
@Override
public CompletableFuture<Void> publish(VehicleEvent event) {
if (!(event instanceof VehicleEvent.RawArchive raw)) {
return CompletableFuture.completedFuture(null);
}
try {
byte[] bytes = raw.rawBytes();
if (bytes == null) {
throw new IllegalArgumentException("raw archive bytes must not be null");
}
store.put(RawArchiveKeys.key(raw), new ByteArrayInputStream(bytes), bytes.length);
return CompletableFuture.completedFuture(null);
} catch (Exception e) {
return CompletableFuture.failedFuture(e);
}
}
@Override
public boolean accepts(VehicleEvent event) {
return event instanceof VehicleEvent.RawArchive;
}
}

35
modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java Normal file
View File

@@ -0,0 +1,35 @@
package com.lingniu.ingest.sink.archive.config;
import com.lingniu.ingest.sink.archive.ArchiveStore;
import com.lingniu.ingest.sink.archive.LocalArchiveStore;
import com.lingniu.ingest.sink.archive.RawArchiveEventSink;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
@AutoConfiguration
@EnableConfigurationProperties(SinkArchiveProperties.class)
@ConditionalOnProperty(
prefix = "lingniu.ingest.sink.archive",
name = "enabled",
havingValue = "true",
matchIfMissing = true)
public class SinkArchiveAutoConfiguration {
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.archive", name = "type", havingValue = "local", matchIfMissing = true)
public ArchiveStore archiveStore(SinkArchiveProperties properties) {
return new LocalArchiveStore(properties.getPath());
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnBean(ArchiveStore.class)
public RawArchiveEventSink rawArchiveEventSink(ArchiveStore store) {
return new RawArchiveEventSink(store);
}
}

35
modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java Normal file
View File

@@ -0,0 +1,35 @@
package com.lingniu.ingest.sink.archive.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "lingniu.ingest.sink.archive")
public class SinkArchiveProperties {
private boolean enabled = true;
private String type = "local";
private String path = System.getProperty("java.io.tmpdir") + "/lingniu-archive";
public boolean isEnabled() {
return enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public String getType() {
return type;
}
public void setType(String type) {
this.type = type;
}
public String getPath() {
return path;
}
public void setPath(String path) {
this.path = path;
}
}

74
modules/sinks/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/LocalArchiveStoreTest.java Normal file
View File

@@ -0,0 +1,74 @@
package com.lingniu.ingest.sink.archive;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.ByteArrayInputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class LocalArchiveStoreTest {
@TempDir
Path tempDir;
@Test
void rejectsSymlinkTraversalOutsideRootForWriteReadAndSize() throws Exception {
Path root = tempDir.resolve("archive");
Path outside = tempDir.resolve("outside");
Files.createDirectories(root);
Files.createDirectories(outside);
Files.writeString(outside.resolve("existing.bin"), "secret", StandardCharsets.UTF_8);
Files.createSymbolicLink(root.resolve("link"), outside);
LocalArchiveStore store = new LocalArchiveStore(root);
assertThatThrownBy(() -> store.put(
"link/new.bin",
new ByteArrayInputStream("payload".getBytes(StandardCharsets.UTF_8)),
7))
.isInstanceOfAny(IllegalArgumentException.class, java.io.IOException.class);
assertThat(outside.resolve("new.bin")).doesNotExist();
assertThatThrownBy(() -> store.get("link/existing.bin"))
.isInstanceOfAny(IllegalArgumentException.class, java.io.IOException.class);
assertThatThrownBy(() -> store.size("link/existing.bin"))
.isInstanceOfAny(IllegalArgumentException.class, java.io.IOException.class);
}
@Test
void appendsConcurrentChunksWithoutLosingData() throws Exception {
LocalArchiveStore store = new LocalArchiveStore(tempDir.resolve("archive"));
int chunks = 96;
try (var executor = Executors.newFixedThreadPool(12)) {
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (int i = 0; i < chunks; i++) {
String chunk = "%03d\n".formatted(i);
futures.add(CompletableFuture.runAsync(() -> {
try {
store.append("same/key.bin", chunk.getBytes(StandardCharsets.UTF_8));
} catch (Exception e) {
throw new RuntimeException(e);
}
}, executor));
}
CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
}
String content = Files.readString(tempDir.resolve("archive/same/key.bin"), StandardCharsets.UTF_8);
assertThat(content).hasSize(chunks * 4);
assertThat(content.lines()).containsExactlyInAnyOrderElementsOf(
java.util.stream.IntStream.range(0, chunks)
.mapToObj("%03d"::formatted)
.toList());
}
}

63
modules/sinks/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/RawArchiveEventSinkTest.java Normal file
View File

@@ -0,0 +1,63 @@
package com.lingniu.ingest.sink.archive;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.VehicleEvent;
import org.junit.jupiter.api.Test;
import java.io.InputStream;
import java.time.Instant;
import java.util.Map;
import java.util.concurrent.CompletionException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class RawArchiveEventSinkTest {
@Test
void failsFastWhenRawBytesAreNull() {
RawArchiveEventSink sink = new RawArchiveEventSink(new CapturingArchiveStore());
VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
"event-1",
"VIN123",
ProtocolId.GB32960,
Instant.parse("2026-06-23T00:00:00Z"),
Instant.parse("2026-06-23T00:00:01Z"),
"trace-1",
Map.of(),
2,
1,
null);
assertThatThrownBy(() -> sink.publish(raw).join())
.isInstanceOf(CompletionException.class)
.hasCauseInstanceOf(IllegalArgumentException.class);
}
private static final class CapturingArchiveStore implements ArchiveStore {
@Override
public String put(String key, InputStream data, long length) {
return "archive://" + key;
}
@Override
public String append(String key, byte[] chunk) {
return "archive://" + key;
}
@Override
public InputStream get(String key) {
throw new UnsupportedOperationException();
}
@Override
public boolean exists(String key) {
return false;
}
@Override
public long size(String key) {
return 0;
}
}
}

40
modules/sinks/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfigurationTest.java Normal file
View File

@@ -0,0 +1,40 @@
package com.lingniu.ingest.sink.archive.config;
import com.lingniu.ingest.sink.archive.ArchiveStore;
import com.lingniu.ingest.sink.archive.RawArchiveEventSink;
import org.junit.jupiter.api.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.core.env.MapPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
class SinkArchiveAutoConfigurationTest {
@Test
void localArchiveTypeCreatesStoreAndSink() {
try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) {
context.getEnvironment().getPropertySources().addFirst(new MapPropertySource(
"test",
java.util.Map.of("lingniu.ingest.sink.archive.path", "target/test-archive")));
context.register(SinkArchiveAutoConfiguration.class);
context.refresh();
assertThat(context.getBeansOfType(ArchiveStore.class)).hasSize(1);
assertThat(context.getBeansOfType(RawArchiveEventSink.class)).hasSize(1);
}
}
@Test
void nonLocalArchiveTypeDoesNotCreateSinkWithoutStore() {
try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) {
context.getEnvironment().getPropertySources().addFirst(new MapPropertySource(
"test",
java.util.Map.of("lingniu.ingest.sink.archive.type", "oss")));
context.register(SinkArchiveAutoConfiguration.class);
context.refresh();
assertThat(context.getBeansOfType(ArchiveStore.class)).isEmpty();
assertThat(context.getBeansOfType(RawArchiveEventSink.class)).isEmpty();
}
}
}

3
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java
View File

@@ -71,8 +71,7 @@ public final class EnvelopeMapper {
int size = ra.rawBytes() == null ? 0 : ra.rawBytes().length;
String key = rawArchiveKey(ra);
String uri = rawArchiveUri(ra, key);
// RAW envelope 只携带引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。
// 注意 KafkaEventSink 当前仍过滤 RawArchive；这段用于未来放开 raw 引用转发或消费者测试。
// RAW envelope 只携带 URI/size 引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。
b.putMetadata(RawArchiveKeys.META_KEY, key);
b.putMetadata(RawArchiveKeys.META_URI, uri);
b.putMetadata(RawArchiveKeys.META_EVENT_ID, ra.eventId());

6
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java
View File

@@ -43,13 +43,11 @@ public final class KafkaEventSink implements EventSink, AutoCloseable {
}
/**
* 拒绝 {@link VehicleEvent.RawArchive} —— 当前生产 Kafka sink 只投递实时、会话、告警等轻量事件。
* {@link EnvelopeMapper} 和 {@link TopicRouter} 已具备 RAW envelope/topic 的映射能力，但这里仍
* 显式过滤；如需把 32960 raw 引用转发到 {@code vehicle.raw.archive}，应先移除此过滤并补充测试。
* Split-service mode uses Kafka for both normalized events and raw archive records.
*/
@Override
public boolean accepts(VehicleEvent event) {
return !(event instanceof VehicleEvent.RawArchive);
return true;
}
@Override

14
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java
View File

@@ -57,16 +57,16 @@ public class SinkMqProperties {
public static class Topics {
/** 实时遥测事件 topic；GB32960 RAW-only 架构下可逐步弱化该 topic。 */
private String realtime = "vehicle.realtime";
private String realtime = "vehicle.event.gb32960.v1";
/** 位置事件 topic；通常可由 realtime/RAW 派生。 */
private String location = "vehicle.location";
private String alarm = "vehicle.alarm";
private String session = "vehicle.session";
private String mediaMeta = "vehicle.media.meta";
private String location = "vehicle.event.gb32960.v1";
private String alarm = "vehicle.event.gb32960.v1";
private String session = "vehicle.event.gb32960.v1";
private String mediaMeta = "vehicle.media.meta.v1";
/** RAW 归档引用 topic，payload 应携带 archive URI/size，不建议携带完整 raw bytes。 */
private String rawArchive = "vehicle.raw.archive";
private String rawArchive = "vehicle.raw.gb32960.v1";
/** producer 熔断或 consumer 处理失败时的死信 topic。 */
private String dlq = "vehicle.dlq";
private String dlq = "vehicle.dlq.gb32960.v1";
public String getRealtime() { return realtime; }
public void setRealtime(String realtime) { this.realtime = realtime; }

40
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java Normal file
View File

@@ -0,0 +1,40 @@
package com.lingniu.ingest.sink.mq;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.VehicleEvent;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
class KafkaEventSinkTest {
@Test
void acceptsRawArchiveRecordsForSplitServiceKafkaBoundary() {
KafkaEventSink sink = new KafkaEventSink(
null,
new EnvelopeMapper("node-1"),
new TopicRouter(new SinkMqProperties.Topics()),
"vehicle.dlq.gb32960.v1",
CircuitBreaker.ofDefaults("kafka-test"));
assertThat(sink.accepts(rawArchive())).isTrue();
}
private static VehicleEvent.RawArchive rawArchive() {
return new VehicleEvent.RawArchive(
"raw-1",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-raw-1",
Map.of("platformAccount", "Hyundai"),
0x02,
0,
new byte[] {0x23, 0x23});
}
}

147
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java Normal file
View File

@@ -0,0 +1,147 @@
package com.lingniu.ingest.sink.mq;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.AlarmPayload;
import com.lingniu.ingest.api.event.LocationPayload;
import com.lingniu.ingest.api.event.RealtimePayload;
import com.lingniu.ingest.api.event.VehicleEvent;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
class TopicRouterTest {
@Test
void rawArchiveRoutesToVersionedGb32960RawTopicByDefault() {
TopicRouter router = new TopicRouter(new SinkMqProperties.Topics());
assertThat(router.route(rawArchive())).isEqualTo("vehicle.raw.gb32960.v1");
}
@Test
void normalizedGb32960EventsRouteToVersionedEventTopicByDefault() {
TopicRouter router = new TopicRouter(new SinkMqProperties.Topics());
assertThat(router.route(realtime())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(location())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(alarm())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(login())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(logout())).isEqualTo("vehicle.event.gb32960.v1");
assertThat(router.route(heartbeat())).isEqualTo("vehicle.event.gb32960.v1");
}
@Test
void dlqDefaultsToVersionedGb32960DlqTopic() {
SinkMqProperties.Topics topics = new SinkMqProperties.Topics();
assertThat(topics.getDlq()).isEqualTo("vehicle.dlq.gb32960.v1");
}
private static VehicleEvent.RawArchive rawArchive() {
return new VehicleEvent.RawArchive(
"raw-1",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-raw-1",
Map.of("platformAccount", "Hyundai"),
0x02,
0,
new byte[] {0x23, 0x23});
}
private static VehicleEvent.Realtime realtime() {
return new VehicleEvent.Realtime(
"event-1",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-1",
Map.of("platformAccount", "Hyundai"),
new RealtimePayload(
42.1, 1234.5, 80.0, null, null,
null, null, null, null, null, null,
RealtimePayload.VehicleState.STARTED,
RealtimePayload.ChargingState.UNCHARGED,
RealtimePayload.RunningMode.ELECTRIC,
null, null, null,
113.12, 23.45, null, null, null, null));
}
private static VehicleEvent.Location location() {
return new VehicleEvent.Location(
"event-2",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-2",
Map.of("platformAccount", "Hyundai"),
new LocationPayload(113.12, 23.45, 0, 42.1, 90, 0, 0));
}
private static VehicleEvent.Alarm alarm() {
return new VehicleEvent.Alarm(
"event-3",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-3",
Map.of("platformAccount", "Hyundai"),
new AlarmPayload(
AlarmPayload.AlarmLevel.MINOR,
1,
"GB32960_ALARM",
List.of("1"),
Set.of("bit0"),
113.12,
23.45,
AlarmPayload.SafetyCategory.GENERAL,
false,
AlarmPayload.HydrogenLeakLevel.NONE,
false));
}
private static VehicleEvent.Login login() {
return new VehicleEvent.Login(
"event-4",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-4",
Map.of("platformAccount", "Hyundai"),
"iccid-1",
"V2016");
}
private static VehicleEvent.Logout logout() {
return new VehicleEvent.Logout(
"event-5",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-5",
Map.of("platformAccount", "Hyundai"));
}
private static VehicleEvent.Heartbeat heartbeat() {
return new VehicleEvent.Heartbeat(
"event-6",
"VIN001",
ProtocolId.GB32960,
Instant.parse("2026-06-23T07:00:00Z"),
Instant.parse("2026-06-23T07:00:01Z"),
"trace-6",
Map.of("platformAccount", "Hyundai"));
}
}