diff --git a/deploy/portainer/docker-compose.yml b/deploy/portainer/docker-compose.yml
index 16ea2247..f5c98879 100644
--- a/deploy/portainer/docker-compose.yml
+++ b/deploy/portainer/docker-compose.yml
@@ -22,6 +22,12 @@ x-common-env: &common-env
KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1}
KAFKA_TOPIC_JT808_RAW: ${KAFKA_TOPIC_JT808_RAW:-vehicle.raw.jt808.v1}
KAFKA_TOPIC_JT808_DLQ: ${KAFKA_TOPIC_JT808_DLQ:-vehicle.dlq.jt808.v1}
+ KAFKA_TOPIC_YUTONG_MQTT_EVENT: ${KAFKA_TOPIC_YUTONG_MQTT_EVENT:-vehicle.event.mqtt-yutong.v1}
+ KAFKA_TOPIC_YUTONG_MQTT_RAW: ${KAFKA_TOPIC_YUTONG_MQTT_RAW:-vehicle.raw.mqtt-yutong.v1}
+ KAFKA_TOPIC_YUTONG_MQTT_DLQ: ${KAFKA_TOPIC_YUTONG_MQTT_DLQ:-vehicle.dlq.mqtt-yutong.v1}
+ KAFKA_TOPIC_XINDA_PUSH_EVENT: ${KAFKA_TOPIC_XINDA_PUSH_EVENT:-vehicle.event.xinda-push.v1}
+ KAFKA_TOPIC_XINDA_PUSH_RAW: ${KAFKA_TOPIC_XINDA_PUSH_RAW:-vehicle.raw.xinda-push.v1}
+ KAFKA_TOPIC_XINDA_PUSH_DLQ: ${KAFKA_TOPIC_XINDA_PUSH_DLQ:-vehicle.dlq.xinda-push.v1}
services:
gb32960-ingest-app:
@@ -77,6 +83,80 @@ services:
networks:
- vehicle-ingest
+ yutong-mqtt-app:
+ image: crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com/oneos/yutong-mqtt-app:${LINGNIU_IMAGE_VERSION:?set LINGNIU_IMAGE_VERSION}
+ container_name: yutong-mqtt-app
+ restart: unless-stopped
+ environment:
+ <<: *common-env
+ HTTP_PORT: 20500
+ KAFKA_CONSUMER_ENABLED: "false"
+ KAFKA_NODE_ID: ${KAFKA_NODE_ID_YUTONG_MQTT:-yutong-mqtt-portainer}
+ YUTONG_MQTT_ENABLED: ${YUTONG_MQTT_ENABLED:-false}
+ YUTONG_MQTT_ENDPOINT_NAME: ${YUTONG_MQTT_ENDPOINT_NAME:-yutong}
+ YUTONG_MQTT_URI: ${YUTONG_MQTT_URI:-}
+ YUTONG_MQTT_TOPIC: ${YUTONG_MQTT_TOPIC:-#}
+ YUTONG_MQTT_QOS: ${YUTONG_MQTT_QOS:-1}
+ YUTONG_MQTT_CLIENT_ID: ${YUTONG_MQTT_CLIENT_ID:-lingniu-yutong-mqtt}
+ YUTONG_MQTT_USERNAME: ${YUTONG_MQTT_USERNAME:-}
+ YUTONG_MQTT_PASSWORD: ${YUTONG_MQTT_PASSWORD:-}
+ YUTONG_MQTT_CLEAN_SESSION: ${YUTONG_MQTT_CLEAN_SESSION:-false}
+ YUTONG_MQTT_KEEP_ALIVE_SECONDS: ${YUTONG_MQTT_KEEP_ALIVE_SECONDS:-20}
+ YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS: ${YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS:-10}
+ YUTONG_MQTT_TLS_CA_PEM: ${YUTONG_MQTT_TLS_CA_PEM:-}
+ YUTONG_MQTT_TLS_CLIENT_PEM: ${YUTONG_MQTT_TLS_CLIENT_PEM:-}
+ YUTONG_MQTT_TLS_CLIENT_KEY: ${YUTONG_MQTT_TLS_CLIENT_KEY:-}
+ SINK_ARCHIVE_PATH: /archive/
+ VEHICLE_IDENTITY_STORE: ${VEHICLE_IDENTITY_STORE:-file}
+ VEHICLE_IDENTITY_FILE: /data/vehicle-identity.jsonl
+ VEHICLE_IDENTITY_MYSQL_TABLE: ${VEHICLE_IDENTITY_MYSQL_TABLE:-vehicle_identity_binding}
+ VEHICLE_IDENTITY_MYSQL_JDBC_URL: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:-}
+ VEHICLE_IDENTITY_MYSQL_USERNAME: ${VEHICLE_IDENTITY_MYSQL_USERNAME:-}
+ VEHICLE_IDENTITY_MYSQL_PASSWORD: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:-}
+ VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL: ${VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL:-60s}
+ ports:
+ - "${YUTONG_MQTT_HTTP_PORT:-20500}:20500"
+ volumes:
+ - vehicle-history-archive:/archive
+ - yutong-mqtt-data:/data
+ networks:
+ - vehicle-ingest
+
+ xinda-push-app:
+ image: crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com/oneos/xinda-push-app:${LINGNIU_IMAGE_VERSION:?set LINGNIU_IMAGE_VERSION}
+ container_name: xinda-push-app
+ restart: unless-stopped
+ environment:
+ <<: *common-env
+ HTTP_PORT: 20600
+ KAFKA_CONSUMER_ENABLED: "false"
+ KAFKA_NODE_ID: ${KAFKA_NODE_ID_XINDA_PUSH:-xinda-push-portainer}
+ XINDA_PUSH_ENABLED: ${XINDA_PUSH_ENABLED:-false}
+ XINDA_PUSH_HOST: ${XINDA_PUSH_HOST:-}
+ XINDA_PUSH_PORT: ${XINDA_PUSH_PORT:-10100}
+ XINDA_PUSH_USERNAME: ${XINDA_PUSH_USERNAME:-}
+ XINDA_PUSH_PASSWORD: ${XINDA_PUSH_PASSWORD:-}
+ XINDA_PUSH_CLIENT_DESC: ${XINDA_PUSH_CLIENT_DESC:-lingniu-ingest}
+ XINDA_PUSH_MSG_ID_LOCATION: ${XINDA_PUSH_MSG_ID_LOCATION:-0200}
+ XINDA_PUSH_MSG_ID_ALARM: ${XINDA_PUSH_MSG_ID_ALARM:-0300}
+ XINDA_PUSH_MSG_ID_PASSTHROUGH: ${XINDA_PUSH_MSG_ID_PASSTHROUGH:-0401}
+ XINDA_PUSH_RECONNECT_INTERVAL_SEC: ${XINDA_PUSH_RECONNECT_INTERVAL_SEC:-5}
+ SINK_ARCHIVE_PATH: /archive/
+ VEHICLE_IDENTITY_STORE: ${VEHICLE_IDENTITY_STORE:-file}
+ VEHICLE_IDENTITY_FILE: /data/vehicle-identity.jsonl
+ VEHICLE_IDENTITY_MYSQL_TABLE: ${VEHICLE_IDENTITY_MYSQL_TABLE:-vehicle_identity_binding}
+ VEHICLE_IDENTITY_MYSQL_JDBC_URL: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:-}
+ VEHICLE_IDENTITY_MYSQL_USERNAME: ${VEHICLE_IDENTITY_MYSQL_USERNAME:-}
+ VEHICLE_IDENTITY_MYSQL_PASSWORD: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:-}
+ VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL: ${VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL:-60s}
+ ports:
+ - "${XINDA_PUSH_HTTP_PORT:-20600}:20600"
+ volumes:
+ - vehicle-history-archive:/archive
+ - xinda-push-data:/data
+ networks:
+ - vehicle-ingest
+
vehicle-history-app:
image: crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com/oneos/vehicle-history-app:${LINGNIU_IMAGE_VERSION:?set LINGNIU_IMAGE_VERSION}
container_name: vehicle-history-app
@@ -84,6 +164,8 @@ services:
depends_on:
- gb32960-ingest-app
- jt808-ingest-app
+ - yutong-mqtt-app
+ - xinda-push-app
environment:
<<: *common-env
HTTP_PORT: 20200
@@ -123,6 +205,8 @@ services:
depends_on:
- gb32960-ingest-app
- jt808-ingest-app
+ - yutong-mqtt-app
+ - xinda-push-app
environment:
<<: *common-env
HTTP_PORT: 20300
@@ -147,6 +231,8 @@ networks:
volumes:
gb32960-ingest-data:
jt808-ingest-data:
+ yutong-mqtt-data:
+ xinda-push-data:
vehicle-history-archive:
vehicle-history-event-store:
vehicle-analytics-stat:
diff --git a/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java b/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java
index 4d01b568..b66239a5 100644
--- a/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java
+++ b/modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryKafkaConsumerConfiguration.java
@@ -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");
diff --git a/modules/apps/vehicle-history-app/src/main/resources/application.yml b/modules/apps/vehicle-history-app/src/main/resources/application.yml
index 270ac15a..6bcea3bf 100644
--- a/modules/apps/vehicle-history-app/src/main/resources/application.yml
+++ b/modules/apps/vehicle-history-app/src/main/resources/application.yml
@@ -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
diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java
index cee8c987..493ec4cf 100644
--- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java
+++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java
@@ -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() {
diff --git a/modules/apps/xinda-push-app/pom.xml b/modules/apps/xinda-push-app/pom.xml
new file mode 100644
index 00000000..487c7930
--- /dev/null
+++ b/modules/apps/xinda-push-app/pom.xml
@@ -0,0 +1,79 @@
+
+
+ 4.0.0
+
+ com.lingniu.ingest
+ lingniu-vehicle-ingest
+ 0.1.0-SNAPSHOT
+ ../../../pom.xml
+
+
+ xinda-push-app
+ xinda-push-app
+ Xinda Push ingress runtime: platform push client, frame parse, identity mapping, Kafka production, and raw archive.
+
+
+
+ com.lingniu.ingest
+ ingest-core
+
+
+ com.lingniu.ingest
+ observability
+
+
+ com.lingniu.ingest
+ vehicle-identity
+
+
+ com.lingniu.ingest
+ inbound-xinda-push
+
+
+ com.lingniu.ingest
+ sink-mq
+
+
+ com.lingniu.ingest
+ sink-archive
+
+
+ org.springframework.boot
+ spring-boot-starter-web
+
+
+ com.alibaba.cloud
+ spring-cloud-starter-alibaba-nacos-config
+
+
+ org.springdoc
+ springdoc-openapi-starter-webmvc-ui
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+
+
+ xinda-push-app
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+ --sun-misc-unsafe-memory-access=allow
+
+
+
+
+ repackage
+ build-info
+
+
+
+
+
+
+
diff --git a/modules/apps/xinda-push-app/src/main/java/com/lingniu/ingest/xindapushapp/XindaPushApplication.java b/modules/apps/xinda-push-app/src/main/java/com/lingniu/ingest/xindapushapp/XindaPushApplication.java
new file mode 100644
index 00000000..7d558c3d
--- /dev/null
+++ b/modules/apps/xinda-push-app/src/main/java/com/lingniu/ingest/xindapushapp/XindaPushApplication.java
@@ -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);
+ }
+ }
+}
diff --git a/modules/apps/xinda-push-app/src/main/resources/application.yml b/modules/apps/xinda-push-app/src/main/resources/application.yml
new file mode 100644
index 00000000..ee566266
--- /dev/null
+++ b/modules/apps/xinda-push-app/src/main/resources/application.yml
@@ -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
diff --git a/modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppCompositionTest.java b/modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppCompositionTest.java
new file mode 100644
index 00000000..a68d6d05
--- /dev/null
+++ b/modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppCompositionTest.java
@@ -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 kafkaProducer() {
+ return mock(KafkaProducer.class);
+ }
+}
diff --git a/modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppDefaultsTest.java b/modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppDefaultsTest.java
new file mode 100644
index 00000000..3c6914b1
--- /dev/null
+++ b/modules/apps/xinda-push-app/src/test/java/com/lingniu/ingest/xindapushapp/XindaPushAppDefaultsTest.java
@@ -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;
+ }
+}
diff --git a/modules/apps/yutong-mqtt-app/pom.xml b/modules/apps/yutong-mqtt-app/pom.xml
new file mode 100644
index 00000000..70c8ca1b
--- /dev/null
+++ b/modules/apps/yutong-mqtt-app/pom.xml
@@ -0,0 +1,79 @@
+
+
+ 4.0.0
+
+ com.lingniu.ingest
+ lingniu-vehicle-ingest
+ 0.1.0-SNAPSHOT
+ ../../../pom.xml
+
+
+ yutong-mqtt-app
+ yutong-mqtt-app
+ Yutong MQTT ingress runtime: MQTT subscribe, JSON parse, identity mapping, Kafka production, and raw archive.
+
+
+
+ com.lingniu.ingest
+ ingest-core
+
+
+ com.lingniu.ingest
+ observability
+
+
+ com.lingniu.ingest
+ vehicle-identity
+
+
+ com.lingniu.ingest
+ inbound-mqtt
+
+
+ com.lingniu.ingest
+ sink-mq
+
+
+ com.lingniu.ingest
+ sink-archive
+
+
+ org.springframework.boot
+ spring-boot-starter-web
+
+
+ com.alibaba.cloud
+ spring-cloud-starter-alibaba-nacos-config
+
+
+ org.springdoc
+ springdoc-openapi-starter-webmvc-ui
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+
+
+ yutong-mqtt-app
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+ --sun-misc-unsafe-memory-access=allow
+
+
+
+
+ repackage
+ build-info
+
+
+
+
+
+
+
diff --git a/modules/apps/yutong-mqtt-app/src/main/java/com/lingniu/ingest/yutongmqttapp/YutongMqttApplication.java b/modules/apps/yutong-mqtt-app/src/main/java/com/lingniu/ingest/yutongmqttapp/YutongMqttApplication.java
new file mode 100644
index 00000000..facdb868
--- /dev/null
+++ b/modules/apps/yutong-mqtt-app/src/main/java/com/lingniu/ingest/yutongmqttapp/YutongMqttApplication.java
@@ -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);
+ }
+ }
+}
diff --git a/modules/apps/yutong-mqtt-app/src/main/resources/application.yml b/modules/apps/yutong-mqtt-app/src/main/resources/application.yml
new file mode 100644
index 00000000..5b84ccb0
--- /dev/null
+++ b/modules/apps/yutong-mqtt-app/src/main/resources/application.yml
@@ -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
diff --git a/modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppCompositionTest.java b/modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppCompositionTest.java
new file mode 100644
index 00000000..11068ab4
--- /dev/null
+++ b/modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppCompositionTest.java
@@ -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 kafkaProducer() {
+ return mock(KafkaProducer.class);
+ }
+}
diff --git a/modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppDefaultsTest.java b/modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppDefaultsTest.java
new file mode 100644
index 00000000..efcafb4c
--- /dev/null
+++ b/modules/apps/yutong-mqtt-app/src/test/java/com/lingniu/ingest/yutongmqttapp/YutongMqttAppDefaultsTest.java
@@ -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;
+ }
+}
diff --git a/pom.xml b/pom.xml
index f00e4a3e..d4af1051 100644
--- a/pom.xml
+++ b/pom.xml
@@ -41,6 +41,8 @@
modules/apps/command-gateway
modules/apps/gb32960-ingest-app
modules/apps/jt808-ingest-app
+ modules/apps/yutong-mqtt-app
+ modules/apps/xinda-push-app
modules/apps/vehicle-history-app
modules/apps/vehicle-analytics-app
@@ -262,6 +264,16 @@
jt808-ingest-app
${project.version}
+
+ com.lingniu.ingest
+ yutong-mqtt-app
+ ${project.version}
+
+
+ com.lingniu.ingest
+ xinda-push-app
+ ${project.version}
+
com.lingniu.ingest
vehicle-history-app
diff --git a/woodpecker.yml b/woodpecker.yml
index 05d87d68..e2877fe0 100644
--- a/woodpecker.yml
+++ b/woodpecker.yml
@@ -13,7 +13,7 @@ steps:
commands: |
cd $CI_WORKSPACE
- mvn -B -ntp -am -pl :gb32960-ingest-app,:jt808-ingest-app,:vehicle-history-app,:vehicle-analytics-app package -DskipTests
+ mvn -B -ntp -am -pl :gb32960-ingest-app,:jt808-ingest-app,:yutong-mqtt-app,:xinda-push-app,:vehicle-history-app,:vehicle-analytics-app package -DskipTests
BRANCH_NAME=$(echo ${CI_COMMIT_BRANCH:-local} | tr / -)
PROJECT_VERSION=$(mvn -q -DforceStdout help:evaluate -Dexpression=project.version)
@@ -132,6 +132,78 @@ steps:
docker push $IMAGE
+ - name: docker-build-yutong-mqtt-app
+ image: docker:24.0.5-cli
+ when:
+ event:
+ - push
+ - pull_request
+ - manual
+ branch:
+ - master
+ - develop
+ - main
+ volumes:
+ - /var/run/docker.sock:/var/run/docker.sock
+ environment:
+ ALIYUN_ACR_USERNAME:
+ from_secret: aliyun_acr_username
+ ALIYUN_ACR_PASSWORD:
+ from_secret: aliyun_acr_password
+ commands: |
+ cd $CI_WORKSPACE
+ APP_NAME=yutong-mqtt-app
+ APP_VERSION=$(cat $CI_WORKSPACE/project_version.txt)
+ IMAGE=crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com/oneos/$APP_NAME:$APP_VERSION
+
+ docker build \
+ --build-arg APP_NAME=$APP_NAME \
+ --build-arg APP_VERSION=$APP_VERSION \
+ -t $IMAGE .
+
+ printf '%s' "$ALIYUN_ACR_PASSWORD" | docker login \
+ --username "$ALIYUN_ACR_USERNAME" \
+ --password-stdin \
+ crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com
+
+ docker push $IMAGE
+
+ - name: docker-build-xinda-push-app
+ image: docker:24.0.5-cli
+ when:
+ event:
+ - push
+ - pull_request
+ - manual
+ branch:
+ - master
+ - develop
+ - main
+ volumes:
+ - /var/run/docker.sock:/var/run/docker.sock
+ environment:
+ ALIYUN_ACR_USERNAME:
+ from_secret: aliyun_acr_username
+ ALIYUN_ACR_PASSWORD:
+ from_secret: aliyun_acr_password
+ commands: |
+ cd $CI_WORKSPACE
+ APP_NAME=xinda-push-app
+ APP_VERSION=$(cat $CI_WORKSPACE/project_version.txt)
+ IMAGE=crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com/oneos/$APP_NAME:$APP_VERSION
+
+ docker build \
+ --build-arg APP_NAME=$APP_NAME \
+ --build-arg APP_VERSION=$APP_VERSION \
+ -t $IMAGE .
+
+ printf '%s' "$ALIYUN_ACR_PASSWORD" | docker login \
+ --username "$ALIYUN_ACR_USERNAME" \
+ --password-stdin \
+ crpi-85r4m0ackrm3qpje.cn-shanghai.personal.cr.aliyuncs.com
+
+ docker push $IMAGE
+
- name: docker-build-vehicle-analytics-app
image: docker:24.0.5-cli
when: