diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/KafkaConfig.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/KafkaConfig.java index 73840d3e5..e2dabd45a 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/KafkaConfig.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/KafkaConfig.java @@ -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 createConsumer(ExportDBFServiceSettings settings) { return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); } + + @Autowired + @Bean + public Producer createProducer(ExportDBFServiceSettings settings) { + if (settings.getKafkaProducer() != null) { + return KafkaProducerFactory.producer(settings.getKafkaProducer()); + } else + return null; + } + + @Autowired(required = false) + @Bean + public KafkaSender kafkaSender(Producer 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 imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); + return imdg::insert; + }) + .build(); + } } diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/ExportDBFServiceSettings.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/ExportDBFServiceSettings.java index 8166f7931..9f3b145f0 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/ExportDBFServiceSettings.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/ExportDBFServiceSettings.java @@ -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; + } } diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/Processor.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/Processor.java index bf2341148..bbbdb6d8b 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/Processor.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/Processor.java @@ -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", diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/data/ResultContainer.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/data/ResultContainer.java index 0ad0d79fd..a84f3a8cb 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/data/ResultContainer.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/data/ResultContainer.java @@ -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; + } } diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java index 43e393138..c67363f54 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java @@ -111,7 +111,7 @@ public class ExportFromHazelcast extends Stage implements InitializingBean { } } - return StageResult.COMPLETE; + return StageResult.OK; } /** diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/Journal.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/Journal.java new file mode 100644 index 000000000..1416359e9 --- /dev/null +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/Journal.java @@ -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 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 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"); + } +} diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/Stage.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/Stage.java index cb9870d0d..c42b6b4b9 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/Stage.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/Stage.java @@ -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; + } } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/system/JournalSdf.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/system/JournalSdf.java new file mode 100644 index 000000000..2c2e9de22 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/system/JournalSdf.java @@ -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; + } +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/deserialize/LocalTimeDeserializer.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/deserialize/LocalTimeDeserializer.java new file mode 100644 index 000000000..0fa9aa642 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/deserialize/LocalTimeDeserializer.java @@ -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 { + 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); + } +}