http://jira.mfd.msk:8088/browse/CLS-73 выгрузка sdf10 добавил заготовку для журналирования
This commit is contained in:
ialbert
parent
e463d83084
commit
d25e8d9f4f
9 changed files with 248 additions and 2 deletions
|
|
@ -1,13 +1,21 @@
|
|||
package ru.spcex.clearing.dbf.exporter.config;
|
||||
|
||||
import org.apache.kafka.clients.consumer.Consumer;
|
||||
import org.apache.kafka.clients.producer.Producer;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Scope;
|
||||
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.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
|
||||
public class KafkaConfig {
|
||||
|
|
@ -18,4 +26,31 @@ public class KafkaConfig {
|
|||
public Consumer<String, Object> createConsumer(ExportDBFServiceSettings settings) {
|
||||
return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
|
||||
}
|
||||
|
||||
@Autowired
|
||||
@Bean
|
||||
public Producer<String, Object> createProducer(ExportDBFServiceSettings settings) {
|
||||
if (settings.getKafkaProducer() != null) {
|
||||
return KafkaProducerFactory.producer(settings.getKafkaProducer());
|
||||
} else
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
|
|||
import org.springframework.context.annotation.PropertySource;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
|
||||
import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
|
||||
import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
|
||||
|
||||
@Component
|
||||
|
|
@ -12,6 +13,7 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
|
|||
public class ExportDBFServiceSettings {
|
||||
private HazelcastClientParams hazelcast;
|
||||
private KafkaConsumerSettings kafkaConsumer;
|
||||
private KafkaProducerSettings kafkaProducer;
|
||||
private Common common;
|
||||
private Store store;
|
||||
private Cron cron;
|
||||
|
|
@ -55,4 +57,12 @@ public class ExportDBFServiceSettings {
|
|||
public void setCron(Cron cron) {
|
||||
this.cron = cron;
|
||||
}
|
||||
|
||||
public KafkaProducerSettings getKafkaProducer() {
|
||||
return kafkaProducer;
|
||||
}
|
||||
|
||||
public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
|
||||
this.kafkaProducer = kafkaProducer;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,9 +25,12 @@ public class Processor {
|
|||
StageResult result = null;
|
||||
long startMills = System.currentTimeMillis();
|
||||
for (Stage currStage : pipeline) {
|
||||
if (result != null
|
||||
&& Arrays.asList(StageResult.ERROR, StageResult.COMPLETE).contains(result)
|
||||
&& currStage.skipCompleted()) continue;
|
||||
log.info("uuid {}. Stage: {}", task.getUuid(), currStage.getClass().getSimpleName());
|
||||
result = currStage.process(task);
|
||||
if (Arrays.asList(StageResult.ERROR, StageResult.COMPLETE).contains(result)) break;
|
||||
task.setLastStageResult(result);
|
||||
}
|
||||
long endMills = System.currentTimeMillis();
|
||||
log.info("uuid {}. Task completed, result: {}, time working: {} ms",
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
package ru.spcex.clearing.dbf.exporter.logic.data;
|
||||
|
||||
import ru.spcex.clearing.dbf.exporter.logic.data.enums.StageResult;
|
||||
import ru.spcex.clearing.dbf.exporter.logic.data.enums.Table;
|
||||
|
||||
import java.io.File;
|
||||
|
|
@ -13,6 +14,7 @@ public class ResultContainer {
|
|||
private Table tableForExport;
|
||||
private File fileForExport;
|
||||
private Long groupId;
|
||||
private StageResult lastStageResult;
|
||||
|
||||
protected ResultContainer() {}
|
||||
|
||||
|
|
@ -54,4 +56,12 @@ public class ResultContainer {
|
|||
public void setGroupId(Long groupId) {
|
||||
this.groupId = groupId;
|
||||
}
|
||||
|
||||
public StageResult getLastStageResult() {
|
||||
return lastStageResult;
|
||||
}
|
||||
|
||||
public void setLastStageResult(StageResult lastStageResult) {
|
||||
this.lastStageResult = lastStageResult;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -111,7 +111,7 @@ public class ExportFromHazelcast extends Stage implements InitializingBean {
|
|||
}
|
||||
}
|
||||
|
||||
return StageResult.COMPLETE;
|
||||
return StageResult.OK;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -0,0 +1,83 @@
|
|||
package ru.spcex.clearing.dbf.exporter.logic.stages;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.spcex.clearing.dbf.exporter.logic.data.ResultContainer;
|
||||
import ru.spcex.clearing.dbf.exporter.logic.data.enums.StageResult;
|
||||
import ru.spcex.clearing.dbf.exporter.logic.data.enums.Table;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.system.JournalSdf;
|
||||
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalTime;
|
||||
import java.util.EnumMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Выгрузка данных из мапы hazelcast и их запись в файлы
|
||||
*/
|
||||
@Component
|
||||
public class Journal extends Stage implements InitializingBean {
|
||||
private final Logger log = LoggerFactory.getLogger(getClass());
|
||||
private final KafkaSender kafkaSender;
|
||||
|
||||
@Autowired(required = false)
|
||||
public Journal(KafkaSender kafkaSender) {
|
||||
this.kafkaSender = kafkaSender;
|
||||
if (kafkaSender == null) {
|
||||
log.warn("kafka producer settings missing; kafka messages to journal settings are not enabled.");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean skipCompleted() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public StageResult process(ResultContainer resultContainer) {
|
||||
if (kafkaSender == null) {
|
||||
return StageResult.COMPLETE;
|
||||
}
|
||||
JournalSdf journalSdf = new JournalSdf();
|
||||
//todo read file attibutes
|
||||
journalSdf.setRegistrationDate(LocalDate.now());
|
||||
journalSdf.setRegistrationTime(LocalTime.now());
|
||||
journalSdf.setRegistrationNumber(resultContainer.getGroupId());
|
||||
journalSdf.setDocumentName(documentNames.get(resultContainer.getTableForExport()));
|
||||
journalSdf.setDossierNumber(dossierNumber.get(resultContainer.getTableForExport()));
|
||||
journalSdf.setResultStatus(StageResult.ERROR.equals(resultContainer.getLastStageResult()) ? "NACK" : "ACK");
|
||||
return StageResult.COMPLETE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
}
|
||||
|
||||
private static final Map<Table, String> documentNames = new EnumMap<>(Table.class);
|
||||
static {
|
||||
documentNames.put(Table.S_DF02, "Уведомлений об исполнении операции загрузки денежных средств или уведомление об ошибке");
|
||||
documentNames.put(Table.S_DF03, "Сводное платёжное поручение по итогу проведения расчетов, направляемое в РО");
|
||||
//documentNames.put(Table.S_DF05, "Уведомление о завершении расчетов в секции");
|
||||
documentNames.put(Table.S_DF08, "Запрос остатков по всем счетам, направляемый в РО");
|
||||
documentNames.put(Table.S_DF10, "Подтверждение о загрузке по поступлению на клиринговый счет");
|
||||
documentNames.put(Table.S_DF11, "Распоряжение на списание с ТБС УК на Клиринговый счет (по итогам проведения расчетов по возврату депозита) / Распоряжение на списание с ТБС УК на Корреспонденский счет УК");
|
||||
//documentNames.put(Table.S_DF13, "Распоряжение на списание денежных средств УК категории В (с клирингового счета)");
|
||||
documentNames.put(Table.S_DF17, "Подтверждение о загрузке Уведомления о возврате ден.ср. по договору депозита");
|
||||
}
|
||||
|
||||
private static final Map<Table, String> dossierNumber = new EnumMap<>(Table.class);
|
||||
static {
|
||||
dossierNumber.put(Table.S_DF02, "07-50");
|
||||
dossierNumber.put(Table.S_DF03, "07-51");
|
||||
//dossierNumber.put(Table.S_DF05, "07-53");
|
||||
dossierNumber.put(Table.S_DF08, "07-55");
|
||||
dossierNumber.put(Table.S_DF10, "07-56");
|
||||
dossierNumber.put(Table.S_DF11, "07-36");
|
||||
//dossierNumber.put(Table.S_DF13, "07-48");
|
||||
dossierNumber.put(Table.S_DF17, "07-58");
|
||||
}
|
||||
}
|
||||
|
|
@ -9,4 +9,8 @@ public abstract class Stage {
|
|||
protected Logger log = LoggerFactory.getLogger(getClass());
|
||||
|
||||
public abstract StageResult process(ResultContainer resultContainer);
|
||||
|
||||
public boolean skipCompleted() {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,79 @@
|
|||
package ru.spcex.clearing.platform.messaging.domain.cud.system;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
|
||||
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
|
||||
import ru.spcex.clearing.platform.messaging.domain.json.deserialize.LocalDateDeserializer;
|
||||
import ru.spcex.clearing.platform.messaging.domain.json.deserialize.LocalTimeDeserializer;
|
||||
import ru.spcex.clearing.platform.messaging.domain.json.serialize.LocalDateSerializer;
|
||||
import ru.spcex.clearing.platform.messaging.domain.json.serialize.LocalTimeSerializer;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalTime;
|
||||
|
||||
public class JournalSdf {
|
||||
@JsonProperty
|
||||
@JsonSerialize(using = LocalDateSerializer.class)
|
||||
@JsonDeserialize(using = LocalDateDeserializer.class)
|
||||
private LocalDate registrationDate;
|
||||
@JsonProperty
|
||||
@JsonSerialize(using = LocalTimeSerializer.class)
|
||||
@JsonDeserialize(using = LocalTimeDeserializer.class)
|
||||
private LocalTime registrationTime;
|
||||
@JsonProperty
|
||||
private Long registrationNumber;
|
||||
@JsonProperty
|
||||
private String documentName;
|
||||
@JsonProperty
|
||||
private String dossierNumber;
|
||||
@JsonProperty
|
||||
private String resultStatus;
|
||||
|
||||
public LocalDate getRegistrationDate() {
|
||||
return registrationDate;
|
||||
}
|
||||
|
||||
public void setRegistrationDate(LocalDate registrationDate) {
|
||||
this.registrationDate = registrationDate;
|
||||
}
|
||||
|
||||
public LocalTime getRegistrationTime() {
|
||||
return registrationTime;
|
||||
}
|
||||
|
||||
public void setRegistrationTime(LocalTime registrationTime) {
|
||||
this.registrationTime = registrationTime;
|
||||
}
|
||||
|
||||
public Long getRegistrationNumber() {
|
||||
return registrationNumber;
|
||||
}
|
||||
|
||||
public void setRegistrationNumber(Long registrationNumber) {
|
||||
this.registrationNumber = registrationNumber;
|
||||
}
|
||||
|
||||
public String getDocumentName() {
|
||||
return documentName;
|
||||
}
|
||||
|
||||
public void setDocumentName(String documentName) {
|
||||
this.documentName = documentName;
|
||||
}
|
||||
|
||||
public String getDossierNumber() {
|
||||
return dossierNumber;
|
||||
}
|
||||
|
||||
public void setDossierNumber(String dossierNumber) {
|
||||
this.dossierNumber = dossierNumber;
|
||||
}
|
||||
|
||||
public String getResultStatus() {
|
||||
return resultStatus;
|
||||
}
|
||||
|
||||
public void setResultStatus(String resultStatus) {
|
||||
this.resultStatus = resultStatus;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
package ru.spcex.clearing.platform.messaging.domain.json.deserialize;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonParser;
|
||||
import com.fasterxml.jackson.databind.DeserializationContext;
|
||||
import com.fasterxml.jackson.databind.JsonDeserializer;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.time.LocalTime;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
|
||||
public class LocalTimeDeserializer extends JsonDeserializer<LocalTime> {
|
||||
private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss");
|
||||
|
||||
@Override
|
||||
public LocalTime deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
|
||||
String time = p.getText();
|
||||
if (time == null || time.trim().length() == 0) {
|
||||
return null;
|
||||
}
|
||||
return LocalTime.parse(time, formatter);
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue