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

feat: add jt808 ingest app
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Details

Browse Source
This commit is contained in:
lingniu
2026-06-26 11:17:27 +08:00
parent db871b7531
commit 5e619c3634
16 changed files with 415 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
modules/apps/jt808-ingest-app/pom.xml Normal file
View File

@@ -0,0 +1,79 @@
<?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>jt808-ingest-app</artifactId>
<name>jt808-ingest-app</name>
<description>JT808 protocol ingress runtime: TCP decode, identity mapping, Kafka production, and downlink session routing.</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>vehicle-identity</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>protocol-jt808</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>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<dependency>
<groupId>org.springdoc</groupId>
<artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>jt808-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>
<goal>build-info</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

21
modules/apps/jt808-ingest-app/src/main/java/com/lingniu/ingest/jt808app/Jt808IngestApplication.java Normal file
View File

@@ -0,0 +1,21 @@
package com.lingniu.ingest.jt808app;
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 Jt808IngestApplication {
private static final Logger log = LoggerFactory.getLogger(Jt808IngestApplication.class);
public static void main(String[] args) {
try {
SpringApplication.run(Jt808IngestApplication.class, args);
} catch (Throwable t) {
log.error("JT808 ingest application failed to start, forcing JVM exit", t);
System.exit(1);
}
}
}

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

@@ -0,0 +1,99 @@
spring:
application:
name: jt808-ingest-app
config:
import:
- optional:nacos:${spring.application.name}.${NACOS_CONFIG_FILE_EXTENSION:yml}?group=${NACOS_GROUP:DEFAULT_GROUP}&refreshEnabled=${NACOS_REFRESH_ENABLED:true}
cloud:
nacos:
config:
enabled: ${NACOS_CONFIG_ENABLED:true}
server-addr: ${NACOS_SERVER_ADDR:127.0.0.1:8848}
namespace: ${NACOS_NAMESPACE:}
group: ${NACOS_GROUP:DEFAULT_GROUP}
file-extension: ${NACOS_CONFIG_FILE_EXTENSION:yml}
username: ${NACOS_USERNAME:}
password: ${NACOS_PASSWORD:}
threads:
virtual:
enabled: true
server:
port: ${HTTP_PORT:20400}
springdoc:
swagger-ui:
path: /swagger-ui.html
api-docs:
path: /v3/api-docs
lingniu:
ingest:
jt808:
enabled: true
port: ${JT808_PORT:10808}
worker-threads: ${JT808_WORKER_THREADS:0}
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:jt808-ingest-local}
topics:
realtime: ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
location: ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
alarm: ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
session: ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
media-meta: ${KAFKA_TOPIC_MEDIA_META:vehicle.media.meta.v1}
raw-archive: ${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}
dlq: ${KAFKA_TOPIC_JT808_DLQ:vehicle.dlq.jt808.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: jt808-ingest-app
logging:
level:
root: INFO
com.lingniu.ingest: INFO

69
modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java Normal file
View File

@@ -0,0 +1,69 @@
package com.lingniu.ingest.jt808app;
import com.lingniu.ingest.core.config.IngestCoreAutoConfiguration;
import com.lingniu.ingest.identity.config.VehicleIdentityAutoConfiguration;
import com.lingniu.ingest.protocol.jt808.codec.Jt808MessageDecoder;
import com.lingniu.ingest.protocol.jt808.config.Jt808AutoConfiguration;
import com.lingniu.ingest.protocol.jt808.inbound.Jt808NettyServer;
import com.lingniu.ingest.session.config.SessionCoreAutoConfiguration;
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.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 Jt808IngestAppCompositionTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
IngestCoreAutoConfiguration.class,
SessionCoreAutoConfiguration.class,
VehicleIdentityAutoConfiguration.class,
SinkMqAutoConfiguration.class,
Jt808AutoConfiguration.class))
.withAllowBeanDefinitionOverriding(true)
.withBean("kafkaProducer", KafkaProducer.class, Jt808IngestAppCompositionTest::kafkaProducer)
.withPropertyValues(
"lingniu.ingest.jt808.enabled=true",
"lingniu.ingest.jt808.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 createsJt808ListenerWithoutHistoryStorageBoundaries() {
contextRunner.run(context -> {
assertThat(context).hasSingleBean(Jt808MessageDecoder.class);
assertThat(context).hasSingleBean(Jt808NettyServer.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 JT808 ingest runtime", className)
.isFalse();
}
}

46
modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java Normal file
View File

@@ -0,0 +1,46 @@
package com.lingniu.ingest.jt808app;
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 Jt808IngestAppDefaultsTest {
@Test
void applicationDefaultsKeepJt808IngestAsTcpKafkaProducerOnly() {
Properties properties = applicationProperties();
assertThat(properties)
.containsEntry("spring.application.name", "jt808-ingest-app")
.containsEntry(
"spring.config.import[0]",
"optional:nacos:${spring.application.name}.${NACOS_CONFIG_FILE_EXTENSION:yml}?group=${NACOS_GROUP:DEFAULT_GROUP}&refreshEnabled=${NACOS_REFRESH_ENABLED:true}")
.containsEntry("spring.cloud.nacos.config.server-addr", "${NACOS_SERVER_ADDR:127.0.0.1:8848}")
.containsEntry("server.port", "${HTTP_PORT:20400}")
.containsEntry("lingniu.ingest.jt808.enabled", true)
.containsEntry("lingniu.ingest.jt808.port", "${JT808_PORT:10808}")
.containsEntry("lingniu.ingest.sink.mq.enabled", "${KAFKA_ENABLED:true}")
.containsEntry("lingniu.ingest.sink.mq.consumer.enabled", false)
.containsEntry("lingniu.ingest.sink.mq.topics.realtime", "${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}")
.containsEntry("lingniu.ingest.sink.mq.topics.raw-archive", "${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}")
.containsEntry("lingniu.ingest.sink.mq.topics.dlq", "${KAFKA_TOPIC_JT808_DLQ:vehicle.dlq.jt808.v1}")
.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;
}
}