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

feat: add yutong mqtt and xinda push apps
Some checks failed
ci/woodpecker/push/woodpecker Pipeline was canceled
Details

Browse Source
This commit is contained in:
lingniu
2026-06-29 23:33:45 +08:00
parent 1682b3f83f
commit 4d683809ad
16 changed files with 908 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java
View File

@@ -72,9 +72,13 @@ public class VehicleHistoryKafkaConsumerConfiguration {
"eventHistoryEnvelopeConsumerProcessor", processor,
"eventHistoryGb32960EnvelopeConsumerProcessor", processor,
"eventHistoryJt808EnvelopeConsumerProcessor", processor,
"eventHistoryYutongMqttEnvelopeConsumerProcessor", processor,
"eventHistoryXindaPushEnvelopeConsumerProcessor", processor,
"eventHistoryRawEnvelopeConsumerProcessor", rawProcessor,
"eventHistoryGb32960RawEnvelopeConsumerProcessor", rawProcessor,
"eventHistoryJt808RawEnvelopeConsumerProcessor", rawProcessor),
"eventHistoryJt808RawEnvelopeConsumerProcessor", rawProcessor,
"eventHistoryYutongMqttRawEnvelopeConsumerProcessor", rawProcessor,
"eventHistoryXindaPushRawEnvelopeConsumerProcessor", rawProcessor),
props);
if (workers.isEmpty()) {
throw new IllegalStateException("no vehicle history kafka consumer workers created; check consumer bindings");

20
modules/apps/vehicle-history-app/src/main/resources/application.yml
View File

@@ -60,6 +60,16 @@ lingniu:
group-id: ${KAFKA_GROUP_HISTORY_JT808_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-jt808-event}
topics:
- ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
eventHistoryYutongMqttEnvelopeConsumerProcessor:
enabled: true
group-id: ${KAFKA_GROUP_HISTORY_YUTONG_MQTT_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-yutong-mqtt-event}
topics:
- ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:vehicle.event.mqtt-yutong.v1}
eventHistoryXindaPushEnvelopeConsumerProcessor:
enabled: true
group-id: ${KAFKA_GROUP_HISTORY_XINDA_PUSH_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-xinda-push-event}
topics:
- ${KAFKA_TOPIC_XINDA_PUSH_EVENT:vehicle.event.xinda-push.v1}
eventHistoryGb32960RawEnvelopeConsumerProcessor:
enabled: true
group-id: ${KAFKA_GROUP_HISTORY_GB32960_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-gb32960-raw}
@@ -70,6 +80,16 @@ lingniu:
group-id: ${KAFKA_GROUP_HISTORY_JT808_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-jt808-raw}
topics:
- ${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}
eventHistoryYutongMqttRawEnvelopeConsumerProcessor:
enabled: true
group-id: ${KAFKA_GROUP_HISTORY_YUTONG_MQTT_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-yutong-mqtt-raw}
topics:
- ${KAFKA_TOPIC_YUTONG_MQTT_RAW:vehicle.raw.mqtt-yutong.v1}
eventHistoryXindaPushRawEnvelopeConsumerProcessor:
enabled: true
group-id: ${KAFKA_GROUP_HISTORY_XINDA_PUSH_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-xinda-push-raw}
topics:
- ${KAFKA_TOPIC_XINDA_PUSH_RAW:vehicle.raw.xinda-push.v1}
archive:
enabled: ${SINK_ARCHIVE_ENABLED:true}
type: local

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

