refactor: slim history app tdengine runtime

This commit is contained in:
lingniu
2026-07-01 03:37:08 +08:00
parent 140b3f3995
commit 3908c1e752
6 changed files with 14 additions and 30 deletions

View File

@@ -29,10 +29,6 @@
<groupId>com.lingniu.ingest</groupId> <groupId>com.lingniu.ingest</groupId>
<artifactId>sink-archive</artifactId> <artifactId>sink-archive</artifactId>
</dependency> </dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>event-file-store</artifactId>
</dependency>
<dependency> <dependency>
<groupId>com.lingniu.ingest</groupId> <groupId>com.lingniu.ingest</groupId>
<artifactId>tdengine-history-store</artifactId> <artifactId>tdengine-history-store</artifactId>

View File

@@ -84,12 +84,6 @@ lingniu:
enabled: ${SINK_ARCHIVE_ENABLED:true} enabled: ${SINK_ARCHIVE_ENABLED:true}
type: local type: local
path: ${SINK_ARCHIVE_PATH:./archive/} path: ${SINK_ARCHIVE_PATH:./archive/}
event-file-store:
enabled: ${EVENT_FILE_STORE_ENABLED:false}
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}
tdengine-history: tdengine-history:
enabled: ${TDENGINE_HISTORY_ENABLED:false} enabled: ${TDENGINE_HISTORY_ENABLED:false}
database: ${TDENGINE_HISTORY_DATABASE:vehicle_history} database: ${TDENGINE_HISTORY_DATABASE:vehicle_history}

View File

@@ -1,10 +1,6 @@
package com.lingniu.ingest.historyapp; 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.api.consumer.EnvelopeDeadLetterSink; import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.eventhistory.EventHistoryController;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor; import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService; import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.Gb32960FrameController;
@@ -42,23 +38,21 @@ class VehicleHistoryAppCompositionTest {
Path tempDir; Path tempDir;
@Test @Test
void createsHistoryStorageAndQueryBeansWithoutGb32960TcpServer() { void createsTdengineHistoryStorageAndQueryBeansWithoutGb32960TcpServerOrEventFileStore() {
new ApplicationContextRunner() new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of( .withConfiguration(AutoConfigurations.of(
SinkArchiveAutoConfiguration.class, SinkArchiveAutoConfiguration.class,
EventFileStoreAutoConfiguration.class,
TdengineHistoryAutoConfiguration.class, TdengineHistoryAutoConfiguration.class,
SinkMqAutoConfiguration.class, SinkMqAutoConfiguration.class,
Gb32960AutoConfiguration.class, Gb32960AutoConfiguration.class,
EventHistoryAutoConfiguration.class)) EventHistoryAutoConfiguration.class))
.withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class)
.withAllowBeanDefinitionOverriding(true) .withAllowBeanDefinitionOverriding(true)
.withBean("kafkaProducer", KafkaProducer.class, VehicleHistoryAppCompositionTest::kafkaProducer) .withBean("kafkaProducer", KafkaProducer.class, VehicleHistoryAppCompositionTest::kafkaProducer)
.withPropertyValues( .withPropertyValues(
"lingniu.ingest.sink.archive.enabled=true", "lingniu.ingest.sink.archive.enabled=true",
"lingniu.ingest.sink.archive.type=local", "lingniu.ingest.sink.archive.type=local",
"lingniu.ingest.sink.archive.path=" + tempDir.resolve("archive"), "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.tdengine-history.enabled=true", "lingniu.ingest.tdengine-history.enabled=true",
"lingniu.ingest.tdengine-history.database=vehicle_history_test", "lingniu.ingest.tdengine-history.database=vehicle_history_test",
"lingniu.ingest.event-history.enabled=true", "lingniu.ingest.event-history.enabled=true",
@@ -73,14 +67,13 @@ class VehicleHistoryAppCompositionTest {
.run(context -> { .run(context -> {
assertThat(context).hasSingleBean(ArchiveStore.class); assertThat(context).hasSingleBean(ArchiveStore.class);
assertThat(context).hasSingleBean(RawArchiveEventSink.class); assertThat(context).hasSingleBean(RawArchiveEventSink.class);
assertThat(context).hasSingleBean(EventFileStore.class); assertThat(context).doesNotHaveBean("eventFileStore");
assertThat(context).hasSingleBean(EventFileStoreSink.class); assertThat(context).doesNotHaveBean("eventFileStoreSink");
assertThat(context).hasSingleBean(TdengineHistorySchema.class); assertThat(context).hasSingleBean(TdengineHistorySchema.class);
assertThat(context).hasSingleBean(KafkaEventSink.class); assertThat(context).hasSingleBean(KafkaEventSink.class);
assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class); assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class); assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context).hasSingleBean(EventHistoryController.class);
assertThat(context).hasSingleBean(Gb32960MessageDecoder.class); assertThat(context).hasSingleBean(Gb32960MessageDecoder.class);
assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class); assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class);
assertThat(context).hasSingleBean(LocationHistoryController.class); assertThat(context).hasSingleBean(LocationHistoryController.class);
@@ -99,11 +92,10 @@ class VehicleHistoryAppCompositionTest {
.withBean(TdengineHistoryWriter.class, () -> mock(TdengineHistoryWriter.class)) .withBean(TdengineHistoryWriter.class, () -> mock(TdengineHistoryWriter.class))
.withBean(EnvelopeDeadLetterSink.class, () -> mock(EnvelopeDeadLetterSink.class)) .withBean(EnvelopeDeadLetterSink.class, () -> mock(EnvelopeDeadLetterSink.class))
.withPropertyValues( .withPropertyValues(
"lingniu.ingest.event-file-store.enabled=false",
"lingniu.ingest.sink.mq.consumer.enabled=false") "lingniu.ingest.sink.mq.consumer.enabled=false")
.run(context -> { .run(context -> {
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class); assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context).doesNotHaveBean(EventFileStore.class); assertThat(context).doesNotHaveBean("eventFileStore");
}); });
} }

