treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

http://jira.mfd.msk:8088/browse/CLS-316

Browse source 
---
mid swt importer
This commit is contained in:
aalehin 2023-05-27 13:37:29 +03:00
parent 56441c39cb
commit 9fa165de2e
36 changed files with 1529 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
clearing-parent/pom.xml
View file

@ -39,6 +39,7 @@
<module>trade-importer</module>
<module>lim-exporter</module>
<module>swt-exporter</module>
<module>swt-importer</module>
</modules>
<properties>

103
clearing-parent/swt-importer/pom.xml Normal file
View file

@ -0,0 +1,103 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>clearing-parent</artifactId>
<groupId>ru.spcex.clearing</groupId>
<version>SPCEX-1.0.0.0</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>swt-importer</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
<dependencies>
<!-- Spring boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<!-- JDBC -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
</dependency>
<dependency>
<groupId>com.mchange</groupId>
<artifactId>c3p0</artifactId>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
</dependency>
<!-- DBF files -->
<dependency>
<groupId>com.github.albfernandez</groupId>
<artifactId>javadbf</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-imdg-api</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-imdg-api-hazelcast-impl</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.clearing</groupId>
<artifactId>classes</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-messaging</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-enum</artifactId>
</dependency>
</dependencies>
<build>
<resources>
<resource>
<directory>src/main/resources</directory>
<excludes>
<exclude>application.properties</exclude>
</excludes>
<filtering>false</filtering>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
<configuration>
<finalName>${project.artifactId}</finalName>
</configuration>
</plugin>
</plugins>
</build>
</project>

18
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/SWTImporterApplication.java Normal file
View file

@ -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);
}
}
}

46
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/ImporterImdgConfig.java Normal file
View file

@ -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());
}
}

60
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/KafkaConfig.java Normal file
View file

@ -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<String, Object> pf(ImportSWTServiceSettings settings) {
KafkaProducerSettings kafkaSettings = settings.getKafkaProducer();
return KafkaProducerFactory.producerFactory(kafkaSettings);
}
@Bean("kafkaTemplate")
public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> pf) {
return new KafkaTemplate<>(pf);
}
@Autowired
@Bean
public Supplier<KafkaSender> kafkaSender(KafkaTemplate<String, Object> kafkaTemplate,
ImdgProvider imdgProvider) {
ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
return () -> KafkaSender
.setup()
.setKafkaTemplate(kafkaTemplate)
.idGenerator(imdgIdGenerator::nextId)
.imdgProvider(s -> {
Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
return imdg::insert;
})
.build();
}
@Autowired
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@Bean
public Consumer<String, Object> createConsumer(ImportSWTServiceSettings settings) {
return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
}
}

67
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SWTImporterConfig.java Normal file
View file

@ -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<Stage> pipeline() {
List<Stage> 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<ETable, AbstractTable> getMapOfTables() {
Map<ETable, AbstractTable> 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;
}
}

32
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Common.java Normal file
View file

@ -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;
}
}

14
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Cron.java Normal file
View file

@ -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;
}
}

68
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/ImportSWTServiceSettings.java Normal file
View file

@ -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;
}
}

41
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java Normal file
View file

@ -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;
}
}

37
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/ImporterController.java Normal file
View file

@ -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");
}
}

43
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/controller/LauncherCommandReceiver.java Normal file
View file

@ -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<String, Object> 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();
}
}

6
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/exceptions/ConfigException.java Normal file
View file

@ -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); }
}

44
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/Processor.java Normal file
View file

@ -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<Stage> pipeline;
public Processor(@Qualifier("pipeline") List<Stage> 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);
}
}

51
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ColumnStructureSWT.java Normal file
View file

@ -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;
}
}

18
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/DBStructureSWT.java Normal file
View file

@ -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<ETable, TableStructureSWT> tables = new EnumMap<>(ETable.class);
public Map<ETable, TableStructureSWT> getTables() {
return tables;
}
public void setTables(Map<ETable, TableStructureSWT> tables) {
this.tables = tables;
}
}

70
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/ResultContainer.java Normal file
View file

@ -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;
}
}

25
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/TableStructureSWT.java Normal file
View file

@ -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<String, ColumnStructureSWT> columns = new HashMap<>();
public Map<String, ColumnStructureSWT> getColumns() {
return columns;
}
public void setColumns(Map<String, ColumnStructureSWT> columns) {
this.columns = columns;
}
public ColumnStructureSWT getIgnoreCase(String columnName) {
for (Map.Entry<String, ColumnStructureSWT> entry : columns.entrySet()) {
String currColumnName = entry.getKey();
if (currColumnName.equalsIgnoreCase(columnName)) return entry.getValue();
}
return null;
}
}

46
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ColumnType.java Normal file
View file