@@ -66,6 +66,24 @@ class VehicleHistoryAppDefaultsTest {
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808EnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttEnvelopeConsumerProcessor.enabled",
true)
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttEnvelopeConsumerProcessor.group-id",
"${KAFKA_GROUP_HISTORY_YUTONG_MQTT_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-yutong-mqtt-event}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttEnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_YUTONG_MQTT_EVENT:vehicle.event.mqtt-yutong.v1}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryXindaPushEnvelopeConsumerProcessor.enabled",
true)
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryXindaPushEnvelopeConsumerProcessor.group-id",
"${KAFKA_GROUP_HISTORY_XINDA_PUSH_EVENT:${KAFKA_GROUP_HISTORY:vehicle-history}-xinda-push-event}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryXindaPushEnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_XINDA_PUSH_EVENT:vehicle.event.xinda-push.v1}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryGb32960RawEnvelopeConsumerProcessor.enabled",
true)
@@ -83,7 +101,25 @@ class VehicleHistoryAppDefaultsTest {
"${KAFKA_GROUP_HISTORY_JT808_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-jt808-raw}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryJt808RawEnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}");
"${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttRawEnvelopeConsumerProcessor.enabled",
true)
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttRawEnvelopeConsumerProcessor.group-id",
"${KAFKA_GROUP_HISTORY_YUTONG_MQTT_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-yutong-mqtt-raw}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttRawEnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_YUTONG_MQTT_RAW:vehicle.raw.mqtt-yutong.v1}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryXindaPushRawEnvelopeConsumerProcessor.enabled",
true)
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryXindaPushRawEnvelopeConsumerProcessor.group-id",
"${KAFKA_GROUP_HISTORY_XINDA_PUSH_RAW:${KAFKA_GROUP_HISTORY:vehicle-history}-xinda-push-raw}")
.containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryXindaPushRawEnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_XINDA_PUSH_RAW:vehicle.raw.xinda-push.v1}");
}
private static Properties applicationProperties() {

79
modules/apps/xinda-push-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>xinda-push-app</artifactId>
<name>xinda-push-app</name>
<description>Xinda Push ingress runtime: platform push client, frame parse, identity mapping, Kafka production, and raw archive.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-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>inbound-xinda-push</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>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>xinda-push-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/xinda-push-app/src/main/java/com/lingniu/ingest/xindapushapp/XindaPushApplication.java Normal file
View File

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

115
modules/apps/xinda-push-app/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,115 @@
spring:
application:
name: xinda-push-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:20600}
springdoc:
swagger-ui:
path: /swagger-ui.html
api-docs:
path: /v3/api-docs
lingniu:
ingest:
xinda-push:
enabled: ${XINDA_PUSH_ENABLED:false}
host: ${XINDA_PUSH_HOST:}
port: ${XINDA_PUSH_PORT:10100}
username: ${XINDA_PUSH_USERNAME:}
password: ${XINDA_PUSH_PASSWORD:}
client-desc: ${XINDA_PUSH_CLIENT_DESC:lingniu-ingest}
subscribe-msg-ids:
- ${XINDA_PUSH_MSG_ID_LOCATION:0200}
- ${XINDA_PUSH_MSG_ID_ALARM:0300}
- ${XINDA_PUSH_MSG_ID_PASSTHROUGH:0401}
reconnect-interval-sec: ${XINDA_PUSH_RECONNECT_INTERVAL_SEC:5}
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
identity:
store: ${VEHICLE_IDENTITY_STORE:file}
file:
path: ${VEHICLE_IDENTITY_FILE:./data/vehicle-identity.jsonl}
mysql:
table: ${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding}
jdbc-url: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:}
username: ${VEHICLE_IDENTITY_MYSQL_USERNAME:}
password: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:}
driver-class-name: ${VEHICLE_IDENTITY_MYSQL_DRIVER_CLASS_NAME:com.mysql.cj.jdbc.Driver}
refresh-interval: ${VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL:60s}
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:xinda-push-local}
topics:
realtime: ${KAFKA_TOPIC_XINDA_PUSH_EVENT:vehicle.event.xinda-push.v1}
location: ${KAFKA_TOPIC_XINDA_PUSH_EVENT:vehicle.event.xinda-push.v1}
alarm: ${KAFKA_TOPIC_XINDA_PUSH_EVENT:vehicle.event.xinda-push.v1}
session: ${KAFKA_TOPIC_XINDA_PUSH_EVENT:vehicle.event.xinda-push.v1}
media-meta: ${KAFKA_TOPIC_MEDIA_META:vehicle.media.meta.v1}
raw-archive: ${KAFKA_TOPIC_XINDA_PUSH_RAW:vehicle.raw.xinda-push.v1}
dlq: ${KAFKA_TOPIC_XINDA_PUSH_DLQ:vehicle.dlq.xinda-push.v1}
consumer:
enabled: false
archive:
enabled: ${SINK_ARCHIVE_ENABLED:true}
type: local
path: ${SINK_ARCHIVE_PATH:./archive/}
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: xinda-push-app
logging:
level:
root: INFO
com.lingniu.ingest: INFO

