treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kafkaSender for backend-api

Browse source
This commit is contained in:
ialbert 2023-05-26 16:56:08 +03:00
parent ac39aa5856
commit 7f542d6736
1 changed files with 42 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java
View file

@ -1,5 +1,8 @@
package ru.spcex.clearing.backendapi.config; package ru.spcex.clearing.backendapi.config;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.MockConsumer; import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.clients.consumer.OffsetResetStrategy;
@ -9,9 +12,23 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile; import org.springframework.context.annotation.Profile;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import ru.spcex.clearing.backendapi.config.element.BackendApiSettings; import ru.spcex.clearing.backendapi.config.element.BackendApiSettings;
import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider;
import javax.annotation.PostConstruct;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
@Configuration @Configuration
public class KafkaConfig { public class KafkaConfig {
@ -23,6 +40,31 @@ public class KafkaConfig {
return KafkaProducerFactory.producer(settings.getKafkaProducer()); return KafkaProducerFactory.producer(settings.getKafkaProducer());
} }
@Bean
public ProducerFactory<String, Object> pf(BackendApiSettings settings) {
KafkaProducerSettings kafkaSettings = settings.getKafkaProducer();
return KafkaProducerFactory.producerFactory(kafkaSettings);
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplateS(ProducerFactory<String, Object> pf) {
return new KafkaTemplate<>(pf);
}
@Profile("!kafkaDisabled")
@Autowired
@Bean
public KafkaSender kafkaSender(KafkaTemplate<String, Object> kafkaTemplate, ImdgProvider imdgProvider) {
ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
return KafkaSender.setup()
.saveRequestInfo(false)
.setKafkaTemplate(kafkaTemplate)
.idGenerator(imdgIdGenerator::nextId)
.build();
}
@Profile("kafkaDisabled") @Profile("kafkaDisabled")
@Bean @Bean
public MockProducer<String, Object> createMockProducer() { public MockProducer<String, Object> createMockProducer() {