@ -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;
}
}

33
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/ETable.java Normal file
View file

@ -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);
}
}

7
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/enums/StageResult.java Normal file
View file

@ -0,0 +1,7 @@
package ru.spcex.clearing.swt.importer.logic.data.enums;
public enum StageResult {
OK,
ERROR,
COMPLETE
}

46
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/AbstractTable.java Normal file
View file

@ -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<T extends SpcexObjectBase> {
private final String prefix;
private final Class<T> clazz;
private final String nameOfMap;
protected HazelcastService hazelcastService;
protected ImdgHazelcast<T> map;
protected String filename;
protected Long fileId;
protected AbstractTable(String prefix, Class<T> 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<T>) hazelcastService.getImdg(nameOfMap, clazz);
}
}

33
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf08Table.java Normal file
View file

@ -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<SDf08> {
private static final String PREFIX = ETable.S_DF_08.name();
private static final Class<SDf08> 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;
}
}

33
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf10Table.java Normal file
View file

@ -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<SDf10> {
private static final String PREFIX = ETable.S_DF_10.name();
private static final Class<SDf10> 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;
}
}

29
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/data/tables/SDf13Table.java Normal file
View file

@ -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<SDf13> {
private static final String PREFIX = ETable.S_DF_13.name();
private static final Class<SDf13> 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;
}
}

79
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ChangeDirOfFileStage.java Normal file
View file

@ -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;
}
}
}

73
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ImportToDB.java Normal file
View file

@ -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<ETable, AbstractTable> mappingEnumTableObjectTable;
private final SWTImportKafkaMessenger kafkaMessenger;
public ImportToDB(ImportSWTServiceSettings settings,
HazelcastService hazelcastService,
@Qualifier("mapOfTable") Map<ETable, AbstractTable> 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;
}
}

44
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/LoadFileFromDisk.java Normal file
View file

@ -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;
}
}

51
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/SWTImportKafkaMessenger.java Normal file
View file

@ -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<KafkaSender> kafka;
private final Map<ETable, Consumer<Long>> messengers;
public SWTImportKafkaMessenger(Supplier<KafkaSender> 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<Long> 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);
}
}

16
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/Stage.java Normal file
View file

@ -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;
}
}

44
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/logic/stages/ValidateFields.java Normal file
View file

@ -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);
}
}
}

56
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java Normal file
View file

@ -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<ETable, List<File>> checkNewFiles() {
return checkNewFiles(null);
}
public Map<ETable, List<File>> checkNewFiles(ETable specificTable) {
Map<ETable, List<File>> newFiles = new EnumMap<>(ETable.class);
String srcDir = settings.getStore().getSrcDir();
List<File> 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<File> currList = newFiles.computeIfAbsent(currTable, list -> new LinkedList<>());
currList.add(dbfFile);
}
return newFiles;
}
private List<File> 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<File> resultFiles = new LinkedList<>();
if (dbfFiles != null && dbfFiles.length >= 1) {
resultFiles.addAll(Arrays.asList(dbfFiles));
}
return resultFiles;
}
}

119
clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/SWTImporterService.java Normal file
View file

@ -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<Path> 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<ETable, List<File>> files) {
return files
.entrySet()
.stream()
.flatMap((Function<Map.Entry<ETable, List<File>>, Stream<String>>) entry -> {
List<String> 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<ETable, List<File>> 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<ETable, List<File>> 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<ETable, List<File>> newFilesEntry : newFiles.entrySet()) {
ETable currTable = newFilesEntry.getKey();
List<File> 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<ETable, List<File>> getFiles(ETable specificTable) {
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles(specificTable);
Map<ETable, List<File>> 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<ETable, List<File>> filesFromTask) {
filesFromTask.forEach((table, files)
-> files.forEach(file -> filesCurrentlyInProcess.remove(file.toPath())));
}
}

34
clearing-parent/swt-importer/src/main/resources/application.properties Normal file
View file

@ -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

37
clearing-parent/swt-importer/src/main/resources/logback.xml Normal file
View file

@ -0,0 +1,37 @@
<configuration>
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<charset>UTF-8</charset>
<pattern>%date{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>./logs/swt-importer.log</file>
<encoder>
<charset>UTF-8</charset>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>
../logs/swt-importer.%i.log
</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>10</maxIndex>
</rollingPolicy>
<triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
<maxFileSize>500MB</maxFileSize>
</triggeringPolicy>
</appender>
<root level="warn">
<appender-ref ref="CONSOLE"/>
<appender-ref ref="FILE"/>
</root>
<logger name="ru.spcex" level="debug" additivity="false">
<appender-ref ref="FILE"/>
<appender-ref ref="CONSOLE"/>
</logger>
</configuration>

10
platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/SdfTable.java
View file

@ -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;