68
modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppCompositionTest.java Normal file
View File

@@ -0,0 +1,68 @@
package com.lingniu.ingest.xindapushapp;
import com.lingniu.ingest.core.config.IngestCoreAutoConfiguration;
import com.lingniu.ingest.identity.config.VehicleIdentityAutoConfiguration;
import com.lingniu.ingest.inbound.xinda.XindaPushClient;
import com.lingniu.ingest.inbound.xinda.XindaPushEventMapper;
import com.lingniu.ingest.inbound.xinda.XindaPushRealtimeHandler;
import com.lingniu.ingest.inbound.xinda.config.XindaPushAutoConfiguration;
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.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
class XindaPushAppCompositionTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
IngestCoreAutoConfiguration.class,
VehicleIdentityAutoConfiguration.class,
SinkMqAutoConfiguration.class,
SinkArchiveAutoConfiguration.class,
XindaPushAutoConfiguration.class))
.withAllowBeanDefinitionOverriding(true)
.withBean("kafkaProducer", KafkaProducer.class, XindaPushAppCompositionTest::kafkaProducer)
.withPropertyValues(
"lingniu.ingest.xinda-push.enabled=true",
"lingniu.ingest.xinda-push.host=127.0.0.1",
"lingniu.ingest.xinda-push.port=10100",
"lingniu.ingest.xinda-push.username=test",
"lingniu.ingest.xinda-push.password=test",
"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.sink.archive.enabled=true",
"lingniu.ingest.sink.archive.path=target/test-archive-xinda",
"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 createsXindaPushIngressAndKafkaSinksWithoutHistoryStorage() {
contextRunner.run(context -> {
assertThat(context).hasSingleBean(XindaPushClient.class);
assertThat(context).hasSingleBean(XindaPushEventMapper.class);
assertThat(context).hasSingleBean(XindaPushRealtimeHandler.class);
assertThat(context).hasSingleBean(KafkaEventSink.class);
assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
assertThat(context).hasSingleBean(ArchiveStore.class);
assertThat(context).hasSingleBean(RawArchiveEventSink.class);
});
}
@SuppressWarnings("unchecked")
private static KafkaProducer<String, byte[]> kafkaProducer() {
return mock(KafkaProducer.class);
}
}

52
modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppDefaultsTest.java Normal file
View File

@@ -0,0 +1,52 @@
package com.lingniu.ingest.xindapushapp;
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 XindaPushAppDefaultsTest {
@Test
void applicationDefaultsKeepXindaPushAsKafkaProducerOnly() {
Properties properties = applicationProperties();
assertThat(properties)
.containsEntry("spring.application.name", "xinda-push-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:20600}")
.containsEntry("lingniu.ingest.xinda-push.enabled", "${XINDA_PUSH_ENABLED:false}")
.containsEntry("lingniu.ingest.xinda-push.host", "${XINDA_PUSH_HOST:}")
.containsEntry("lingniu.ingest.xinda-push.port", "${XINDA_PUSH_PORT:10100}")
.containsEntry("lingniu.ingest.xinda-push.username", "${XINDA_PUSH_USERNAME:}")
.containsEntry("lingniu.ingest.xinda-push.subscribe-msg-ids[0]", "${XINDA_PUSH_MSG_ID_LOCATION:0200}")
.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_XINDA_PUSH_EVENT:vehicle.event.xinda-push.v1}")
.containsEntry("lingniu.ingest.sink.mq.topics.raw-archive", "${KAFKA_TOPIC_XINDA_PUSH_RAW:vehicle.raw.xinda-push.v1}")
.containsEntry("lingniu.ingest.sink.mq.topics.dlq", "${KAFKA_TOPIC_XINDA_PUSH_DLQ:vehicle.dlq.xinda-push.v1}")
.containsEntry("lingniu.ingest.identity.store", "${VEHICLE_IDENTITY_STORE:file}")
.containsEntry("lingniu.ingest.identity.mysql.table", "${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding}")
.containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}")
.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;
}
}

