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

http://jira.mfd.msk:8088/browse/CLS-36

Browse source
This commit is contained in:
ialbert 2022-09-28 18:49:27 +03:00
parent 2e46a484ff
commit 15cad890c9
2 changed files with 20 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java
View file

@ -12,27 +12,26 @@ import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.ImdgProvider;
import java.util.function.Supplier;
@Configuration @Configuration
public class KafkaConfig { public class KafkaConfig {
@Autowired @Autowired
@Bean @Bean
public Producer<String, Object> createProducer(ImportDBFServiceSettings settings) { public Supplier<KafkaSender> kafkaSender(ImportDBFServiceSettings settings, ImdgProvider imdgProvider) {
return KafkaProducerFactory.producer(settings.getKafkaProducer()); return () -> {
} Producer<String, Object> kafkaProducer = KafkaProducerFactory.producer(settings.getKafkaProducer());
ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
@Autowired return KafkaSender
@Bean .setup()
public KafkaSender kafkaSender(Producer<String, Object> kafkaProducer, ImdgProvider imdgProvider) { .producer(kafkaProducer)
ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator(); .idGenerator(imdgIdGenerator::nextId)
return KafkaSender .imdgProvider(s -> {
.setup() Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
.producer(kafkaProducer) return imdg::insert;
.idGenerator(imdgIdGenerator::nextId) })
.imdgProvider(s -> { .build();
Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class); };
return imdg::insert;
})
.build();
} }
} }

7
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ImportToDB.java
View file

@ -18,6 +18,7 @@ import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.nio.charset.Charset; import java.nio.charset.Charset;
import java.util.Map; import java.util.Map;
import java.util.function.Supplier;
/** /**
* Заливка проверенных данных в базу * Заливка проверенных данных в базу
@ -27,11 +28,11 @@ public class ImportToDB extends Stage {
private final AProperties properties; private final AProperties properties;
private final HazelcastService hazelcastService; private final HazelcastService hazelcastService;
private final Map<ETable, AbstractTable> mappingEnumTableObjectTable; private final Map<ETable, AbstractTable> mappingEnumTableObjectTable;
private final KafkaSender kafka; private final Supplier<KafkaSender> kafka;
public ImportToDB(@Qualifier("dbfImporterProperties") AProperties properties, public ImportToDB(@Qualifier("dbfImporterProperties") AProperties properties,
HazelcastService hazelcastService, @Qualifier("mapOfTable") Map<ETable, AbstractTable> mappingEnumTableObjectTable, HazelcastService hazelcastService, @Qualifier("mapOfTable") Map<ETable, AbstractTable> mappingEnumTableObjectTable,
KafkaSender kafka) { Supplier<KafkaSender> kafka) {
this.properties = properties; this.properties = properties;
this.hazelcastService = hazelcastService; this.hazelcastService = hazelcastService;
this.mappingEnumTableObjectTable = mappingEnumTableObjectTable; this.mappingEnumTableObjectTable = mappingEnumTableObjectTable;
@ -70,6 +71,6 @@ public class ImportToDB extends Stage {
private void sendStatementRequest(Long fileId) { private void sendStatementRequest(Long fileId) {
StatementRequest statementRequest = new StatementRequest(); StatementRequest statementRequest = new StatementRequest();
statementRequest.setSdf01GroupId(fileId); statementRequest.setSdf01GroupId(fileId);
kafka.sendRequestToQueue(Consts.STATEMENT_PROCESS, statementRequest); kafka.get().sendRequestToQueue(Consts.STATEMENT_PROCESS, statementRequest);
} }
} }