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;