From 8ea3397e16495ae5af243911c230c9f09e87f33e Mon Sep 17 00:00:00 2001 From: AKurakin Date: Sat, 27 May 2023 12:59:58 +0300 Subject: [PATCH 1/5] =?UTF-8?q?lim-exporter=20swt-exporter=20http://jira.m?= =?UTF-8?q?fd.msk:8088/browse/CLS-262=20=D0=BF=D0=BE=D0=BF=D1=80=D0=B0?= =?UTF-8?q?=D0=B2=D0=B8=D0=BB=20=D0=B8=D0=BD=D0=B8=D1=86=D0=B8=D0=B0=D0=BB?= =?UTF-8?q?=D0=B8=D0=B7=D0=B0=D1=86=D0=B8=D1=8E=20kafka?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../clearing/lim/exporter/config/KafkaConfig.java | 7 +++++++ .../src/main/resources/application.properties | 4 ---- .../exporter/services/AbstractExporterServiceTest.java | 10 ++++++++++ .../clearing/swt/exporter/config/KafkaConfig.java | 7 +++++++ platform-parent/platform-imdg-api/pom.xml | 5 +++++ 5 files changed, 29 insertions(+), 4 deletions(-) diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/KafkaConfig.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/KafkaConfig.java index 6721788f6..79c009425 100644 --- a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/KafkaConfig.java +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/KafkaConfig.java @@ -13,6 +13,7 @@ import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.lim.exporter.config.settings.ExportLimServiceSettings; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; +import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; import ru.spcex.clearing.platform.messaging.service.RequestInfo; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.platform.imdg.api.Imdg; @@ -35,6 +36,12 @@ public class KafkaConfig { } @Bean + public ProducerFactory pf(ExportLimServiceSettings settings) { + KafkaProducerSettings kafkaSettings = settings.getKafkaProducer(); + return KafkaProducerFactory.producerFactory(kafkaSettings); // Использует специальный сериализатор в JSON + } + + @Bean("kafkaTemplate") public KafkaTemplate kafkaTemplate(ProducerFactory pf) { return new KafkaTemplate<>(pf); } diff --git a/clearing-parent/lim-exporter/src/main/resources/application.properties b/clearing-parent/lim-exporter/src/main/resources/application.properties index 114f0954c..2c595a4ac 100644 --- a/clearing-parent/lim-exporter/src/main/resources/application.properties +++ b/clearing-parent/lim-exporter/src/main/resources/application.properties @@ -13,10 +13,6 @@ export-lim-service.store.password=password export-lim-service.store.server-ip=10.230.238.53 export-lim-service.store.server-port=2222 -export-lim-service.hazelcast.cluster-members=127.0.0.1:5701 -export-lim-service.hazelcast.login=dev -export-lim-service.hazelcast.password=dev-pass - export-lim-service.kafka-consumer.bootstrap-servers=localhost:9092 export-lim-service.kafka-consumer.group-id=dev-group-balance-service export-lim-service.kafka-consumer.enable-auto-commit=false diff --git a/clearing-parent/lim-exporter/src/test/java/ru/spcex/clearing/lim/exporter/services/AbstractExporterServiceTest.java b/clearing-parent/lim-exporter/src/test/java/ru/spcex/clearing/lim/exporter/services/AbstractExporterServiceTest.java index d1a6bfa32..bc857354d 100644 --- a/clearing-parent/lim-exporter/src/test/java/ru/spcex/clearing/lim/exporter/services/AbstractExporterServiceTest.java +++ b/clearing-parent/lim-exporter/src/test/java/ru/spcex/clearing/lim/exporter/services/AbstractExporterServiceTest.java @@ -2,6 +2,10 @@ package ru.spcex.clearing.lim.exporter.services; import org.junit.jupiter.api.Test; import ru.spcex.clearing.lim.exporter.AbstractServiceTest; +import ru.spcex.clearing.lim.exporter.config.KafkaConfig; +import ru.spcex.clearing.lim.exporter.config.settings.ExportLimServiceSettings; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSenderBuilderImpl; import ru.spcex.clearing.test.TestUtils; import javax.annotation.PostConstruct; @@ -23,5 +27,11 @@ class AbstractExporterServiceTest extends AbstractServiceTest { String fileName = "limits_money_202305241832.lim"; moneyExporterService.sendLimExportedNotification(fileName); //TestUtils.waitingSendAndCheckRecord(null, mockProducer); + + /* + Для интеграционного теста см. + KafkaProducerSettings kafkaSettings = settings.getKafkaProducer(); + return KafkaProducerFactory.producerFactory(kafkaSettings); // Использует специальный сериализатор в JSON + */ } } \ No newline at end of file diff --git a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/config/KafkaConfig.java b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/config/KafkaConfig.java index 3ad111793..69df8cc0a 100644 --- a/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/config/KafkaConfig.java +++ b/clearing-parent/swt-exporter/src/main/java/ru/spcex/clearing/swt/exporter/config/KafkaConfig.java @@ -10,6 +10,7 @@ import org.springframework.context.annotation.Scope; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; import ru.spcex.clearing.swt.exporter.config.settings.ExportSwtServiceSettings; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; @@ -35,6 +36,12 @@ public class KafkaConfig { } @Bean + public ProducerFactory pf(ExportSwtServiceSettings settings) { + KafkaProducerSettings kafkaSettings = settings.getKafkaProducer(); + return KafkaProducerFactory.producerFactory(kafkaSettings); // Использует специальный сериализатор в JSON + } + + @Bean("kafkaTemplate") public KafkaTemplate kafkaTemplate(ProducerFactory pf) { return new KafkaTemplate<>(pf); } diff --git a/platform-parent/platform-imdg-api/pom.xml b/platform-parent/platform-imdg-api/pom.xml index 69d60ff13..3f2517bfa 100644 --- a/platform-parent/platform-imdg-api/pom.xml +++ b/platform-parent/platform-imdg-api/pom.xml @@ -39,6 +39,11 @@ junit-jupiter test + + org.junit.jupiter + junit-jupiter-api + test + org.assertj assertj-core From 56441c39cb3bfcc0ea69e7575d914d8a90fcbcfa Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Sat, 27 May 2023 13:11:06 +0300 Subject: [PATCH 2/5] fix SDF57Executor and registryBuilder --- .../builder/PaymentInstructionBuilder.java | 3 +- .../service/executors/Sdf57Executor.java | 10 ++-- .../session/stage/impl/RegistryBuilder.java | 53 +++++++++++++++---- 3 files changed, 53 insertions(+), 13 deletions(-) diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/PaymentInstructionBuilder.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/PaymentInstructionBuilder.java index ff1f9ecde..80479f8c9 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/PaymentInstructionBuilder.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/builder/PaymentInstructionBuilder.java @@ -7,6 +7,7 @@ import ru.clearing.classes.statics.data.payment.PaymentInstruction; import ru.clearing.classes.statics.data.registry.Registry; import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.platform.enumeration.CompanySymbol; +import ru.spcex.platform.enumeration.InOutDirection; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; @@ -66,7 +67,7 @@ public class PaymentInstructionBuilder { } paymentInstruction.setDebitLeg_accountId(registry.getAccountId()); - + paymentInstruction.setDebitLeg_direction(InOutDirection.in.getKey()); // Account account = accountImdg.getSingleObjectByID(registry.get()); // paymentInstruction.setDebitLeg_account(); return paymentInstruction; diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf57Executor.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf57Executor.java index 3efdc6c1e..9749bc9b6 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf57Executor.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf57Executor.java @@ -171,11 +171,12 @@ public class Sdf57Executor extends AbstractExecutor { Registry registry = createRegistryByStatement(stmt, company, account, dsgn); Registry registryB = registry.clone(); + registryB.setId(imdgProvider.getImdgIdGenerator().nextId()); registryB.setRegistryUnit(RegistryUnit.B.getKey()); registryB.setRegistryCode(RegistryUtil.clearingCode(registryB)); - registryB.setId(imdgProvider.getImdgIdGenerator().nextId()); registryB.setBalance(BigDecimal.ZERO); registryB.setDebit(BigDecimal.ZERO); + registryB.setCredit(BigDecimal.ZERO); Registry registryF = registry.clone(); registryF.setRegistryUnit(RegistryUnit.F.getKey()); @@ -291,10 +292,13 @@ public class Sdf57Executor extends AbstractExecutor { //Если значение statement.inOutDirection=OUT, то: //= текущее значение registry.balance - statement.amount switch (inOutDirection) { - case in -> rgs.setBalance(statement.getAmount()); + case in -> { + rgs.setBalance(statement.getAmount()); + rgs.setCredit(statement.getAmount()); + } case out -> { rgs.setBalance(statement.getAmount().negate()); - rgs.setDebit(statement.getAmount()); + rgs.setDebit(statement.getAmount().negate()); } } rgs.setBalanceDimension(BalanceDimension.MONY.getKey()); //fixme ! смотри описание и ссылка на начало html'ки diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/RegistryBuilder.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/RegistryBuilder.java index 18f16b394..b19291e9f 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/RegistryBuilder.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/RegistryBuilder.java @@ -2,10 +2,13 @@ package ru.spcex.clearing.session.stage.impl; import ru.clearing.classes.statics.data.account.Account; import ru.clearing.classes.statics.data.account.ClearingAccount; +import ru.clearing.classes.statics.data.account.DepoAccount; +import ru.clearing.classes.statics.data.account.InformationAccount; import ru.clearing.classes.statics.data.company.Company; import ru.clearing.classes.statics.data.execution.ExecutionCommon; import ru.clearing.classes.statics.data.execution.ExecutionDeposit; import ru.clearing.classes.statics.data.execution.ExecutionFond; +import ru.clearing.classes.statics.data.misc.Currency; import ru.clearing.classes.statics.data.misc.Market; import ru.clearing.classes.statics.data.misc.Session; import ru.clearing.classes.statics.data.registry.Registry; @@ -20,6 +23,7 @@ import ru.spcex.platform.imdg.api.ImdgProvider; import java.time.Instant; import java.time.LocalDate; import java.time.format.DateTimeFormatter; +import java.util.Map; public class RegistryBuilder { private Imdg companyImdg; @@ -28,6 +32,9 @@ public class RegistryBuilder { private Imdg marketImdg; private Imdg sessionImdg; private Imdg clearingAccountImdg; + private Imdg informationAccountImdg; + private Imdg depoAccountImdg; + private Imdg currencyImdg; private ExecutionCommon exec; private RegistryDesignation regDsgn; @@ -45,6 +52,9 @@ public class RegistryBuilder { this.marketImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Market, Market.class); this.sessionImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Session, Session.class); this.clearingAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_ClearingAccount, ClearingAccount.class); + this.depoAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_DepoAccount, DepoAccount.class); + this.informationAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_InformationAccount, InformationAccount.class); + this.currencyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Currency, Currency.class); return this; } @@ -69,32 +79,51 @@ public class RegistryBuilder { reg.setShortName(company.getShortName()); reg.setFullName(company.getFullName()); TradingClearingRegistry tcr = searchTradingClearingRegistry(); - ClearingAccount accountForStatement = null; + String capacityByAccount = null; + boolean isMoney = false; + Account account = accountImdg.getSingleObjectByID(reg.getAccountId()); if ((regDsgn.equals(RegistryDesignation.O) && side.isBuy()) || (regDsgn.equals(RegistryDesignation.T) && side.isSell())) { + isMoney = true; reg.setAccountId(tcr.getMoneyAccountId()); reg.setRegistryInstrumentType(RegistryInstrumentType.M.getKey()); reg.setBalanceDimension(BalanceDimension.MONY.getKey()); //fixme add second leg code branch - accountForStatement = clearingAccountImdg.getSingleObjectByID(tcr.getMoneyAccountId()); + Currency currency = currencyImdg.getSingleObjectByFieldValues(Map.of("currencyCode", exec.getSettlementCurrency())); + reg.setSecurityId(currency.getId()); + reg.setSecuritySymbol(currency.getCurrencyCode()); + if (AccountType.Clrn.equalsByKey(account.getAccountType())) { + ClearingAccount clearingAccount = clearingAccountImdg.getSingleObjectByID(tcr.getMoneyAccountId()); + if (clearingAccount != null) { + capacityByAccount = clearingAccount.getClearingAccountType(); + } + } else if (AccountType.Info.equalsByKey(account.getAccountType())){ + InformationAccount informationAccount = informationAccountImdg.getSingleObjectByID(tcr.getMoneyAccountId()); + if (informationAccount != null) { + capacityByAccount = "A"; //todo сделать enum + } + } } else if ((regDsgn.equals(RegistryDesignation.T) && side.isBuy()) || (regDsgn.equals(RegistryDesignation.O) && side.isSell())) { + isMoney = false; reg.setAccountId(tcr.getDepoAccountId()); reg.setRegistryInstrumentType(RegistryInstrumentType.S.getKey()); reg.setBalanceDimension(BalanceDimension.PICS.getKey()); //fixme add second leg code branch - accountForStatement = clearingAccountImdg.getSingleObjectByID(tcr.getDepoAccountId()); + reg.setSecurityId(exec.getSecurityId()); + reg.setSecuritySymbol(exec.getSecuritySymbol()); + DepoAccount depoAccount = depoAccountImdg.getSingleObjectByID(tcr.getDepoAccountId()); + if (depoAccount != null) { + capacityByAccount = depoAccount.getDepoAccountType(); + } } - Account account = accountImdg.getSingleObjectByID(reg.getAccountId()); reg.setAccountType(account.getAccountType()); reg.setAccount(account.getAccount()); reg.setRegistryDesignation(regDsgn.getKey()); - if (accountForStatement != null) { - reg.setRegistryCapacity(accountForStatement.getClearingAccountType()); + if (capacityByAccount != null) { + reg.setRegistryCapacity(capacityByAccount); } reg.setRegistryUnit(RegistryUnit.T.getKey()); reg.setRegistryCode(RegistryUtil.clearingCode(reg)); reg.setTradingClearingRegistryId(exec.getTradingClearingRegistryId()); reg.setTradingClearingRegistry(tcr.getCode()); reg.setRegistryStatus(RegistryStatus.PROC.getKey()); - reg.setSecurityId(exec.getSecurityId()); - reg.setSecuritySymbol(exec.getSecuritySymbol()); switch (exec.type()) { case ExecutionDeposit -> { ExecutionDeposit execDep = (ExecutionDeposit) this.exec; @@ -112,7 +141,13 @@ public class RegistryBuilder { } case ExecutionFond -> { ExecutionFond execFond = (ExecutionFond) this.exec; - reg.setBalance(execFond.getSettlementAmount()); + if (isMoney) { + //todo ОКРУГЛЕНИЕ(fixedIncomeSecurity.nominalValue * quantity * price/100) + ОКРУГЛЕНИЕ (executionFond.interestAmount) + reg.setBalance(execFond.getSettlementAmount()); + } else { + reg.setBalance(execFond.getQuantity()); + } + reg.setSettlementDate(execFond.getSettlementDate()); reg.setSettlementCode(execFond.getSettlementCode()); } From 9fa165de2ec5317e1b72010fe042d8f463e700bb Mon Sep 17 00:00:00 2001 From: aalehin Date: Sat, 27 May 2023 13:37:29 +0300 Subject: [PATCH 3/5] http://jira.mfd.msk:8088/browse/CLS-316 --- mid swt importer --- clearing-parent/pom.xml | 1 + clearing-parent/swt-importer/pom.xml | 103 +++++++++++++++ .../swt/importer/SWTImporterApplication.java | 18 +++ .../importer/config/ImporterImdgConfig.java | 46 +++++++ .../swt/importer/config/KafkaConfig.java | 60 +++++++++ .../importer/config/SWTImporterConfig.java | 67 ++++++++++ .../swt/importer/config/settings/Common.java | 32 +++++ .../swt/importer/config/settings/Cron.java | 14 +++ .../settings/ImportSWTServiceSettings.java | 68 ++++++++++ .../swt/importer/config/settings/Store.java | 41 ++++++ .../controller/ImporterController.java | 37 ++++++ .../controller/LauncherCommandReceiver.java | 43 +++++++ .../importer/exceptions/ConfigException.java | 6 + .../swt/importer/logic/Processor.java | 44 +++++++ .../logic/data/ColumnStructureSWT.java | 51 ++++++++ .../importer/logic/data/DBStructureSWT.java | 18 +++ .../importer/logic/data/ResultContainer.java | 70 +++++++++++ .../logic/data/TableStructureSWT.java | 25 ++++ .../importer/logic/data/enums/ColumnType.java | 46 +++++++ .../swt/importer/logic/data/enums/ETable.java | 33 +++++ .../logic/data/enums/StageResult.java | 7 ++ .../logic/data/tables/AbstractTable.java | 46 +++++++ .../logic/data/tables/SDf08Table.java | 33 +++++ .../logic/data/tables/SDf10Table.java | 33 +++++ .../logic/data/tables/SDf13Table.java | 29 +++++ .../logic/stages/ChangeDirOfFileStage.java | 79 ++++++++++++ .../swt/importer/logic/stages/ImportToDB.java | 73 +++++++++++ .../logic/stages/LoadFileFromDisk.java | 44 +++++++ .../logic/stages/SWTImportKafkaMessenger.java | 51 ++++++++ .../swt/importer/logic/stages/Stage.java | 16 +++ .../importer/logic/stages/ValidateFields.java | 44 +++++++ .../swt/importer/services/FileChecker.java | 56 +++++++++ .../importer/services/SWTImporterService.java | 119 ++++++++++++++++++ .../src/main/resources/application.properties | 34 +++++ .../src/main/resources/logback.xml | 37 ++++++ .../spcex/platform/enumeration/SdfTable.java | 10 +- 36 files changed, 1529 insertions(+), 5 deletions(-) create mode 100644 clearing-parent/swt-importer/pom.xml create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/SWTImporterApplication.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/ImporterImdgConfig.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/KafkaConfig.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SWTImporterConfig.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Common.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Cron.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/ImportSWTServiceSettings.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/ImporterController.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/LauncherCommandReceiver.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/exceptions/ConfigException.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/Processor.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ColumnStructureSWT.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/DBStructureSWT.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ResultContainer.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/TableStructureSWT.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ColumnType.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ETable.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/StageResult.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/AbstractTable.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf08Table.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf10Table.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf13Table.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ChangeDirOfFileStage.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ImportToDB.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/LoadFileFromDisk.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/SWTImportKafkaMessenger.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/Stage.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ValidateFields.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/SWTImporterService.java create mode 100644 clearing-parent/swt-importer/src/main/resources/application.properties create mode 100644 clearing-parent/swt-importer/src/main/resources/logback.xml diff --git a/clearing-parent/pom.xml b/clearing-parent/pom.xml index b00850e36..4e17f0658 100644 --- a/clearing-parent/pom.xml +++ b/clearing-parent/pom.xml @@ -39,6 +39,7 @@ trade-importer lim-exporter swt-exporter + swt-importer diff --git a/clearing-parent/swt-importer/pom.xml b/clearing-parent/swt-importer/pom.xml new file mode 100644 index 000000000..35515baf1 --- /dev/null +++ b/clearing-parent/swt-importer/pom.xml @@ -0,0 +1,103 @@ + + + + clearing-parent + ru.spcex.clearing + SPCEX-1.0.0.0 + + 4.0.0 + + swt-importer + + + 17 + 17 + + + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-autoconfigure + + + + + org.springframework + spring-jdbc + + + com.mchange + c3p0 + + + org.postgresql + postgresql + + + + + com.github.albfernandez + javadbf + + + ru.spcex.platform + platform-imdg-api + + + ru.spcex.platform + platform-imdg-api-hazelcast-impl + + + ru.spcex.clearing + classes + + + ru.spcex.platform + platform-messaging + + + ru.spcex.platform + platform-enum + + + + + + + src/main/resources + + application.properties + + false + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + repackage + + + + + ${project.artifactId} + + + + + + \ No newline at end of file diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/SWTImporterApplication.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/SWTImporterApplication.java new file mode 100644 index 000000000..534e987ee --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/SWTImporterApplication.java @@ -0,0 +1,18 @@ +package ru.spcex.clearing.swt.importer; + +import org.slf4j.LoggerFactory; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; + +@SpringBootApplication +public class SWTImporterApplication { + public static void main(String[] args) { + try { + SpringApplicationBuilder builder = new SpringApplicationBuilder(SWTImporterApplication.class); + builder.run(args); + } catch (Exception e) { + LoggerFactory.getLogger(SWTImporterApplication.class).error("SWT-Loader start failed: {} -> {}", e.getClass().getSimpleName(), e.getMessage()); + System.exit(-1); + } + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/ImporterImdgConfig.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/ImporterImdgConfig.java new file mode 100644 index 000000000..26dd3bc45 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/ImporterImdgConfig.java @@ -0,0 +1,46 @@ +package ru.spcex.clearing.swt.importer.config; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService; + +@Configuration +public class ImporterImdgConfig { + private static ThreadPoolTaskExecutor createThreadPoolTaskExecutor(int maxPoolSz, boolean waitForCompletion) { + ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor(); + if (maxPoolSz > 2) { + pool.setKeepAliveSeconds(60); + pool.setAllowCoreThreadTimeOut(true); + } + pool.setCorePoolSize(maxPoolSz); + pool.setWaitForTasksToCompleteOnShutdown(waitForCompletion); + return pool; + } + + @Bean(name = "taskExecutorHazelcastClientInitializer") + public ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer() { + return createThreadPoolTaskExecutor(1, true); + } + + @Bean(name = "taskExecutorIdGeneratorAwaiter") + public ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter() { + return createThreadPoolTaskExecutor(1, false); + } + + @Autowired + @Bean + public HazelcastService imdgProvider( + @Qualifier("taskExecutorHazelcastClientInitializer") ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer, + @Qualifier("taskExecutorIdGeneratorAwaiter") ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter, + ImportSWTServiceSettings settings + ) { + return new HazelcastService(taskExecutorHazelcastClientInitializer, + taskExecutorIdGeneratorAwaiter, + settings.getHazelcast()); + } + +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/KafkaConfig.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/KafkaConfig.java new file mode 100644 index 000000000..f3a9a6117 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/KafkaConfig.java @@ -0,0 +1,60 @@ +package ru.spcex.clearing.swt.importer.config; + +import org.apache.kafka.clients.consumer.Consumer; +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 org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +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.config.element.KafkaProducerSettings; +import ru.spcex.clearing.platform.messaging.service.RequestInfo; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgId; +import ru.spcex.platform.imdg.api.ImdgProvider; + +import java.util.function.Supplier; + +@Configuration +public class KafkaConfig { + + @Bean + public ProducerFactory pf(ImportSWTServiceSettings settings) { + KafkaProducerSettings kafkaSettings = settings.getKafkaProducer(); + return KafkaProducerFactory.producerFactory(kafkaSettings); + } + + @Bean("kafkaTemplate") + public KafkaTemplate kafkaTemplate(ProducerFactory pf) { + return new KafkaTemplate<>(pf); + } + + @Autowired + @Bean + public Supplier kafkaSender(KafkaTemplate kafkaTemplate, + ImdgProvider imdgProvider) { + ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator(); + return () -> KafkaSender + .setup() + .setKafkaTemplate(kafkaTemplate) + .idGenerator(imdgIdGenerator::nextId) + .imdgProvider(s -> { + Imdg imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); + return imdg::insert; + }) + .build(); + } + + @Autowired + @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) + @Bean + public Consumer createConsumer(ImportSWTServiceSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SWTImporterConfig.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SWTImporterConfig.java new file mode 100644 index 000000000..efffb5d6c --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SWTImporterConfig.java @@ -0,0 +1,67 @@ +package ru.spcex.clearing.swt.importer.config; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; +import ru.spcex.clearing.swt.importer.logic.data.tables.AbstractTable; +import ru.spcex.clearing.swt.importer.logic.data.tables.SDf08Table; +import ru.spcex.clearing.swt.importer.logic.data.tables.SDf10Table; +import ru.spcex.clearing.swt.importer.logic.data.tables.SDf13Table; +import ru.spcex.clearing.swt.importer.logic.stages.*; + +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; + +@Configuration +@EnableConfigurationProperties +@ComponentScan(basePackages = {"ru.spcex.clearing.swt.importer"}) +public class SWTImporterConfig { + private final ImportSWTServiceSettings settings; + private final ApplicationContext context; + + + public SWTImporterConfig(ImportSWTServiceSettings settings, ApplicationContext context) { + this.settings = settings; + this.context = context; + } + + @Bean("pipeline") + public List pipeline() { + List pipeline = new LinkedList<>(); + pipeline.add(context.getBean(LoadFileFromDisk.class)); + pipeline.add(context.getBean(ValidateFields.class)); + pipeline.add(context.getBean(ImportToDB.class)); + pipeline.add(context.getBean(ChangeDirOfFileStage.class)); + return pipeline; + } + + @Bean("executor") + public ThreadPoolTaskExecutor executor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setMaxPoolSize(settings.getCommon().getThreadsCount()); + executor.setCorePoolSize(settings.getCommon().getThreadsCount()); + executor.setThreadNamePrefix("swt-importer-thread-"); + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setAwaitTerminationSeconds(300); + executor.initialize(); + return executor; + } + + @Bean("mapOfTable") + public Map getMapOfTables() { + Map map = new HashMap<>(); + map.put(ETable.S_DF_08, new SDf08Table()); + map.put(ETable.S_DF_10, new SDf10Table()); + map.put(ETable.S_DF_13, new SDf13Table()); + return map; + } + + +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Common.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Common.java new file mode 100644 index 000000000..5069cd71c --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Common.java @@ -0,0 +1,32 @@ +package ru.spcex.clearing.swt.importer.config.settings; + +public class Common { + + private String encodingSource; + private int insertBatchSize; + private int threadsCount; + + public String getEncodingSource() { + return encodingSource; + } + + public void setEncodingSource(String encodingSource) { + this.encodingSource = encodingSource; + } + + public int getInsertBatchSize() { + return insertBatchSize; + } + + public void setInsertBatchSize(int insertBatchSize) { + this.insertBatchSize = insertBatchSize; + } + + public int getThreadsCount() { + return threadsCount; + } + + public void setThreadsCount(int threadsCount) { + this.threadsCount = threadsCount; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Cron.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Cron.java new file mode 100644 index 000000000..d747eab18 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Cron.java @@ -0,0 +1,14 @@ +package ru.spcex.clearing.swt.importer.config.settings; + +public class Cron { + + private String checkSrcDirCron; + + public String getCheckSrcDirCron() { + return checkSrcDirCron; + } + + public void setCheckSrcDirCron(String checkSrcDirCron) { + this.checkSrcDirCron = checkSrcDirCron; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/ImportSWTServiceSettings.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/ImportSWTServiceSettings.java new file mode 100644 index 000000000..5b00ee62d --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/ImportSWTServiceSettings.java @@ -0,0 +1,68 @@ +package ru.spcex.clearing.swt.importer.config.settings; + +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 +@PropertySource("file:${spring.config.location}/application.properties") +@ConfigurationProperties("import-swt-service") +public class ImportSWTServiceSettings { + private HazelcastClientParams hazelcast; + private KafkaProducerSettings kafkaProducer; + private KafkaConsumerSettings kafkaConsumer; + private Common common; + private Store store; + private Cron cron; + + public HazelcastClientParams getHazelcast() { + return hazelcast; + } + + public void setHazelcast(HazelcastClientParams hazelcast) { + this.hazelcast = hazelcast; + } + + public KafkaProducerSettings getKafkaProducer() { + return kafkaProducer; + } + + public void setKafkaProducer(KafkaProducerSettings kafkaProducer) { + this.kafkaProducer = kafkaProducer; + } + + public KafkaConsumerSettings getKafkaConsumer() { + return kafkaConsumer; + } + + public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) { + this.kafkaConsumer = kafkaConsumer; + } + + public Common getCommon() { + return common; + } + + public void setCommon(Common common) { + this.common = common; + } + + public Store getStore() { + return store; + } + + public void setStore(Store store) { + this.store = store; + } + + public Cron getCron() { + return cron; + } + + public void setCron(Cron cron) { + this.cron = cron; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java new file mode 100644 index 000000000..f74a91998 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java @@ -0,0 +1,41 @@ +package ru.spcex.clearing.swt.importer.config.settings; + +public class Store { + + private String srcDir; + private String outDir; + private String outDirError; + private boolean deleteSrcFiles = true; + + public String getSrcDir() { + return srcDir; + } + + public void setSrcDir(String srcDir) { + this.srcDir = srcDir; + } + + public String getOutDir() { + return outDir; + } + + public void setOutDir(String outDir) { + this.outDir = outDir; + } + + public boolean isDeleteSrcFiles() { + return deleteSrcFiles; + } + + public void setDeleteSrcFiles(boolean deleteSrcFiles) { + this.deleteSrcFiles = deleteSrcFiles; + } + + public String getOutDirError() { + return outDirError; + } + + public void setOutDirError(String outDirError) { + this.outDirError = outDirError; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/ImporterController.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/ImporterController.java new file mode 100644 index 000000000..ee2470f5f --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/ImporterController.java @@ -0,0 +1,37 @@ +package ru.spcex.clearing.swt.importer.controller; + + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.http.MediaType; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.ResponseBody; +import ru.spcex.clearing.swt.importer.services.SWTImporterService; + +@Controller("/") +public class ImporterController implements InitializingBean { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final SWTImporterService importerService; + + public ImporterController(@Qualifier("swtImporterService") SWTImporterService importerService) { + this.importerService = importerService; + } + + @RequestMapping(method = RequestMethod.GET, path = "/import", produces = MediaType.TEXT_PLAIN_VALUE) + @ResponseBody + public String checkFolder() { + log.info("Call check method for importer controller"); + importerService.run(); + return "dbf import complete, see log"; + } + + + @Override + public void afterPropertiesSet() throws Exception { + log.info("controller started"); + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/LauncherCommandReceiver.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/LauncherCommandReceiver.java new file mode 100644 index 000000000..1ca177ca6 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/LauncherCommandReceiver.java @@ -0,0 +1,43 @@ +package ru.spcex.clearing.swt.importer.controller; + +import org.apache.kafka.clients.consumer.Consumer; +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.Service; +import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; +import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; +import ru.spcex.clearing.swt.importer.services.SWTImporterService; +import ru.spcex.platform.enumeration.Task; + +@Service +public class LauncherCommandReceiver extends QueueConsumer implements InitializingBean { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final SWTImporterService importer; + + @Autowired + public LauncherCommandReceiver(Consumer kafkaQueue, + SWTImporterService importer) { + super(kafkaQueue); + this.importer = importer; + } + + /** + * В реальности не используется + */ + @Override + public void afterPropertiesSet() { + callback(LauncherCommandRequest.class) + .setConsumer(action -> importer.run(ETable.S_DF_08)) + .forDestination(Task.createOrderConfirm.topic(), callbacks::put); // CORC + callback(LauncherCommandRequest.class) + .setConsumer(action -> importer.run(ETable.S_DF_10)) + .forDestination(Task.accrualOfBalance.topic(), callbacks::put); // GBAL + callback(LauncherCommandRequest.class) + .setConsumer(action -> importer.run(ETable.S_DF_13)) + .forDestination(Task.getBalance.topic(), callbacks::put); // GBLD + init(); + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/exceptions/ConfigException.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/exceptions/ConfigException.java new file mode 100644 index 000000000..e91682613 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/exceptions/ConfigException.java @@ -0,0 +1,6 @@ +package ru.spcex.clearing.swt.importer.exceptions; + +public class ConfigException extends RuntimeException { + public ConfigException(String msg) { super(msg); } + public ConfigException(String msg, Throwable cause) { super(msg, cause); } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/Processor.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/Processor.java new file mode 100644 index 000000000..a34e67c5c --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/Processor.java @@ -0,0 +1,44 @@ +package ru.spcex.clearing.swt.importer.logic; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.swt.importer.logic.data.ResultContainer; +import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult; +import ru.spcex.clearing.swt.importer.logic.stages.Stage; + +import java.util.Arrays; +import java.util.List; + +@Component("processor") +public class Processor { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final List pipeline; + + public Processor(@Qualifier("pipeline") List pipeline) { + this.pipeline = pipeline; + } + + public void process(ResultContainer task) { + log.info("uuid {}. Task started", task.getUuid()); + long startMills = System.currentTimeMillis(); + for (Stage currStage : pipeline) { + if (statusIsFinal(task.getLastStageStatus()) && currStage.skipCompleted()) { + continue; + } + log.info("uuid {}. Stage: {}", task.getUuid(), currStage.getClass().getSimpleName()); + task.setLastStageStatus(currStage.process(task)); + log.info("uuid {}. Stage {} finished with status {}", task.getUuid(), currStage.getClass().getSimpleName(), task.getLastStageStatus()); + } + long endMills = System.currentTimeMillis(); + log.info("uuid {}. Task completed, result: {}, time working: {} ms", + task.getUuid(), + task, + endMills - startMills); + } + + private boolean statusIsFinal(StageResult previousStageStatus) { + return Arrays.asList(StageResult.ERROR, StageResult.COMPLETE).contains(previousStageStatus); + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ColumnStructureSWT.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ColumnStructureSWT.java new file mode 100644 index 000000000..390873c0f --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ColumnStructureSWT.java @@ -0,0 +1,51 @@ +package ru.spcex.clearing.swt.importer.logic.data; + +import ru.spcex.clearing.swt.importer.logic.data.enums.ColumnType; + +public class ColumnStructureSWT { + private String name; + private ColumnType type; + private int length; + private int decimalDigits; + private String comment; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public ColumnType getType() { + return type; + } + + public void setType(ColumnType type) { + this.type = type; + } + + public int getLength() { + return length; + } + + public void setLength(int length) { + this.length = length; + } + + public int getDecimalDigits() { + return decimalDigits; + } + + public void setDecimalDigits(int decimalDigits) { + this.decimalDigits = decimalDigits; + } + + public String getComment() { + return comment; + } + + public void setComment(String comment) { + this.comment = comment; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/DBStructureSWT.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/DBStructureSWT.java new file mode 100644 index 000000000..a8570935f --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/DBStructureSWT.java @@ -0,0 +1,18 @@ +package ru.spcex.clearing.swt.importer.logic.data; + +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; + +import java.util.EnumMap; +import java.util.Map; + +public class DBStructureSWT { + private Map tables = new EnumMap<>(ETable.class); + + public Map getTables() { + return tables; + } + + public void setTables(Map tables) { + this.tables = tables; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ResultContainer.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ResultContainer.java new file mode 100644 index 000000000..f145b8215 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ResultContainer.java @@ -0,0 +1,70 @@ +package ru.spcex.clearing.swt.importer.logic.data; + +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; +import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult; + +import java.io.File; +import java.util.UUID; + +/** + * Контейнер для передачи результата между стадиями + */ +public class ResultContainer { + private UUID uuid; + private ETable swtTable; + private File swtFile; + private StageResult lastStageStatus; + + private byte[] swtSource; + + protected ResultContainer() { + } + + public static ResultContainer createNewTask(ETable dbfTable, File dbfFile) { + ResultContainer container = new ResultContainer(); + container.swtTable = dbfTable; + container.swtFile = dbfFile; + container.uuid = UUID.randomUUID(); + return container; + } + + public ETable getSwtTable() { + return swtTable; + } + + public void setSwtTable(ETable swtTable) { + this.swtTable = swtTable; + } + + public byte[] getSwtSource() { + return swtSource; + } + + public void setSwtSource(byte[] swtSource) { + this.swtSource = swtSource; + } + + public File getSwtFile() { + return swtFile; + } + + public void setSwtFile(File swtFile) { + this.swtFile = swtFile; + } + + public UUID getUuid() { + return uuid; + } + + public void setUuid(UUID uuid) { + this.uuid = uuid; + } + + public StageResult getLastStageStatus() { + return lastStageStatus; + } + + public void setLastStageStatus(StageResult lastStageRes) { + this.lastStageStatus = lastStageRes; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/TableStructureSWT.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/TableStructureSWT.java new file mode 100644 index 000000000..c19be9fdd --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/TableStructureSWT.java @@ -0,0 +1,25 @@ +package ru.spcex.clearing.swt.importer.logic.data; + +import java.util.HashMap; +import java.util.Map; + +public class TableStructureSWT { + private Map columns = new HashMap<>(); + + public Map getColumns() { + return columns; + } + + public void setColumns(Map columns) { + this.columns = columns; + } + + public ColumnStructureSWT getIgnoreCase(String columnName) { + for (Map.Entry entry : columns.entrySet()) { + String currColumnName = entry.getKey(); + if (currColumnName.equalsIgnoreCase(columnName)) return entry.getValue(); + } + return null; + } + +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ColumnType.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ColumnType.java new file mode 100644 index 000000000..2a1f5a751 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ColumnType.java @@ -0,0 +1,46 @@ +package ru.spcex.clearing.swt.importer.logic.data.enums; + +import com.linuxense.javadbf.DBFDataType; + +import java.sql.Types; + +/** + * Типы данных в таблицах. + * Ставит в соответствие типы PostGRE и DBF + */ +public enum ColumnType { + VARCHAR(DBFDataType.CHARACTER, Types.VARCHAR), + CHARACTER(DBFDataType.CHARACTER, Types.CHAR), + NUMERIC(DBFDataType.NUMERIC, Types.NUMERIC), + DATE(DBFDataType.DATE, Types.DATE); + + private final DBFDataType dbfType; + private final int sqlType; + + ColumnType(DBFDataType dbfType, int postgreSqlType) { + this.dbfType = dbfType; + this.sqlType = postgreSqlType; + } + + public DBFDataType getDbfType() { + return dbfType; + } + + public int getSqlType() { + return sqlType; + } + + public static ColumnType getForSQLType(int sqlType) { + for (ColumnType columnType : values()) { + if (columnType.sqlType == sqlType) return columnType; + } + return null; + } + + public static ColumnType getForDBFType(DBFDataType dbfType) { + for (ColumnType columnType : values()) { + if (columnType.dbfType == dbfType) return columnType; + } + return null; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ETable.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ETable.java new file mode 100644 index 000000000..25a632513 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ETable.java @@ -0,0 +1,33 @@ +package ru.spcex.clearing.swt.importer.logic.data.enums; + +public enum ETable { + S_DF_08("S_DF08"), + S_DF_10("S_DF10"), + S_DF_13("S_DF13"); + //S_DF20("S_DF20"); + + private final String prefix; + + ETable(String prefix) { + this.prefix = prefix; + } + + public static ETable getTableForFilename(String filename) { + for (ETable table : ETable.values()) { + if (table.fileForThisTable(filename)) return table; + } + return null; + } + + public static ETable tableForName(String name) { + for (ETable table : values()) { + if (name.equalsIgnoreCase(table.name())) + return table; + } + return null; + } + + public boolean fileForThisTable(String filename) { + return filename != null && filename.startsWith(prefix); + } + } diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/StageResult.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/StageResult.java new file mode 100644 index 000000000..570ae98f0 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/StageResult.java @@ -0,0 +1,7 @@ +package ru.spcex.clearing.swt.importer.logic.data.enums; + +public enum StageResult { + OK, + ERROR, + COMPLETE +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/AbstractTable.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/AbstractTable.java new file mode 100644 index 000000000..93b60acd2 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/AbstractTable.java @@ -0,0 +1,46 @@ +package ru.spcex.clearing.swt.importer.logic.data.tables; + +import ru.spcex.platform.classes.base.SpcexObjectBase; +import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgHazelcast; +import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService; + +public abstract class AbstractTable { + private final String prefix; + private final Class clazz; + private final String nameOfMap; + protected HazelcastService hazelcastService; + protected ImdgHazelcast map; + protected String filename; + protected Long fileId; + + protected AbstractTable(String prefix, Class clazz, String nameOfMap) { + this.prefix = prefix; + this.clazz = clazz; + this.nameOfMap = nameOfMap; + } + + public void setFileId(Long fileId) { + this.fileId = fileId; + } + + public void setFilename(String filename) { + this.filename = filename; + } + + public void setHazelcastService(HazelcastService hazelcastService) { + this.hazelcastService = hazelcastService; + } + + public abstract T getEntity(Object[] entity); + + public void injectEntity(T obj) { + if (map == null) { + bootMap(); + } + map.insert(obj); + } + + private void bootMap() { + map = (ImdgHazelcast) hazelcastService.getImdg(nameOfMap, clazz); + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf08Table.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf08Table.java new file mode 100644 index 000000000..193b3ed80 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf08Table.java @@ -0,0 +1,33 @@ +package ru.spcex.clearing.swt.importer.logic.data.tables; + +import ru.clearing.classes.statics.data.sdf.SDf08; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; + +import java.time.Instant; + +public class SDf08Table extends AbstractTable { + + private static final String PREFIX = ETable.S_DF_08.name(); + private static final Class CLAZZ = SDf08.class; + private static final String NAME_OF_HZ_MAP = IMDGDistributedNames.Map_SDf08; + + public SDf08Table() { + super(PREFIX, CLAZZ, NAME_OF_HZ_MAP); + } + + @Override + public SDf08 getEntity(Object[] entity) { + SDf08 result = new SDf08(); + result.setOutDocument((String) entity[0]); + result.setDepoCode((String) entity[1]); + result.setQuantity((String) entity[2]); + result.setSecurityCode((String) entity[3]); + result.setClientName((String) entity[4]); + result.setFileName(filename); + result.setGenerationTime(Instant.now()); + result.setGenerationId(fileId); + return result; + } + +} \ No newline at end of file diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf10Table.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf10Table.java new file mode 100644 index 000000000..c1a885b20 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf10Table.java @@ -0,0 +1,33 @@ +package ru.spcex.clearing.swt.importer.logic.data.tables; + +import ru.clearing.classes.statics.data.sdf.SDf10; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; + +import java.time.Instant; + +public class SDf10Table extends AbstractTable { + + private static final String PREFIX = ETable.S_DF_10.name(); + private static final Class CLAZZ = SDf10.class; + private static final String NAME_OF_HZ_MAP = IMDGDistributedNames.Map_SDf10; + + public SDf10Table() { + super(PREFIX, CLAZZ, NAME_OF_HZ_MAP); + } + + @Override + public SDf10 getEntity(Object[] entity) { + SDf10 result = new SDf10(); + result.setOutDocument((String) entity[0]); + result.setDepoCode((String) entity[1]); + result.setQuantity((String) entity[2]); + result.setSecurityCode((String) entity[3]); + result.setClientName((String) entity[4]); + result.setFileName(filename); + result.setGenerationTime(Instant.now()); + result.setGenerationId(fileId); + return result; + } + +} \ No newline at end of file diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf13Table.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf13Table.java new file mode 100644 index 000000000..63385b7e5 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf13Table.java @@ -0,0 +1,29 @@ +package ru.spcex.clearing.swt.importer.logic.data.tables; + +import ru.clearing.classes.statics.data.sdf.SDf13; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; + +import java.time.Instant; + +public class SDf13Table extends AbstractTable { + + private static final String PREFIX = ETable.S_DF_13.name(); + private static final Class CLAZZ = SDf13.class; + private static final String NAME_OF_HZ_MAP = IMDGDistributedNames.Map_SDf13; + + public SDf13Table() { + super(PREFIX, CLAZZ, NAME_OF_HZ_MAP); + } + + @Override + public SDf13 getEntity(Object[] entity) { + SDf13 result = new SDf13(); + + result.setFileName(filename); + result.setGenerationTime(Instant.now()); + result.setGenerationId(fileId); + return result; + } + +} \ No newline at end of file diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ChangeDirOfFileStage.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ChangeDirOfFileStage.java new file mode 100644 index 000000000..60dfa77fb --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ChangeDirOfFileStage.java @@ -0,0 +1,79 @@ +package ru.spcex.clearing.swt.importer.logic.stages; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.clearing.swt.importer.logic.data.ResultContainer; +import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult; +import ru.spcex.platform.utils.time.TimeUtil; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.time.Instant; +import java.time.format.DateTimeFormatter; + +import static java.nio.file.StandardCopyOption.REPLACE_EXISTING; + + +@Component +public class ChangeDirOfFileStage extends Stage { + + private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy.MM.dd HH.mm.ss"); + private final ImportSWTServiceSettings settings; + + @Autowired + public ChangeDirOfFileStage(ImportSWTServiceSettings settings) { + this.settings = settings; + } + + @Override + public boolean skipCompleted() { + return false; + } + + @Override + public StageResult process(ResultContainer resultContainer) { + File srcDir = new File(settings.getStore().getSrcDir()); + File outDir; + if (resultContainer.getLastStageStatus().equals(StageResult.ERROR)) { + outDir = new File(settings.getStore().getOutDirError()); + + } else { + outDir = new File(settings.getStore().getOutDir()); + } + File dbfFile = resultContainer.getSwtFile(); + + if (!srcDir.exists()) { + log.error("SettlementHouse_DocIn does not exists!"); + return StageResult.ERROR; + } + if (!outDir.exists()) { + log.warn("ouput directory {} does not exists! Trying to made new!", outDir.getAbsolutePath()); + if (outDir.mkdir()) { + log.error("Made dir {} successfully!", outDir.getName()); + } else { + log.error("SettlementHouse_DocIn does not exists!"); + return StageResult.ERROR; + } + } + if (!dbfFile.exists()) { + log.error("SWT file to save does not exists!"); + return StageResult.ERROR; + } + String newNameOfFile = TimeUtil.formatInstant(Instant.now(), FORMATTER) + '_' + dbfFile.getName(); + + try { + Files.move(dbfFile.toPath(), outDir.toPath().resolve(newNameOfFile), REPLACE_EXISTING); + } catch (IOException e) { + log.error("Could not move file {}", e.getMessage()); + return StageResult.ERROR; + } + + if (resultContainer.getLastStageStatus().equals(StageResult.ERROR)) { + return StageResult.ERROR; + } else { + return StageResult.COMPLETE; + } + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ImportToDB.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ImportToDB.java new file mode 100644 index 000000000..3bd3fed06 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ImportToDB.java @@ -0,0 +1,73 @@ +package ru.spcex.clearing.swt.importer.logic.stages; + +import com.linuxense.javadbf.DBFReader; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.clearing.swt.importer.logic.data.ResultContainer; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; +import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult; +import ru.spcex.clearing.swt.importer.logic.data.tables.AbstractTable; +import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService; +import ru.spcex.platform.utils.log.ExceptionUtils; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.Charset; +import java.util.Map; + +/** + * Заливка проверенных данных в базу + */ +@Component +public class ImportToDB extends Stage { + private final ImportSWTServiceSettings settings; + private final HazelcastService hazelcastService; + private final Map mappingEnumTableObjectTable; + private final SWTImportKafkaMessenger kafkaMessenger; + + public ImportToDB(ImportSWTServiceSettings settings, + HazelcastService hazelcastService, + @Qualifier("mapOfTable") Map mappingEnumTableObjectTable, + SWTImportKafkaMessenger kafkaMessenger) { + this.settings = settings; + this.hazelcastService = hazelcastService; + this.mappingEnumTableObjectTable = mappingEnumTableObjectTable; + this.kafkaMessenger = kafkaMessenger; + } + + @Override + public StageResult process(ResultContainer resultContainer) { + ETable currTable = resultContainer.getSwtTable(); + byte[] source = resultContainer.getSwtSource(); + Charset sourceCharset = Charset.forName(settings.getCommon().getEncodingSource()); + + try (InputStream is = new ByteArrayInputStream(source); + DBFReader dbfReader = new DBFReader(is, sourceCharset)) { + AbstractTable table = mappingEnumTableObjectTable.get(currTable); + table.setHazelcastService(hazelcastService); + table.setFilename(resultContainer.getSwtFile().getName()); + Long fileId = hazelcastService.getImdgIdGenerator().nextId(); + table.setFileId(fileId); + for (int i = 0; i < dbfReader.getRecordCount(); i++) { + Object[] entity = dbfReader.nextRecord(); + if (entity == null) { + log.warn("record index {} null", i); + continue; + } + table.injectEntity(table.getEntity(entity)); + } + kafkaMessenger.notifySystemIfNeeded(currTable, fileId); + } catch (IOException exception) { + log.warn(exception.getMessage()); + return StageResult.ERROR; + } catch (Throwable e) { + log.error(ExceptionUtils.getStackTrace(e)); + return StageResult.ERROR; + } + + + return StageResult.OK; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/LoadFileFromDisk.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/LoadFileFromDisk.java new file mode 100644 index 000000000..1ed315a33 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/LoadFileFromDisk.java @@ -0,0 +1,44 @@ +package ru.spcex.clearing.swt.importer.logic.stages; + +import org.springframework.stereotype.Component; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.clearing.swt.importer.logic.data.ResultContainer; +import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Objects; +import java.util.UUID; + +@Component +public class LoadFileFromDisk extends Stage { + private final ImportSWTServiceSettings settings; + + public LoadFileFromDisk(ImportSWTServiceSettings settings) { + this.settings = settings; + } + + @Override + public StageResult process(ResultContainer resultContainer) { + Objects.requireNonNull(resultContainer.getSwtFile()); + UUID taskUuid = resultContainer.getUuid(); + + File dbfFile = resultContainer.getSwtFile(); + byte[] fileBytes; + try { + log.debug("uuid {}. Read all bytes from source file {}", taskUuid, resultContainer.getSwtFile().getName()); + fileBytes = Files.readAllBytes(Paths.get(dbfFile.getAbsolutePath())); + if (fileBytes.length == 0) throw new IOException("Empty file"); + if (settings.getStore().isDeleteSrcFiles()) Files.delete(dbfFile.toPath()); + } catch (IOException e) { + log.error(String.format("uuid %s. Can't read file %s", taskUuid, resultContainer.getSwtFile().getName()), e); + return StageResult.ERROR; + } + log.debug("uuid {}. Read all bytes from source file complete", taskUuid); + resultContainer.setSwtSource(fileBytes); + return StageResult.OK; + } + +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/SWTImportKafkaMessenger.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/SWTImportKafkaMessenger.java new file mode 100644 index 000000000..f37ec2930 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/SWTImportKafkaMessenger.java @@ -0,0 +1,51 @@ +package ru.spcex.clearing.swt.importer.logic.stages; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; +import ru.spcex.platform.enumeration.SdfTable; + +import java.util.HashMap; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Supplier; + +@Component +public class SWTImportKafkaMessenger implements InitializingBean { + private final Supplier kafka; + private final Map> messengers; + + public SWTImportKafkaMessenger(Supplier kafka) { + this.kafka = kafka; + this.messengers = new HashMap<>(); + } + + @Override + public void afterPropertiesSet() { + messengers.put(ETable.S_DF_08, groupId -> messageBalance(groupId, SdfTable.SDF_08)); + messengers.put(ETable.S_DF_10, groupId -> messageBalance(groupId, SdfTable.SDF_10)); + messengers.put(ETable.S_DF_13, groupId -> messageBalance(groupId, SdfTable.SDF_13)); + } + + /** + * отправляет в кафку сообщение, при необходимости + * (обрабатывается, например, в balance-service, clearing-service) + */ + public void notifySystemIfNeeded(ETable table, Long groupId) { + Consumer messenger = messengers.get(table); + if (messenger != null) { + messenger.accept(groupId); + } + } + + private void messageBalance(Long groupId, SdfTable table) { + StatementRequest statementRequest = new StatementRequest(); + statementRequest.setGroupId(groupId); + statementRequest.setTable(table); + kafka.get().sendRequestToQueue(Consts.STATEMENT_PROCESS, statementRequest); + } + +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/Stage.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/Stage.java new file mode 100644 index 000000000..b8b8fbc73 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/Stage.java @@ -0,0 +1,16 @@ +package ru.spcex.clearing.swt.importer.logic.stages; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.spcex.clearing.swt.importer.logic.data.ResultContainer; +import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult; + +public abstract class Stage { + protected Logger log = LoggerFactory.getLogger(getClass()); + + public abstract StageResult process(ResultContainer resultContainer); + + public boolean skipCompleted() { + return true; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ValidateFields.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ValidateFields.java new file mode 100644 index 000000000..0451716da --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ValidateFields.java @@ -0,0 +1,44 @@ +package ru.spcex.clearing.swt.importer.logic.stages; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.clearing.swt.importer.exceptions.ConfigException; +import ru.spcex.clearing.swt.importer.logic.data.ResultContainer; +import ru.spcex.clearing.swt.importer.logic.data.enums.StageResult; + +import java.nio.charset.Charset; + +/** + * Фильтрация источника на предмет соответствия полей + */ +@Component +public class ValidateFields extends Stage implements InitializingBean { + private final ImportSWTServiceSettings settings; + private Charset swtCharset; + + public ValidateFields(ImportSWTServiceSettings settings) { + this.settings = settings; + } + + @Override + public StageResult process(ResultContainer resultContainer) { + return StageResult.OK; + } + + /** + * 1. Проверяет кодировку из настройки dbf.encoding-source + */ + @Override + public void afterPropertiesSet() throws Exception { + initSWTCharset(); + } + + private void initSWTCharset() { + try { + swtCharset = Charset.forName(settings.getCommon().getEncodingSource()); + } catch (Exception e) { + throw new ConfigException("Unknown encoding from properties: " + settings.getCommon().getEncodingSource(), e); + } + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java new file mode 100644 index 000000000..e4d2d9a26 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java @@ -0,0 +1,56 @@ +package ru.spcex.clearing.swt.importer.services; + +import org.springframework.stereotype.Service; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; + +import java.io.File; +import java.util.*; + +@Service("fileChecker") +public class FileChecker { + private final ImportSWTServiceSettings settings; + + public FileChecker(ImportSWTServiceSettings settings) { + this.settings = settings; + } + + public Map> checkNewFiles() { + return checkNewFiles(null); + } + + public Map> checkNewFiles(ETable specificTable) { + Map> newFiles = new EnumMap<>(ETable.class); + + String srcDir = settings.getStore().getSrcDir(); + List dbfFiles = lsDBF(srcDir); + if (dbfFiles.isEmpty()) return newFiles; + + for (File dbfFile : dbfFiles) { + if (dbfFile.isDirectory()) continue; + + ETable currTable = ETable.getTableForFilename(dbfFile.getName()); + if (currTable == null) continue; + if (specificTable != null && !specificTable.equals(currTable)) continue; + List currList = newFiles.computeIfAbsent(currTable, list -> new LinkedList<>()); + currList.add(dbfFile); + } + return newFiles; + } + + private List lsDBF(String dbfDirPath) { + File dbfDir = new File(dbfDirPath); + File[] dbfFiles = dbfDir.listFiles((dir, name) -> { + int formatPosition = name.lastIndexOf("."); + if (formatPosition == -1 || formatPosition == name.length() - 1) return false; + return "dbf".equalsIgnoreCase(name.substring(formatPosition + 1)); + }); + + List resultFiles = new LinkedList<>(); + if (dbfFiles != null && dbfFiles.length >= 1) { + resultFiles.addAll(Arrays.asList(dbfFiles)); + } + + return resultFiles; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/SWTImporterService.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/SWTImporterService.java new file mode 100644 index 000000000..b7b65395f --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/SWTImporterService.java @@ -0,0 +1,119 @@ +package ru.spcex.clearing.swt.importer.services; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.stereotype.Service; +import ru.spcex.clearing.swt.importer.logic.Processor; +import ru.spcex.clearing.swt.importer.logic.data.ResultContainer; +import ru.spcex.clearing.swt.importer.logic.data.enums.ETable; + +import java.io.File; +import java.nio.file.Path; +import java.util.*; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +@Service("swtImporterService") +@EnableScheduling +public class SWTImporterService { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final FileChecker fileChecker; + private final ThreadPoolTaskExecutor executorService; + private final Processor processor; + private final Set filesCurrentlyInProcess; + + public SWTImporterService(@Qualifier("fileChecker") FileChecker messageListener, + @Qualifier("executor") ThreadPoolTaskExecutor executorService, + @Qualifier("processor") Processor processor) { + this.fileChecker = messageListener; + this.executorService = executorService; + this.processor = processor; + this.filesCurrentlyInProcess = new HashSet<>(); + } + + private static String forLogging(Map> files) { + return files + .entrySet() + .stream() + .flatMap((Function>, Stream>) entry -> { + List r = new LinkedList<>(); + for (File file : entry.getValue()) { + r.add(entry.getKey().name() + " " + file.toPath()); + } + return r.stream(); + }) + .collect(Collectors.joining(";", "[", "]")); + } + + private static String forLoggingSizeOnly(Map> files) { + return files + .entrySet() + .stream() + .map(entry -> entry.getKey().name() + " " + entry.getValue().size()) + .collect(Collectors.joining(";", "[", "]")); + } + + @Scheduled(cron = "${import-swt-service.scheduler.check-src-dir-cron}") + public void run() { + run(null); + } + + public void run(ETable specificTable) { + if (specificTable == null) { + log.trace("adding import task"); + } else { + log.info("adding import task {}", specificTable); + } + executorService.execute(() -> { + Map> newFiles = null; + try { + log.trace("checking new files... {}", specificTable != null ? specificTable.name() : ""); + newFiles = getFiles(specificTable); + if (newFiles.size() > 0 && log.isDebugEnabled()) { + log.debug("following files will be processed {}", forLogging(newFiles)); + } else if (newFiles.size() > 0) { + log.info("following files will be processed {}", forLoggingSizeOnly(newFiles)); + } else { + log.trace("no files were found"); + } + for (Map.Entry> newFilesEntry : newFiles.entrySet()) { + ETable currTable = newFilesEntry.getKey(); + List fileList = newFilesEntry.getValue(); + for (File swtFile : fileList) { + processor.process(ResultContainer.createNewTask(currTable, swtFile)); + } + } + } finally { + if (newFiles != null && newFiles.size() > 0) { + cleanFiles(newFiles); + } + } + }); + } + + private synchronized Map> getFiles(ETable specificTable) { + Map> newFiles = fileChecker.checkNewFiles(specificTable); + Map> newFilesFiltered = newFiles.entrySet() + .stream() + .map(entry -> + new AbstractMap.SimpleEntry<>(entry.getKey(), entry.getValue() + .stream() + .filter(file -> !filesCurrentlyInProcess.contains(file.toPath())) + .collect(Collectors.toList()))) + .collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue)); + newFilesFiltered.forEach((table, files) + -> files.forEach(file -> filesCurrentlyInProcess.add(file.toPath()))); + return newFilesFiltered; + } + + private synchronized void cleanFiles(Map> filesFromTask) { + filesFromTask.forEach((table, files) + -> files.forEach(file -> filesCurrentlyInProcess.remove(file.toPath()))); + } + +} diff --git a/clearing-parent/swt-importer/src/main/resources/application.properties b/clearing-parent/swt-importer/src/main/resources/application.properties new file mode 100644 index 000000000..93ff59338 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/resources/application.properties @@ -0,0 +1,34 @@ +server.port=8080 +server.servlet.context-path=/importer-swt +spring.main.web-application-type=servlet + +import-swt-service.scheduler.check-src-dir-cron=* * * * 1 ? + +import-swt-service.store.delete-src-files=false +import-swt-service.store.src-dir=/opt/clearing/file/importer/ +import-swt-service.store.out-dir=/opt/clearing/file/importer/loaded/ +import-swt-service.store.out-dir-error=/opt/clearing/file/importer/error/ + +import-swt-service.common.encoding-source=cp866 +import-swt-service.common.insert-batch-size=100 +import-swt-service.common.threads-count=10 + +import-swt-service.hazelcast.cluster-members=10.200.200.181:5701 +import-swt-service.hazelcast.login=dev +import-swt-service.hazelcast.password=dev-pass + + +import-swt-service.kafka-producer.bootstrap-servers=localhost:9092 +import-swt-service.kafka-producer.acks=all +import-swt-service.kafka-producer.retries=0 +import-swt-service.kafka-producer.batch-size=16384 +import-swt-service.kafka-producer.linger-ms=1 +import-swt-service.kafka-producer.buffer-memory=33554432 + +import-swt-service.kafka-consumer.bootstrap-servers=localhost:9092 +import-swt-service.kafka-consumer.group-id=dev-group-clearing-service +import-swt-service.kafka-consumer.enable-auto-commit=false +import-swt-service.kafka-consumer.session-timeout-ms=30000 +import-swt-service.kafka-consumer.auto-offset-reset=latest +import-swt-service.kafka-consumer.linger-ms=1 +import-swt-service.kafka-consumer.buffer-memory=33554432 \ No newline at end of file diff --git a/clearing-parent/swt-importer/src/main/resources/logback.xml b/clearing-parent/swt-importer/src/main/resources/logback.xml new file mode 100644 index 000000000..242290041 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/resources/logback.xml @@ -0,0 +1,37 @@ + + + + + UTF-8 + %date{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + ./logs/swt-importer.log + + UTF-8 + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + ../logs/swt-importer.%i.log + + 1 + 10 + + + 500MB + + + + + + + + + + + + + \ No newline at end of file diff --git a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java index 6a920d430..9cadeda18 100644 --- a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java +++ b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java @@ -3,16 +3,16 @@ package ru.spcex.platform.enumeration; import ru.spcex.platform.utils.enumeration.IEnumKey; public enum SdfTable implements IEnumKey { - SDF_01("SDF_01"), SDF_04("SDF_04"), SDF_08("SDF_08"), - SDF_13("SDF_13"), SDF_57("SDF_57"), SDF_09("SDF_09"), - SDF_16("SDF_16"); + SDF_08("SDF_08"), + SDF_10("SDF_10"), + SDF_13("SDF_13"); + + private final String key; SdfTable(String key) { this.key = key; } - private final String key; - @Override public String getKey() { return this.key; From a4b70c663582f804ca4ddca23f80d8941cf83372 Mon Sep 17 00:00:00 2001 From: AKurakin Date: Sat, 27 May 2023 13:42:30 +0300 Subject: [PATCH 4/5] =?UTF-8?q?company-service=20reports-service=20securit?= =?UTF-8?q?ies-service=20http://jira.mfd.msk:8088/browse/CLS-258=20=D0=BF?= =?UTF-8?q?=D0=BE=D0=BF=D1=80=D0=B0=D0=B2=D0=B8=D0=BB=20EndDtAfterStartDtR?= =?UTF-8?q?ule?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../common/rules/EndDtAfterStartDtRule.java | 39 ++++++++++--------- .../company/service/CompanyService.java | 2 + .../config/validation/ValidationConfig.java | 4 +- .../resources/distr/bin/securities-service.sh | 2 +- 4 files changed, 25 insertions(+), 22 deletions(-) diff --git a/clearing-parent/clearing-validation/src/main/java/ru/spcex/clearing/validation/common/rules/EndDtAfterStartDtRule.java b/clearing-parent/clearing-validation/src/main/java/ru/spcex/clearing/validation/common/rules/EndDtAfterStartDtRule.java index 1403e9338..92302ad9d 100644 --- a/clearing-parent/clearing-validation/src/main/java/ru/spcex/clearing/validation/common/rules/EndDtAfterStartDtRule.java +++ b/clearing-parent/clearing-validation/src/main/java/ru/spcex/clearing/validation/common/rules/EndDtAfterStartDtRule.java @@ -14,60 +14,61 @@ import java.util.function.Function; * @param Класс проверяемого объекта */ public record EndDtAfterStartDtRule( - String endDtFieldName, String startDtFieldName, - Function getterEndDt, + String endDtFieldName, Function getterStartDt, + Function getterEndDt, IErrorEnumId errorEmptyRequiredValue, IErrorEnumId errorEndDtAfterStartDt, boolean required ) implements IValidationRule> { /** - * @param endDtFieldName Название поля класса, содержащий EndDt * @param startDtFieldName Название поля класса, содержащий StartDt - * @param getterEndDt Метод получения проверяемой EndDt + * @param endDtFieldName Название поля класса, содержащий EndDt * @param getterStartDt Метод получения проверяемой StartDt + * @param getterEndDt Метод получения проверяемой EndDt * @param errorEmptyRequiredValue Ошибка, которая будет возвращена, если одна из дат пустая * @param errorEndDtAfterStartDt Ошибка, которая будет возвращена, если EndDt > StartDt * @param required Флаг обязательности поля * @param Класс проверяемого объекта */ - public static EndDtAfterStartDtRule instance(String endDtFieldName, - String startDtFieldName, - Function getterEndDt, + public static EndDtAfterStartDtRule instance(String startDtFieldName, + String endDtFieldName, Function getterStartDt, + Function getterEndDt, IErrorEnumId errorEmptyRequiredValue, IErrorEnumId errorEndDtAfterStartDt, boolean required) { - return new EndDtAfterStartDtRule<>(endDtFieldName, - startDtFieldName, - getterEndDt, + return new EndDtAfterStartDtRule<>(startDtFieldName, + endDtFieldName, getterStartDt, + getterEndDt, errorEmptyRequiredValue, errorEndDtAfterStartDt, required); } /** - * @param endDtFieldName Название поля класса, содержащий EndDt * @param startDtFieldName Название поля класса, содержащий StartDt - * @param getterEndDt Метод получения проверяемой EndDt + * @param endDtFieldName Название поля класса, содержащий EndDt * @param getterStartDt Метод получения проверяемой StartDt + * @param getterEndDt Метод получения проверяемой EndDt * @param errorEmptyRequiredValue Ошибка, которая будет возвращена, если одна из дат пустая - * @param errorEndDtAfterStartDt Ошибка, которая будет возвращена, если EndDt > StartDt + * @param errorEndDtAfterStartDt Ошибка, которая будет возвращена, если StartDt > EndDt * @param Класс проверяемого объекта */ - public static EndDtAfterStartDtRule instance(String endDtFieldName, + public static EndDtAfterStartDtRule instance( String startDtFieldName, - Function getterEndDt, + String endDtFieldName, Function getterStartDt, + Function getterEndDt, IErrorEnumId errorEmptyRequiredValue, IErrorEnumId errorEndDtAfterStartDt) { - return new EndDtAfterStartDtRule<>(endDtFieldName, - startDtFieldName, - getterEndDt, + return new EndDtAfterStartDtRule<>(startDtFieldName, + endDtFieldName, getterStartDt, + getterEndDt, errorEmptyRequiredValue, errorEndDtAfterStartDt, true); @@ -76,8 +77,8 @@ public record EndDtAfterStartDtRule( @Override public Optional validate(ImdgValidationContext context) { R validatedObject = context.getValidatedObject(); - LocalDate endDt = getterEndDt.apply(validatedObject); LocalDate startDt = getterStartDt.apply(validatedObject); + LocalDate endDt = getterEndDt.apply(validatedObject); if (endDt == null) return required ? of(errorEmptyRequiredValue, endDtFieldName) : Optional.empty(); if (startDt == null) return required ? of(errorEmptyRequiredValue, startDtFieldName) : Optional.empty(); if (startDt.isAfter(endDt)) return of(errorEndDtAfterStartDt, startDtFieldName, endDtFieldName); diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java index 855745cf9..148c94093 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java @@ -141,6 +141,7 @@ public class CompanyService extends QueueConsumer implements InitializingBean { companyMap.insert(company); log.debug("company-new request processed, BaseRequest.id = {}, company.id={}", companyNewRequestBaseRequest.getId(), company.getId()); + //todo see this CLS-333.2 relationService.createNewRelation(transaction, company.getId()); if (!(WorkflowStatus.Active.equalsByKey(company.getWorkflowStatus()))) { // Block relationService.onChangeWorkflowStatus(transaction, company, null, company.getWorkflowStatus()); @@ -196,6 +197,7 @@ public class CompanyService extends QueueConsumer implements InitializingBean { if (updateRequest.getWorkflowStatus() != null) { company.setWorkflowStatus(updateRequest.getWorkflowStatus()); if (!Objects.equals(prevStatus, company.getWorkflowStatus())) { + //todo see this CLS-333.2 relationService.onChangeWorkflowStatus(transaction, company, prevStatus, company.getWorkflowStatus()); } else { log.trace("Status was not changed"); diff --git a/clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/validation/ValidationConfig.java b/clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/validation/ValidationConfig.java index f01155eaf..5295a3da6 100644 --- a/clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/validation/ValidationConfig.java +++ b/clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/validation/ValidationConfig.java @@ -135,9 +135,9 @@ public class ValidationConfig { }, ReportError.WrongFieldValue, ReportError.RequiredFieldEmpty), - EndDtAfterStartDtRule.instance("endDate", "startDate", - ReportRequestWithPeriod::getEndDate, + EndDtAfterStartDtRule.instance("startDate", "endDate", ReportRequestWithPeriod::getStartDate, + ReportRequestWithPeriod::getEndDate, ReportError.RequiredFieldEmpty, ReportError.WrongFieldValue) ); diff --git a/z-distr/src/main/resources/distr/bin/securities-service.sh b/z-distr/src/main/resources/distr/bin/securities-service.sh index 7ad5dd7e9..5bee3a9d1 100644 --- a/z-distr/src/main/resources/distr/bin/securities-service.sh +++ b/z-distr/src/main/resources/distr/bin/securities-service.sh @@ -3,7 +3,7 @@ CLEARING_HOME=/opt/mfd/clearing/ cd $CLEARING_HOME/bin -CMD="java -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=7060 -jar securities-service.jar --spring.config.location=$CLEARING_HOME/settings/securities-service/" +CMD="java -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:7060 -jar securities-service.jar --spring.config.location=$CLEARING_HOME/settings/securities-service/" $CMD >/dev/null 2>&1 & From f3067be20d3cbf7c006ad0b7aa2e3308bc5c287b Mon Sep 17 00:00:00 2001 From: aalehin Date: Sat, 27 May 2023 13:50:15 +0300 Subject: [PATCH 5/5] http://jira.mfd.msk:8088/browse/CLS-316 --- FIX UP --- .../java/ru/spcex/platform/enumeration/SdfTable.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java index 9cadeda18..3221f97dd 100644 --- a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java +++ b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java @@ -3,16 +3,21 @@ package ru.spcex.platform.enumeration; import ru.spcex.platform.utils.enumeration.IEnumKey; public enum SdfTable implements IEnumKey { + SDF_01("SDF_01"), + SDF_04("SDF_04"), SDF_08("SDF_08"), + SDF_09("SDF_09"), SDF_10("SDF_10"), - SDF_13("SDF_13"); - - private final String key; + SDF_13("SDF_13"), + SDF_16("SDF_16"), + SDF_57("SDF_57"); SdfTable(String key) { this.key = key; } + private final String key; + @Override public String getKey() { return this.key;