diff --git a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/AbstractExporterService.java b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/AbstractExporterService.java index 61e0b1526..9c399b3a1 100644 --- a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/AbstractExporterService.java +++ b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/AbstractExporterService.java @@ -8,57 +8,70 @@ import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.domain.cud.utilities.LimExportedRequest; import ru.spcex.clearing.platform.messaging.serialization.LogFormatter; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.platform.classes.base.SpcexObjectBase; +import ru.spcex.platform.enumeration.SwtTable; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; +import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.nio.file.Files; +import java.io.OutputStream; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.Collection; import java.util.List; +import java.util.Map; +import java.util.Objects; import static ru.spcex.clearing.platform.messaging.domain.Consts.LIM_EXPORTED; -public abstract class AbstractExporterService { +public abstract class AbstractExporterService { private final Logger log = LoggerFactory.getLogger(getClass()); - protected final Imdg registryImdg; - private final List validStatus = List.of("ACTV", "ROPN"); - private final Imdg tradingClearingRegistryImdg; + protected final SwtTable type; + protected final Imdg sdfImdg; private final DateTimeFormatter dtFormatter = DateTimeFormatter.ofPattern("yyyyMMddHHmmss"); private final KafkaSender kafkaSender; protected final FileStorage fileStorage; protected AbstractExporterService(FileStorage fileStorage, - KafkaSender kafkaSender, ImdgProvider imdgProvider) { + KafkaSender kafkaSender, ImdgProvider imdgProvider, + SwtTable type, + String mapName, Class mapClass) { this.fileStorage = fileStorage; this.kafkaSender = kafkaSender; - this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class); - this.tradingClearingRegistryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class); + this.type = type; + Objects.requireNonNull(type, "SWT table type not set"); + this.sdfImdg = imdgProvider.getImdg(mapName, mapClass); } - public abstract Collection getLimFileRows(); - - public abstract String getTargetFileName(); + public SwtTable getType() { + return type; + } public void process() { String fileName = getTargetFileName(); log.debug("Start export {} Lim file", fileName); + byte[] data; + { + ByteArrayOutputStream outBuffer=new ByteArrayOutputStream(); + Collection records = selectItems();log.debug("Prepared {} record from {} to file {}", + records.size(), sdfImdg.getMapName(), fileName); + makeSWTData(records, outBuffer); + data= outBuffer.toByteArray(); + } try { - // todo возможная оптимизация: посмотреть размеры файлов, возможно обойтись без временного файла - //Files.write(limFilePath, getLimFileRows()); - fileStorage.saveFile(fileName, null); + fileStorage.saveFile(fileName, data); } catch (IOException e) { log.error("Failed export {} file", fileName); throw new RuntimeException(e); } log.debug("Successfully exported {} file", fileName); - sendSwtxportedNotification(fileName); + sendSwtExportedNotification(fileName); } - void sendSwtxportedNotification(String fileName) { + void sendSwtExportedNotification(String fileName) { LimExportedRequest limExportedRequest = new LimExportedRequest(); limExportedRequest.setLimFileName(fileName); log.debug("Send message to kafka \"{}\": {}", LIM_EXPORTED, LogFormatter.toStringWrapper(limExportedRequest)); @@ -75,4 +88,21 @@ public abstract class AbstractExporterService { return result; } + // Выборка + protected Collection selectItems() { + //todo select criteria? + return sdfImdg.getAllValues(); + } + + // Конвертация (см. meta.xml) + protected abstract Map convertRecord(T record); + protected abstract String[] swtHeader(); + + protected void makeSWTData(Collection records, OutputStream out) { + //todo ... header + validation + check + convert types + } + protected String convertItem(Object o) { + //todo date/time/etc. + return String.valueOf(o); + } } diff --git a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/LauncherCommandReceiver.java b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/LauncherCommandReceiver.java index a3ae54ca6..6917dbb29 100644 --- a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/LauncherCommandReceiver.java +++ b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/LauncherCommandReceiver.java @@ -5,34 +5,68 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.stereotype.Service; +import ru.spcex.clearing.platform.messaging.domain.BaseRequest; +import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.importexport.SwtExporterRequest; import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; import ru.spcex.clearing.platform.messaging.service.QueueConsumer; import ru.spcex.clearing.swt.exporter.services.exportimpl.MoneyExporterService; import ru.spcex.platform.enumeration.Task; +import ru.spcex.platform.utils.log.ExceptionUtils; + +import java.util.List; @Service public class LauncherCommandReceiver extends QueueConsumer implements InitializingBean { private final Logger log = LoggerFactory.getLogger(getClass()); - private final MoneyExporterService moneyExporterService; -// private final SecurityExporterService securityExporterService; + protected final List exporterServices; public LauncherCommandReceiver(Consumer kafkaQueue, - MoneyExporterService moneyExporterService - // , SecurityExporterService securityExporterService + List exporterServices ) { super(kafkaQueue); - this.moneyExporterService = moneyExporterService; -// this.securityExporterService = securityExporterService; + this.exporterServices = exporterServices; } @Override public void afterPropertiesSet() { - callback(LauncherCommandRequest.class) - .setConsumer(action -> moneyExporterService.process()) - .forDestination(Task.unloadingSession_LIMM.topic(), callbacks::put); // LIMM // callback(LauncherCommandRequest.class) -// .setConsumer(action -> securityExporterService.process()) -// .forDestination(Task.unloadingSession_LIMS.topic(), callbacks::put); // LIMS +// .setConsumer(this::exportAll) +// .forDestination(Task.unloadingSession_LIMM.topic(), callbacks::put); // todo task name? + callback(SwtExporterRequest.class) + .setConsumer(this::exportSpecial) + .forDestination(Consts.SWT_EXPORTER, callbacks::put); init(); } + + protected void exportAll(BaseRequest request) { + log.info("LauncherCommandRequest request received: {}", request); + for (AbstractExporterService exporter : exporterServices) { + log.debug("Export {}", exporter); + try { + exporter.process(); + } catch (Exception e) { + log.error("One of exporter has error: {}", ExceptionUtils.getStackTrace(e)); + } + } + log.info("All SWT export has finished."); + } + + protected void exportSpecial(BaseRequest request) { + log.info("SwtExporterRequest request received: {}", request); + SwtExporterRequest req = request.getRequestPayload(); + for (AbstractExporterService exporter : exporterServices) { + if (exporter.getType()==req.getType()) { + log.debug("Export {}", exporter); + try { + exporter.process(); + } catch (Exception e) { + log.error("Exporter has error: {}", ExceptionUtils.getStackTrace(e)); + } + } + return; + } + log.error("SWT export not execute for type \"{}\" - unknown command", req.getType()); + //todo return error? + } } diff --git a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/exportimpl/MoneyExporterService.java b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/exportimpl/MoneyExporterService.java index 7576ccad1..d80297d9d 100644 --- a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/exportimpl/MoneyExporterService.java +++ b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/services/exportimpl/MoneyExporterService.java @@ -4,9 +4,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; import ru.clearing.classes.statics.data.registry.Registry; +import ru.clearing.classes.statics.data.sdf.SDf12; +import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.clearing.swt.exporter.services.AbstractExporterService; import ru.spcex.clearing.swt.exporter.services.FileStorage; +import ru.spcex.platform.enumeration.SwtTable; import ru.spcex.platform.imdg.api.ImdgProvider; import java.math.BigDecimal; @@ -24,80 +27,10 @@ public class MoneyExporterService extends AbstractExporterService { public MoneyExporterService(FileStorage fileStorage, KafkaSender kafkaSender, ImdgProvider imdgProvider) { - super(fileStorage, kafkaSender, imdgProvider); + super(fileStorage, kafkaSender, imdgProvider, + SwtTable.SDF_12, + IMDGDistributedNames.Map_SDf12, SDf12.class); } - @Override - public String getTargetFileName() { - return prepareFileName(null, "money"); - } - - @Override - public Collection getLimFileRows() { - log.debug("Started loading and formation of money file lines"); - LocalDate currentDate = LocalDate.now(); - List swtFileRows = new ArrayList<>(); - Collection registriesA = registryImdg.getCollectionObjectsByFieldValues(Map.of( - "registryDesignation", "A", - "registryInstrumentType", "M", - "registryUnit", "F" - )); - Collection registriesD = registryImdg.getCollectionObjectsByFieldValues(Map.of( - "registryDesignation", "D", - "registryInstrumentType", "M", - "registryUnit", "T" - )); - Map> byTcrA = registriesA.stream() - .collect(Collectors.groupingBy(Registry::getTradingClearingRegistry)); - Map> byTcrD = registriesD.stream() - .collect(Collectors.groupingBy(Registry::getTradingClearingRegistry)); - - for (Map.Entry> entryA : byTcrA.entrySet()) { - List registriesListA = entryA.getValue(); - List registriesListB = byTcrD.get(entryA.getKey()); - for (Registry registryA : registriesListA) { -// todo if (checkNotBlocked(registryA)) { -// Registry registryD = findRegistryBySecurityId(registryA.getSecurityId(), registriesListB); -// limFileRows.add(getRow(registryA, registryD)); -// } - } - } - log.debug("Successfully completed the formation of rows: {} for export money", swtFileRows.size()); - return swtFileRows; - } - - public String getRow(Registry registryA, Registry registryD) { - StringBuilder row = new StringBuilder(); - - row.append("MONEY: FIRM_ID = "); - row.append(registryA.getTradingCode()); - - row.append("; TAG = SPVB"); - - row.append("; CURR_CODE = "); - row.append(registryA.getSecuritySymbol()); - - row.append("; CLIENT_CODE = "); - row.append(registryA.getTradingClearingRegistry()); - - row.append("; OPEN_BALANCE = "); - BigDecimal balance = registryA.getBalance() != null ? - registryD != null && registryD.getBalance() != null ? registryA.getBalance().subtract(registryD.getBalance()) : registryA.getBalance() : - BigDecimal.ZERO; - row.append(balance); - - row.append("; OPEN_LIMIT = 0.00"); - - row.append("; LIMIT_KIND = 0;"); - return row.toString(); - } - - private Registry findRegistryBySecurityId(Long securityId, List registries) { - for (Registry registry : registries) { - if (securityId.equals(registry.getSecurityId())) { - return registry; - } - } - return null; - } + //todo impl... } diff --git a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SwtTable.java b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SwtTable.java new file mode 100644 index 000000000..bf8a30217 --- /dev/null +++ b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SwtTable.java @@ -0,0 +1,23 @@ +package ru.spcex.platform.enumeration; + +import ru.spcex.platform.utils.enumeration.IEnumKey; + +public enum SwtTable implements IEnumKey { + SDF_12("SDF_12"), SDF_14("SDF_14"); + + SwtTable(String key) { + this.key = key; + } + + private final String key; + + @Override + public String getKey() { + return this.key; + } + + @Override + public boolean equalsByKey(String key) { + return IEnumKey.super.equalsByKey(key); + } +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java index e0197d92c..5a7c036e4 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java @@ -122,6 +122,7 @@ public interface Consts { String SDF54_PROCESS = "sdf54-process"; String SDF56_PROCESS = "sdf56-process"; String SDF57_PROCESS = "sdf57-process"; + String SWT_EXPORTER = "swt-exporter"; String REVISE_PROCESS = "revise-process"; String EXPORT_PROCESS = "export-process"; String EXPORT_COMPLETED = "export_completed"; diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/importexport/SwtExporterRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/importexport/SwtExporterRequest.java new file mode 100644 index 000000000..41b54f2e5 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/importexport/SwtExporterRequest.java @@ -0,0 +1,17 @@ +package ru.spcex.clearing.platform.messaging.domain.cud.importexport; + +import com.fasterxml.jackson.annotation.JsonProperty; +import ru.spcex.platform.enumeration.SwtTable; + +public class SwtExporterRequest { + @JsonProperty + public SwtTable type; + + public SwtTable getType() { + return type; + } + + public void setType(SwtTable type) { + this.type = type; + } +}