refactor: slim history app consumer config
This commit is contained in:
@@ -3,15 +3,11 @@ 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.api.consumer.EnvelopeDeadLetterSink;
|
||||||
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
|
import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor;
|
||||||
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
|
|
||||||
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;
|
||||||
import com.lingniu.ingest.sink.mq.SinkMqProperties;
|
import com.lingniu.ingest.sink.mq.SinkMqProperties;
|
||||||
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
|
|
||||||
import org.springframework.beans.factory.annotation.Qualifier;
|
import org.springframework.beans.factory.annotation.Qualifier;
|
||||||
import org.springframework.beans.factory.annotation.Value;
|
|
||||||
import org.springframework.beans.factory.ObjectProvider;
|
|
||||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
import org.springframework.context.annotation.Bean;
|
import org.springframework.context.annotation.Bean;
|
||||||
@@ -24,29 +20,6 @@ import java.util.Map;
|
|||||||
@Configuration(proxyBeanMethods = false)
|
@Configuration(proxyBeanMethods = false)
|
||||||
public class VehicleHistoryKafkaConsumerConfiguration {
|
public class VehicleHistoryKafkaConsumerConfiguration {
|
||||||
|
|
||||||
@Bean
|
|
||||||
@ConditionalOnMissingBean
|
|
||||||
public TelemetryEnvelopeRecordMapper telemetryEnvelopeRecordMapper() {
|
|
||||||
return new TelemetryEnvelopeRecordMapper();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
@ConditionalOnMissingBean
|
|
||||||
public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(TelemetryEnvelopeRecordMapper mapper,
|
|
||||||
ObjectProvider<TdengineHistoryWriter> writer,
|
|
||||||
@Value("${lingniu.ingest.tdengine-history.telemetry-fields-enabled:false}")
|
|
||||||
boolean telemetryFieldsEnabled) {
|
|
||||||
TdengineHistoryWriter tdengineWriter = writer.getIfAvailable();
|
|
||||||
return new EventHistoryEnvelopeIngestor(mapper, tdengineWriter, telemetryFieldsEnabled);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
@ConditionalOnMissingBean(name = "eventHistoryEnvelopeConsumerProcessor")
|
|
||||||
public EnvelopeConsumerProcessor eventHistoryEnvelopeConsumerProcessor(EventHistoryEnvelopeIngestor ingestor,
|
|
||||||
EnvelopeDeadLetterSink deadLetterSink) {
|
|
||||||
return new EnvelopeConsumerProcessor("event-history", ingestor, deadLetterSink);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
@Bean
|
||||||
@ConditionalOnMissingBean(name = "eventHistoryRawEnvelopeConsumerProcessor")
|
@ConditionalOnMissingBean(name = "eventHistoryRawEnvelopeConsumerProcessor")
|
||||||
public EnvelopeConsumerProcessor eventHistoryRawEnvelopeConsumerProcessor(EventHistoryEnvelopeIngestor ingestor,
|
public EnvelopeConsumerProcessor eventHistoryRawEnvelopeConsumerProcessor(EventHistoryEnvelopeIngestor ingestor,
|
||||||
|
|||||||
@@ -32,9 +32,12 @@ import org.junit.jupiter.api.Test;
|
|||||||
import org.junit.jupiter.api.io.TempDir;
|
import org.junit.jupiter.api.io.TempDir;
|
||||||
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
|
||||||
|
import java.lang.reflect.Method;
|
||||||
import java.nio.file.Path;
|
import java.nio.file.Path;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
|
import java.util.Arrays;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
|
||||||
import static org.assertj.core.api.Assertions.assertThat;
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
@@ -48,6 +51,19 @@ class VehicleHistoryAppCompositionTest {
|
|||||||
@TempDir
|
@TempDir
|
||||||
Path tempDir;
|
Path tempDir;
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void kafkaConsumerConfigurationDoesNotRedeclareServiceOwnedHistoryBeans() {
|
||||||
|
List<String> beanMethods = Arrays.stream(VehicleHistoryKafkaConsumerConfiguration.class.getDeclaredMethods())
|
||||||
|
.filter(method -> method.isAnnotationPresent(Bean.class))
|
||||||
|
.map(Method::getName)
|
||||||
|
.toList();
|
||||||
|
|
||||||
|
assertThat(beanMethods)
|
||||||
|
.doesNotContain("telemetryEnvelopeRecordMapper")
|
||||||
|
.doesNotContain("eventHistoryEnvelopeIngestor")
|
||||||
|
.doesNotContain("eventHistoryEnvelopeConsumerProcessor");
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
void createsTdengineHistoryStorageAndQueryBeansWithoutGb32960TcpServerOrEventFileStore() {
|
void createsTdengineHistoryStorageAndQueryBeansWithoutGb32960TcpServerOrEventFileStore() {
|
||||||
new ApplicationContextRunner()
|
new ApplicationContextRunner()
|
||||||
@@ -99,10 +115,12 @@ class VehicleHistoryAppCompositionTest {
|
|||||||
@Test
|
@Test
|
||||||
void createsHistoryIngestorWithoutEventFileStoreForTdengineOnlyRuntime() {
|
void createsHistoryIngestorWithoutEventFileStoreForTdengineOnlyRuntime() {
|
||||||
new ApplicationContextRunner()
|
new ApplicationContextRunner()
|
||||||
|
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
|
||||||
.withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class)
|
.withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class)
|
||||||
.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-history.enabled=true",
|
||||||
"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);
|
||||||
@@ -116,11 +134,14 @@ class VehicleHistoryAppCompositionTest {
|
|||||||
CapturingTdengineWriter writer = new CapturingTdengineWriter();
|
CapturingTdengineWriter writer = new CapturingTdengineWriter();
|
||||||
|
|
||||||
new ApplicationContextRunner()
|
new ApplicationContextRunner()
|
||||||
|
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
|
||||||
.withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class)
|
.withUserConfiguration(VehicleHistoryKafkaConsumerConfiguration.class)
|
||||||
.withBean(EventFileStore.class, () -> eventFileStore)
|
.withBean(EventFileStore.class, () -> eventFileStore)
|
||||||
.withBean(TdengineHistoryWriter.class, () -> writer)
|
.withBean(TdengineHistoryWriter.class, () -> writer)
|
||||||
.withBean(EnvelopeDeadLetterSink.class, () -> mock(EnvelopeDeadLetterSink.class))
|
.withBean(EnvelopeDeadLetterSink.class, () -> mock(EnvelopeDeadLetterSink.class))
|
||||||
.withPropertyValues("lingniu.ingest.sink.mq.consumer.enabled=false")
|
.withPropertyValues(
|
||||||
|
"lingniu.ingest.event-history.enabled=true",
|
||||||
|
"lingniu.ingest.sink.mq.consumer.enabled=false")
|
||||||
.run(context -> {
|
.run(context -> {
|
||||||
EventHistoryEnvelopeIngestor ingestor = context.getBean(EventHistoryEnvelopeIngestor.class);
|
EventHistoryEnvelopeIngestor ingestor = context.getBean(EventHistoryEnvelopeIngestor.class);
|
||||||
|
|
||||||
|
|||||||
@@ -61,7 +61,7 @@ public class EventHistoryAutoConfiguration {
|
|||||||
|
|
||||||
@Bean
|
@Bean
|
||||||
@ConditionalOnBean(EventFileStore.class)
|
@ConditionalOnBean(EventFileStore.class)
|
||||||
@ConditionalOnMissingBean
|
@ConditionalOnMissingBean(value = {TdengineHistoryWriter.class, EventHistoryEnvelopeIngestor.class})
|
||||||
public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(EventFileStore store,
|
public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(EventFileStore store,
|
||||||
TelemetryEnvelopeRecordMapper mapper,
|
TelemetryEnvelopeRecordMapper mapper,
|
||||||
ObjectProvider<TdengineHistoryWriter> writer) {
|
ObjectProvider<TdengineHistoryWriter> writer) {
|
||||||
|
|||||||
Reference in New Issue
Block a user