79
modules/apps/yutong-mqtt-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>yutong-mqtt-app</artifactId>
<name>yutong-mqtt-app</name>
<description>Yutong MQTT ingress runtime: MQTT subscribe, JSON parse, identity mapping, Kafka production, and raw archive.</description>
<dependencies>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-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>inbound-mqtt</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>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>yutong-mqtt-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/yutong-mqtt-app/src/main/java/com/lingniu/ingest/yutongmqttapp/YutongMqttApplication.java Normal file
View File

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

121
modules/apps/yutong-mqtt-app/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,121 @@
spring:
application:
name: yutong-mqtt-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:20500}
springdoc:
swagger-ui:
path: /swagger-ui.html
api-docs:
path: /v3/api-docs
lingniu:
ingest:
mqtt:
enabled: ${YUTONG_MQTT_ENABLED:false}
endpoints:
- name: ${YUTONG_MQTT_ENDPOINT_NAME:yutong}
uri: ${YUTONG_MQTT_URI:}
topic: ${YUTONG_MQTT_TOPIC:#}
qos: ${YUTONG_MQTT_QOS:1}
client-id: ${YUTONG_MQTT_CLIENT_ID:lingniu-yutong-mqtt}
username: ${YUTONG_MQTT_USERNAME:}
password: ${YUTONG_MQTT_PASSWORD:}
clean-session: ${YUTONG_MQTT_CLEAN_SESSION:false}
keep-alive-seconds: ${YUTONG_MQTT_KEEP_ALIVE_SECONDS:20}
connection-timeout-seconds: ${YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS:10}
profile: yutong
tls:
ca-pem: ${YUTONG_MQTT_TLS_CA_PEM:}
client-pem: ${YUTONG_MQTT_TLS_CLIENT_PEM:}
client-key: ${YUTONG_MQTT_TLS_CLIENT_KEY:}
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
identity:
store: ${VEHICLE_IDENTITY_STORE:file}
file:
path: ${VEHICLE_IDENTITY_FILE:./data/vehicle-identity.jsonl}
mysql:
table: ${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding}
jdbc-url: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:}
username: ${VEHICLE_IDENTITY_MYSQL_USERNAME:}
password: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:}
driver-class-name: ${VEHICLE_IDENTITY_MYSQL_DRIVER_CLASS_NAME:com.mysql.cj.jdbc.Driver}
refresh-interval: ${VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL:60s}
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:yutong-mqtt-local}
topics:
realtime: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:vehicle.event.mqtt-yutong.v1}
location: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:vehicle.event.mqtt-yutong.v1}
alarm: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:vehicle.event.mqtt-yutong.v1}
session: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:vehicle.event.mqtt-yutong.v1}
media-meta: ${KAFKA_TOPIC_MEDIA_META:vehicle.media.meta.v1}
raw-archive: ${KAFKA_TOPIC_YUTONG_MQTT_RAW:vehicle.raw.mqtt-yutong.v1}
dlq: ${KAFKA_TOPIC_YUTONG_MQTT_DLQ:vehicle.dlq.mqtt-yutong.v1}
consumer:
enabled: false
archive:
enabled: ${SINK_ARCHIVE_ENABLED:true}
type: local
path: ${SINK_ARCHIVE_PATH:./archive/}
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: yutong-mqtt-app
logging:
level:
root: INFO
com.lingniu.ingest: INFO

68
modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppCompositionTest.java Normal file
View File

@@ -0,0 +1,68 @@
package com.lingniu.ingest.yutongmqttapp;
import com.lingniu.ingest.core.config.IngestCoreAutoConfiguration;
import com.lingniu.ingest.identity.config.VehicleIdentityAutoConfiguration;
import com.lingniu.ingest.inbound.mqtt.client.MqttEndpointManager;
import com.lingniu.ingest.inbound.mqtt.config.MqttInboundAutoConfiguration;
import com.lingniu.ingest.inbound.mqtt.handler.MqttRealtimeHandler;
import com.lingniu.ingest.inbound.mqtt.profile.MqttProfileRegistry;
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.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
class YutongMqttAppCompositionTest {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
IngestCoreAutoConfiguration.class,
VehicleIdentityAutoConfiguration.class,
SinkMqAutoConfiguration.class,
SinkArchiveAutoConfiguration.class,
MqttInboundAutoConfiguration.class))
.withAllowBeanDefinitionOverriding(true)
.withBean("kafkaProducer", KafkaProducer.class, YutongMqttAppCompositionTest::kafkaProducer)
.withPropertyValues(
"lingniu.ingest.mqtt.enabled=true",
"lingniu.ingest.mqtt.endpoints[0].name=yutong-test",
"lingniu.ingest.mqtt.endpoints[0].uri=tcp://127.0.0.1:1883",
"lingniu.ingest.mqtt.endpoints[0].topic=/yutong/#",
"lingniu.ingest.mqtt.endpoints[0].profile=yutong",
"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.sink.archive.enabled=true",
"lingniu.ingest.sink.archive.path=target/test-archive-yutong",
"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 createsMqttIngressAndKafkaSinksWithoutHistoryStorage() {
contextRunner.run(context -> {
assertThat(context).hasSingleBean(MqttEndpointManager.class);
assertThat(context).hasSingleBean(MqttProfileRegistry.class);
assertThat(context).hasSingleBean(MqttRealtimeHandler.class);
assertThat(context).hasSingleBean(KafkaEventSink.class);
assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
assertThat(context).hasSingleBean(ArchiveStore.class);
assertThat(context).hasSingleBean(RawArchiveEventSink.class);
});
}
@SuppressWarnings("unchecked")
private static KafkaProducer<String, byte[]> kafkaProducer() {
return mock(KafkaProducer.class);
}
}

51
modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppDefaultsTest.java Normal file
View File

@@ -0,0 +1,51 @@
package com.lingniu.ingest.yutongmqttapp;
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 YutongMqttAppDefaultsTest {
@Test
void applicationDefaultsKeepYutongMqttAsKafkaProducerOnly() {
Properties properties = applicationProperties();
assertThat(properties)
.containsEntry("spring.application.name", "yutong-mqtt-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:20500}")
.containsEntry("lingniu.ingest.mqtt.enabled", "${YUTONG_MQTT_ENABLED:false}")
.containsEntry("lingniu.ingest.mqtt.endpoints[0].uri", "${YUTONG_MQTT_URI:}")
.containsEntry("lingniu.ingest.mqtt.endpoints[0].topic", "${YUTONG_MQTT_TOPIC:#}")
.containsEntry("lingniu.ingest.mqtt.endpoints[0].profile", "yutong")
.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_YUTONG_MQTT_EVENT:vehicle.event.mqtt-yutong.v1}")
.containsEntry("lingniu.ingest.sink.mq.topics.raw-archive", "${KAFKA_TOPIC_YUTONG_MQTT_RAW:vehicle.raw.mqtt-yutong.v1}")
.containsEntry("lingniu.ingest.sink.mq.topics.dlq", "${KAFKA_TOPIC_YUTONG_MQTT_DLQ:vehicle.dlq.mqtt-yutong.v1}")
.containsEntry("lingniu.ingest.identity.store", "${VEHICLE_IDENTITY_STORE:file}")
.containsEntry("lingniu.ingest.identity.mysql.table", "${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding}")
.containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}")
.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;
}
}