View File

@@ -23,7 +23,6 @@ class VehicleHistoryAppDefaultsTest {
.containsEntry("lingniu.ingest.gb32960.enabled", true) .containsEntry("lingniu.ingest.gb32960.enabled", true)
.containsEntry("lingniu.ingest.gb32960.server.enabled", false) .containsEntry("lingniu.ingest.gb32960.server.enabled", false)
.containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}") .containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}")
.containsEntry("lingniu.ingest.event-file-store.enabled", "${EVENT_FILE_STORE_ENABLED:false}")
.containsEntry("lingniu.ingest.tdengine-history.enabled", "${TDENGINE_HISTORY_ENABLED:false}") .containsEntry("lingniu.ingest.tdengine-history.enabled", "${TDENGINE_HISTORY_ENABLED:false}")
.containsEntry("lingniu.ingest.tdengine-history.database", "${TDENGINE_HISTORY_DATABASE:vehicle_history}") .containsEntry("lingniu.ingest.tdengine-history.database", "${TDENGINE_HISTORY_DATABASE:vehicle_history}")
.containsEntry( .containsEntry(
@@ -102,6 +101,9 @@ class VehicleHistoryAppDefaultsTest {
.containsEntry( .containsEntry(
"lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttRawEnvelopeConsumerProcessor.topics[0]", "lingniu.ingest.sink.mq.consumer.bindings.eventHistoryYutongMqttRawEnvelopeConsumerProcessor.topics[0]",
"${KAFKA_TOPIC_YUTONG_MQTT_RAW:vehicle.raw.mqtt-yutong.v1}"); "${KAFKA_TOPIC_YUTONG_MQTT_RAW:vehicle.raw.mqtt-yutong.v1}");
assertThat(properties.stringPropertyNames())
.noneMatch(name -> name.startsWith("lingniu.ingest.event-file-store."));
} }
private static Properties applicationProperties() { private static Properties applicationProperties() {

View File

@@ -9,7 +9,7 @@
</parent> </parent>
<artifactId>event-history-service</artifactId> <artifactId>event-history-service</artifactId>
<name>event-history-service</name> <name>event-history-service</name>
<description>Kafka full-field event consumer + Parquet/DuckDB historical query API.</description> <description>Kafka full-field event consumer + TDengine historical query API.</description>
<dependencies> <dependencies>
<dependency> <dependency>

View File

@@ -34,14 +34,14 @@ import java.nio.file.Path;
/** /**
* Event History 查询/消费服务自动装配。 * Event History 查询/消费服务自动装配。
* *
* <p>该模块有两入口: * <p>该模块有两入口:
* <ul> * <ul>
* <li>HTTP 查询入口:直接查询 {@link EventFileStore}GB32960 专用接口还会回读 RAW archive * <li>TDengine HTTP 查询入口:查询位置、RAW 帧和按需解码的 GB32960 帧。
* <li>Kafka consumer 入口:把其他服务投递的 envelope 再写入 {@link EventFileStore} * <li>兼容 EventFileStore 入口:仅在旧索引显式启用时创建旧记录查询和旧写入路径。
* </ul> * </ul>
* *
* <p>当前 32960 单体运行模式通常只启用 HTTP 查询和本地 event-file-store * <p>当前生产 history app 以 TDengine 为准Kafka consumer 是否启动还取决于
* Kafka consumer 是否启动还取决于 {@code lingniu.ingest.sink.mq.consumer.enabled=true}。 * {@code lingniu.ingest.sink.mq.consumer.enabled=true}。
*/ */
@AutoConfiguration @AutoConfiguration
@AutoConfigureAfter({ @AutoConfigureAfter({