http://jira.mfd.msk:8088/browse/CLS-73 выгрузка sdf10 добавил заготовку для журналирования
This commit is contained in:
parent
d25e8d9f4f
commit
07443234e1
3 changed files with 53 additions and 31 deletions
|
|
@ -8,18 +8,11 @@ import org.springframework.context.annotation.Bean;
|
||||||
import org.springframework.context.annotation.Configuration;
|
import org.springframework.context.annotation.Configuration;
|
||||||
import org.springframework.context.annotation.Scope;
|
import org.springframework.context.annotation.Scope;
|
||||||
import ru.spcex.clearing.dbf.exporter.config.settings.ExportDBFServiceSettings;
|
import ru.spcex.clearing.dbf.exporter.config.settings.ExportDBFServiceSettings;
|
||||||
import ru.spcex.clearing.imdg.IMDGDistributedNames;
|
|
||||||
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.service.RequestInfo;
|
|
||||||
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
|
|
||||||
import ru.spcex.platform.imdg.api.Imdg;
|
|
||||||
import ru.spcex.platform.imdg.api.ImdgId;
|
|
||||||
import ru.spcex.platform.imdg.api.ImdgProvider;
|
|
||||||
|
|
||||||
@Configuration
|
@Configuration
|
||||||
public class KafkaConfig {
|
public class KafkaConfig {
|
||||||
|
|
||||||
@Autowired
|
@Autowired
|
||||||
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||||
@Bean
|
@Bean
|
||||||
|
|
@ -35,22 +28,4 @@ public class KafkaConfig {
|
||||||
} else
|
} else
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Autowired(required = false)
|
|
||||||
@Bean
|
|
||||||
public KafkaSender kafkaSender(Producer<String, Object> kafkaProducer, ImdgProvider imdgProvider) {
|
|
||||||
if (kafkaProducer == null ||imdgProvider == null) {
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
|
|
||||||
return KafkaSender
|
|
||||||
.setup()
|
|
||||||
.producer(kafkaProducer)
|
|
||||||
.idGenerator(imdgIdGenerator::nextId)
|
|
||||||
.imdgProvider(s -> {
|
|
||||||
Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
|
|
||||||
return imdg::insert;
|
|
||||||
})
|
|
||||||
.build();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,47 @@
|
||||||
|
package ru.spcex.clearing.dbf.exporter.config;
|
||||||
|
|
||||||
|
import org.apache.kafka.clients.producer.Producer;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
import ru.spcex.clearing.imdg.IMDGDistributedNames;
|
||||||
|
import ru.spcex.clearing.platform.messaging.service.RequestInfo;
|
||||||
|
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
|
||||||
|
import ru.spcex.platform.imdg.api.Imdg;
|
||||||
|
import ru.spcex.platform.imdg.api.ImdgId;
|
||||||
|
import ru.spcex.platform.imdg.api.ImdgProvider;
|
||||||
|
|
||||||
|
//отдельный конфиг для sender чтобы сделать required false
|
||||||
|
@Configuration
|
||||||
|
public class KafkaSenderConfig {
|
||||||
|
|
||||||
|
private final ImdgProvider imdgProvider;
|
||||||
|
private Producer<String, Object> kafkaProducer;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public KafkaSenderConfig(ImdgProvider imdgProvider) {
|
||||||
|
this.imdgProvider = imdgProvider;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public void setKafkaProducer(Producer<String, Object> kafkaProducer) {
|
||||||
|
this.kafkaProducer = kafkaProducer;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public KafkaSender kafkaSender() {
|
||||||
|
if (kafkaProducer == null ||imdgProvider == null) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
|
||||||
|
return KafkaSender
|
||||||
|
.setup()
|
||||||
|
.producer(kafkaProducer)
|
||||||
|
.idGenerator(imdgIdGenerator::nextId)
|
||||||
|
.imdgProvider(s -> {
|
||||||
|
Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
|
||||||
|
return imdg::insert;
|
||||||
|
})
|
||||||
|
.build();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -16,9 +16,6 @@ import java.time.LocalTime;
|
||||||
import java.util.EnumMap;
|
import java.util.EnumMap;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
|
||||||
/**
|
|
||||||
* Выгрузка данных из мапы hazelcast и их запись в файлы
|
|
||||||
*/
|
|
||||||
@Component
|
@Component
|
||||||
public class Journal extends Stage implements InitializingBean {
|
public class Journal extends Stage implements InitializingBean {
|
||||||
private final Logger log = LoggerFactory.getLogger(getClass());
|
private final Logger log = LoggerFactory.getLogger(getClass());
|
||||||
|
|
@ -27,9 +24,10 @@ public class Journal extends Stage implements InitializingBean {
|
||||||
@Autowired(required = false)
|
@Autowired(required = false)
|
||||||
public Journal(KafkaSender kafkaSender) {
|
public Journal(KafkaSender kafkaSender) {
|
||||||
this.kafkaSender = kafkaSender;
|
this.kafkaSender = kafkaSender;
|
||||||
if (kafkaSender == null) {
|
}
|
||||||
log.warn("kafka producer settings missing; kafka messages to journal settings are not enabled.");
|
|
||||||
}
|
public Journal() {
|
||||||
|
this.kafkaSender = null;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
@ -40,6 +38,7 @@ public class Journal extends Stage implements InitializingBean {
|
||||||
@Override
|
@Override
|
||||||
public StageResult process(ResultContainer resultContainer) {
|
public StageResult process(ResultContainer resultContainer) {
|
||||||
if (kafkaSender == null) {
|
if (kafkaSender == null) {
|
||||||
|
log.info("kafka producer settings missing; kafka messages to journal-service are not enabled.");
|
||||||
return StageResult.COMPLETE;
|
return StageResult.COMPLETE;
|
||||||
}
|
}
|
||||||
JournalSdf journalSdf = new JournalSdf();
|
JournalSdf journalSdf = new JournalSdf();
|
||||||
|
|
@ -50,6 +49,7 @@ public class Journal extends Stage implements InitializingBean {
|
||||||
journalSdf.setDocumentName(documentNames.get(resultContainer.getTableForExport()));
|
journalSdf.setDocumentName(documentNames.get(resultContainer.getTableForExport()));
|
||||||
journalSdf.setDossierNumber(dossierNumber.get(resultContainer.getTableForExport()));
|
journalSdf.setDossierNumber(dossierNumber.get(resultContainer.getTableForExport()));
|
||||||
journalSdf.setResultStatus(StageResult.ERROR.equals(resultContainer.getLastStageResult()) ? "NACK" : "ACK");
|
journalSdf.setResultStatus(StageResult.ERROR.equals(resultContainer.getLastStageResult()) ? "NACK" : "ACK");
|
||||||
|
// kafkaSender.sendRequestToQueue();
|
||||||
return StageResult.COMPLETE;
|
return StageResult.COMPLETE;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue