swt-exporter http://jira.mfd.msk:8088/browse/CLS-317 начало...

2023-05-25 12:10:12 +03:00 · 2023-05-25 12:10:12 +03:00 · cf2b937cff
commit cf2b937cff
parent d3a9c46234
6 changed files with 139 additions and 101 deletions
--- 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<T extends SpcexObjectBase> {
    private final Logger log = LoggerFactory.getLogger(getClass());
-    protected final Imdg<Registry> registryImdg;
-    private final List<String> validStatus = List.of("ACTV", "ROPN");
-    private final Imdg<TradingClearingRegistry> tradingClearingRegistryImdg;
+    protected final SwtTable type;
+    protected final Imdg<T> 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<T> 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<String> 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<T> 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<T> selectItems() {
+        //todo select criteria?
+        return sdfImdg.getAllValues();
+    }
+
+    // Конвертация (см. meta.xml)
+    protected abstract Map<String, Object> convertRecord(T record);
+    protected abstract String[] swtHeader();
+
+    protected void makeSWTData(Collection<T> records, OutputStream out) {
+        //todo ... header + validation + check + convert types
+    }
+    protected String convertItem(Object o) {
+        //todo date/time/etc.
+        return String.valueOf(o);
+    }
 }
--- 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<AbstractExporterService> exporterServices;

    public LauncherCommandReceiver(Consumer<String, Object> kafkaQueue,
-                                   MoneyExporterService moneyExporterService
-          // ,    SecurityExporterService securityExporterService
+                                   List<AbstractExporterService> 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<LauncherCommandRequest> 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<SwtExporterRequest> 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?
+    }
 }
--- 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<String> getLimFileRows() {
-        log.debug("Started loading and formation of money file lines");
-        LocalDate currentDate = LocalDate.now();
-        List<String> swtFileRows = new ArrayList<>();
-        Collection<Registry> registriesA = registryImdg.getCollectionObjectsByFieldValues(Map.of(
-                "registryDesignation", "A",
-                "registryInstrumentType", "M",
-                "registryUnit", "F"
-        ));
-        Collection<Registry> registriesD = registryImdg.getCollectionObjectsByFieldValues(Map.of(
-                "registryDesignation", "D",
-                "registryInstrumentType", "M",
-                "registryUnit", "T"
-        ));
-        Map<String, List<Registry>> byTcrA = registriesA.stream()
-                .collect(Collectors.groupingBy(Registry::getTradingClearingRegistry));
-        Map<String, List<Registry>> byTcrD = registriesD.stream()
-                .collect(Collectors.groupingBy(Registry::getTradingClearingRegistry));
-
-        for (Map.Entry<String, List<Registry>> entryA : byTcrA.entrySet()) {
-            List<Registry> registriesListA = entryA.getValue();
-            List<Registry> 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<Registry> registries) {
-        for (Registry registry : registries) {
-            if (securityId.equals(registry.getSecurityId())) {
-                return registry;
-            }
-        }
-        return null;
-    }
+    //todo impl...
 }
--- 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
@ -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);
+    }
+}
--- 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";
--- 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
@ -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;
+    }
+}