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

fix: create history consumer processor in app runtime

Browse Source
This commit is contained in:
lingniu
2026-06-29 14:58:19 +08:00
parent 00eee9a769
commit 8fe0adcac9
1 changed files with 27 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,11 +1,17 @@
package com.lingniu.ingest.historyapp;
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
import com.lingniu.ingest.eventfilestore.EventFileStore;
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.KafkaEnvelopeConsumerRunner;
import com.lingniu.ingest.sink.mq.KafkaEnvelopeConsumerWorker;
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.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
@@ -18,6 +24,27 @@ import java.util.Map;
@Configuration(proxyBeanMethods = false)
public class VehicleHistoryKafkaConsumerConfiguration {
@Bean
@ConditionalOnMissingBean
public TelemetryEnvelopeRecordMapper telemetryEnvelopeRecordMapper() {
return new TelemetryEnvelopeRecordMapper();
}
@Bean
@ConditionalOnMissingBean
public EventHistoryEnvelopeIngestor eventHistoryEnvelopeIngestor(EventFileStore store,
TelemetryEnvelopeRecordMapper mapper,
ObjectProvider<TdengineHistoryWriter> writer) {
return new EventHistoryEnvelopeIngestor(store, mapper, writer.getIfAvailable());
}
@Bean
@ConditionalOnMissingBean(name = "eventHistoryEnvelopeConsumerProcessor")
public EnvelopeConsumerProcessor eventHistoryEnvelopeConsumerProcessor(EventHistoryEnvelopeIngestor ingestor,
EnvelopeDeadLetterSink deadLetterSink) {
return new EnvelopeConsumerProcessor("event-history", ingestor, deadLetterSink);
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq.consumer", name = "enabled", havingValue = "true")