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

refactor: centralize history raw consumer processor

Browse Source
This commit is contained in:
lingniu
2026-07-01 04:54:11 +08:00
parent 84a098d84c
commit b841141fed
4 changed files with 16 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,8 +1,6 @@
package com.lingniu.ingest.historyapp; package com.lingniu.ingest.historyapp;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor; import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerFactory; import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerFactory;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerRunner; import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerRunner;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerWorker; import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerWorker;
@@ -20,13 +18,6 @@ import java.util.Map;
@Configuration(proxyBeanMethods = false) @Configuration(proxyBeanMethods = false)
public class VehicleHistoryKafkaConsumerConfiguration { public class VehicleHistoryKafkaConsumerConfiguration {
@Bean
@ConditionalOnMissingBean(name = "eventHistoryRawEnvelopeConsumerProcessor")
public EnvelopeConsumerProcessor eventHistoryRawEnvelopeConsumerProcessor(EventHistoryEnvelopeIngestor ingestor,
EnvelopeDeadLetterSink deadLetterSink) {
return new EnvelopeConsumerProcessor("event-history-raw", ingestor, deadLetterSink);
}
@Bean @Bean
@ConditionalOnMissingBean @ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq.consumer", name = "enabled", havingValue = "true") @ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq.consumer", name = "enabled", havingValue = "true")

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

@@ -61,7 +61,8 @@ class VehicleHistoryAppCompositionTest {
assertThat(beanMethods) assertThat(beanMethods)
.doesNotContain("telemetryEnvelopeRecordMapper") .doesNotContain("telemetryEnvelopeRecordMapper")
.doesNotContain("eventHistoryEnvelopeIngestor") .doesNotContain("eventHistoryEnvelopeIngestor")
.doesNotContain("eventHistoryEnvelopeConsumerProcessor"); .doesNotContain("eventHistoryEnvelopeConsumerProcessor")
.doesNotContain("eventHistoryRawEnvelopeConsumerProcessor");
} }
@Test @Test

8
modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
View File

@@ -87,6 +87,14 @@ public class EventHistoryAutoConfiguration {
return new EnvelopeConsumerProcessor("event-history", ingestor, deadLetterSink); return new EnvelopeConsumerProcessor("event-history", ingestor, deadLetterSink);
} }
@Bean
@ConditionalOnBean({EventHistoryEnvelopeIngestor.class, EnvelopeDeadLetterSink.class})
@ConditionalOnMissingBean(name = "eventHistoryRawEnvelopeConsumerProcessor")
public EnvelopeConsumerProcessor eventHistoryRawEnvelopeConsumerProcessor(EventHistoryEnvelopeIngestor ingestor,
EnvelopeDeadLetterSink deadLetterSink) {
return new EnvelopeConsumerProcessor("event-history-raw", ingestor, deadLetterSink);
}
@Bean @Bean
@ConditionalOnBean(EventFileStore.class) @ConditionalOnBean(EventFileStore.class)
@ConditionalOnMissingBean @ConditionalOnMissingBean

8
modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
View File

@@ -43,7 +43,9 @@ class EventHistoryAutoConfigurationTest {
assertThat(context).hasSingleBean(TelemetryEnvelopeRecordMapper.class); assertThat(context).hasSingleBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class); assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class)) assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class))
.containsOnlyKeys("eventHistoryEnvelopeConsumerProcessor"); .containsOnlyKeys(
"eventHistoryEnvelopeConsumerProcessor",
"eventHistoryRawEnvelopeConsumerProcessor");
assertThat(context).hasSingleBean(EventHistoryController.class); assertThat(context).hasSingleBean(EventHistoryController.class);
assertThat(context).doesNotHaveBean(Gb32960DecodedFrameService.class); assertThat(context).doesNotHaveBean(Gb32960DecodedFrameService.class);
assertThat(context).doesNotHaveBean(Gb32960FrameController.class); assertThat(context).doesNotHaveBean(Gb32960FrameController.class);
@@ -104,7 +106,9 @@ class EventHistoryAutoConfigurationTest {
assertThat(context).hasSingleBean(TelemetryEnvelopeRecordMapper.class); assertThat(context).hasSingleBean(TelemetryEnvelopeRecordMapper.class);
assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class); assertThat(context).hasSingleBean(EventHistoryEnvelopeIngestor.class);
assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class)) assertThat(context.getBeansOfType(EnvelopeConsumerProcessor.class))
.containsOnlyKeys("eventHistoryEnvelopeConsumerProcessor"); .containsOnlyKeys(
"eventHistoryEnvelopeConsumerProcessor",
"eventHistoryRawEnvelopeConsumerProcessor");
}); });
} }