diff --git a/.gitignore b/.gitignore index eae80b53f..f60675edf 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ target /.idea/ *.iml +logs diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFLoaderConfig.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFImporterConfig.java similarity index 75% rename from clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFLoaderConfig.java rename to clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFImporterConfig.java index b7b91eecb..1ac82b01f 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFLoaderConfig.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFImporterConfig.java @@ -8,7 +8,9 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import ru.spcex.clearing.dbf.importer.logic.stages.ImportToDB; +import ru.spcex.clearing.dbf.importer.logic.stages.LoadFileFromDisk; import ru.spcex.clearing.dbf.importer.logic.stages.Stage; import ru.spcex.clearing.dbf.importer.logic.stages.ValidateFields; import ru.spcex.clearing.dbf.importer.properties.AProperties; @@ -21,11 +23,11 @@ import java.util.List; @Configuration @EnableConfigurationProperties @ComponentScan(basePackages = {"ru.spcex.clearing.dbf.importer"}) -public class DBFLoaderConfig { +public class DBFImporterConfig { private final AProperties properties; private final ApplicationContext context; - public DBFLoaderConfig(@Qualifier("dbfImporterProperties") AProperties properties, ApplicationContext context) { + public DBFImporterConfig(@Qualifier("dbfImporterProperties") AProperties properties, ApplicationContext context) { this.properties = properties; this.context = context; } @@ -49,10 +51,18 @@ public class DBFLoaderConfig { 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)); return pipeline; } + @Bean("executor") + public ThreadPoolTaskExecutor executor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setMaxPoolSize(properties.getThreadsCount()); + executor.setThreadNamePrefix("dbf-loader"); + return executor; + } } diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/controller/DefaultController.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/controller/DefaultController.java index a1af6e10e..61e715f88 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/controller/DefaultController.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/controller/DefaultController.java @@ -4,15 +4,22 @@ package ru.spcex.clearing.dbf.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.dbf.importer.services.DBFImporterService; @Controller("/") public class DefaultController implements InitializingBean { private final Logger log = LoggerFactory.getLogger(getClass()); + private final DBFImporterService importerService; + + public DefaultController(@Qualifier("dbfImporterService") DBFImporterService importerService) { + this.importerService = importerService; + } @RequestMapping(method = RequestMethod.GET, path = "/test", produces = MediaType.TEXT_PLAIN_VALUE) @ResponseBody @@ -21,6 +28,12 @@ public class DefaultController implements InitializingBean { return "importer controller test method"; } + @RequestMapping(method = RequestMethod.GET, path = "/check", produces = MediaType.TEXT_PLAIN_VALUE) + public void checkFolder() { + importerService.run(); + } + + @Override public void afterPropertiesSet() throws Exception { log.info("controller started"); diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/data/ResultContainer.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/data/ResultContainer.java index defaa8d8f..7a2471523 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/data/ResultContainer.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/data/ResultContainer.java @@ -2,19 +2,26 @@ package ru.spcex.clearing.dbf.importer.logic.data; import ru.spcex.clearing.dbf.importer.logic.data.enums.Table; +import java.io.File; +import java.util.UUID; + /** * Контейнер для передачи результата между стадиями */ public class ResultContainer { + private UUID uuid; private Table dbfTable; + private File dbfFile; + private byte[] dbfSource; protected ResultContainer() {} - public static ResultContainer createNewTask(Table dbfTable, byte[] dbfSource) { + public static ResultContainer createNewTask(Table dbfTable, File dbfFile) { ResultContainer container = new ResultContainer(); container.dbfTable = dbfTable; - container.dbfSource = dbfSource; + container.dbfFile = dbfFile; + container.uuid = UUID.randomUUID(); return container; } @@ -33,4 +40,20 @@ public class ResultContainer { public void setDbfSource(byte[] dbfSource) { this.dbfSource = dbfSource; } + + public File getDbfFile() { + return dbfFile; + } + + public void setDbfFile(File dbfFile) { + this.dbfFile = dbfFile; + } + + public UUID getUuid() { + return uuid; + } + + public void setUuid(UUID uuid) { + this.uuid = uuid; + } } diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/LoadFileFromDisk.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/LoadFileFromDisk.java new file mode 100644 index 000000000..6674fba18 --- /dev/null +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/LoadFileFromDisk.java @@ -0,0 +1,48 @@ +package ru.spcex.clearing.dbf.importer.logic.stages; + +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer; +import ru.spcex.clearing.dbf.importer.logic.data.enums.StageResult; +import ru.spcex.clearing.dbf.importer.properties.AProperties; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Objects; + +@Component +public class LoadFileFromDisk extends Stage { + private final AProperties properties; + + public LoadFileFromDisk(@Qualifier("dbfImporterProperties") AProperties properties) { + this.properties = properties; + } + + @Override + public StageResult process(ResultContainer resultContainer) { + Objects.requireNonNull(resultContainer.getDbfFile()); + + File dbfFile = resultContainer.getDbfFile(); + byte[] fileBytes; + try { + fileBytes = Files.readAllBytes(Paths.get(dbfFile.getAbsolutePath())); + if (properties.deleteSrcFiles()) { + boolean deleteOk = dbfFile.delete(); + if (!deleteOk) { + log.warn("Can't remove source file {}.", dbfFile.getPath()); + return StageResult.ERROR; + } + } + if (fileBytes.length == 0) throw new IOException("Empty file"); + } catch (IOException e) { + log.error("Can't read file " + dbfFile.getPath() + ". File was skipped.", e); + return StageResult.ERROR; + } + resultContainer.setDbfSource(fileBytes); + + return StageResult.OK; + } + +} diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/Processor.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/Processor.java new file mode 100644 index 000000000..04fa4cb97 --- /dev/null +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/Processor.java @@ -0,0 +1,30 @@ +package ru.spcex.clearing.dbf.importer.logic.stages; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer; +import ru.spcex.clearing.dbf.importer.logic.data.enums.StageResult; + +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("Start work with task " + task.getUuid()); + for (Stage currStage : pipeline) { + log.info("{} stage for task {}", currStage.getClass().getSimpleName(), task.getUuid()); + StageResult result = currStage.process(task); + if (Arrays.asList(StageResult.ERROR, StageResult.COMPLETE).contains(result)) break; + } + } +} diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ValidateFields.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ValidateFields.java index 0efb1aab8..42c765e91 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ValidateFields.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ValidateFields.java @@ -33,7 +33,6 @@ import java.util.Objects; */ @Component public class ValidateFields extends Stage implements InitializingBean { - private final JdbcTemplate dbfJdbcTemplate; private final AProperties properties; diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/properties/AProperties.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/properties/AProperties.java index 824d1ecba..9ea0bfe0b 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/properties/AProperties.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/properties/AProperties.java @@ -24,18 +24,15 @@ public class AProperties { @Value("${dbf.delete-src-files}") private boolean deleteSrcFiles = true; - @Value("${dbf.check-source-timeout}") - private long checkSourceTimeout; - - @Value("${dbf.execute-timeout}") - private long executeTimeout; - @Value("${dbf.encoding-source}") private String dbfEncoding; @Value("${dbf.insert-batch-size}") private int insertBatchSize; + @Value("${dbf.threads-count}") + private int threadsCount; + public String getJdbcUrl() { return jdbcUrl; } @@ -76,14 +73,6 @@ public class AProperties { this.srcDir = srcDir; } - public long getCheckSourceTimeout() { - return checkSourceTimeout; - } - - public void setCheckSourceTimeout(long checkSourceTimeout) { - this.checkSourceTimeout = checkSourceTimeout; - } - public boolean deleteSrcFiles() { return deleteSrcFiles; } @@ -100,14 +89,6 @@ public class AProperties { this.dbfEncoding = dbfEncoding; } - public long getExecuteTimeout() { - return executeTimeout; - } - - public void setExecuteTimeout(long executeTimeout) { - this.executeTimeout = executeTimeout; - } - public int getInsertBatchSize() { return insertBatchSize; } @@ -115,4 +96,12 @@ public class AProperties { 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/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFImporterService.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFImporterService.java new file mode 100644 index 000000000..d5001fdba --- /dev/null +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFImporterService.java @@ -0,0 +1,44 @@ +package ru.spcex.clearing.dbf.importer.services; + +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.dbf.importer.logic.data.ResultContainer; +import ru.spcex.clearing.dbf.importer.logic.data.enums.Table; +import ru.spcex.clearing.dbf.importer.logic.stages.Processor; +import ru.spcex.clearing.dbf.importer.properties.AProperties; + +import java.io.File; +import java.util.List; +import java.util.Map; + +@Service("dbfImporterService") +@EnableScheduling +public class DBFImporterService { + private final FileChecker fileChecker; + private final ThreadPoolTaskExecutor executorService; + private final Processor processor; + + public DBFImporterService(AProperties properties, + @Qualifier("fileChecker") FileChecker messageListener, + @Qualifier("executor") ThreadPoolTaskExecutor executorService, + @Qualifier("processor") Processor processor) { + this.fileChecker = messageListener; + this.executorService = executorService; + this.processor = processor; + } + + @Scheduled(cron = "${dbf.check-src-dir-cron}") + public void run() { + Map> newFiles = fileChecker.checkNewFiles(); + for (Map.Entry> newFilesEntry : newFiles.entrySet()) { + Table currTable = newFilesEntry.getKey(); + List fileList = newFilesEntry.getValue(); + for (File dbfFile : fileList) { + executorService.execute(() -> processor.process(ResultContainer.createNewTask(currTable, dbfFile))); + } + } + } +} diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFLoaderService.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFLoaderService.java deleted file mode 100644 index d965efd04..000000000 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFLoaderService.java +++ /dev/null @@ -1,37 +0,0 @@ -package ru.spcex.clearing.dbf.importer.services; - -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.stereotype.Service; -import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer; -import ru.spcex.clearing.dbf.importer.logic.data.enums.StageResult; -import ru.spcex.clearing.dbf.importer.logic.stages.Stage; -import ru.spcex.clearing.dbf.importer.properties.AProperties; - -import java.util.Arrays; -import java.util.List; - -@Service -public class DBFLoaderService { - private final AProperties properties; - private final MessageListener messageListener; - private final List pipeline; - - public DBFLoaderService(AProperties properties, - @Qualifier("fileMessageListener") MessageListener messageListener, - @Qualifier("pipeline") List pipeline) { - this.properties = properties; - this.messageListener = messageListener; - this.pipeline = pipeline; - } - - public void run() { - List newTaskList = messageListener.getTasks(); - for (ResultContainer newTask : newTaskList) { - for (Stage stage : pipeline) { - StageResult result = stage.process(newTask); - if (Arrays.asList(StageResult.ERROR, StageResult.COMPLETE).contains(result)) - break; - } - } - } -} diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileMessageListener.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileChecker.java similarity index 53% rename from clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileMessageListener.java rename to clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileChecker.java index 308c17201..e3a9f803a 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileMessageListener.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileChecker.java @@ -7,51 +7,38 @@ import ru.spcex.clearing.dbf.importer.logic.data.enums.Table; import ru.spcex.clearing.dbf.importer.properties.AProperties; import java.io.File; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Paths; -import java.util.Arrays; -import java.util.LinkedList; -import java.util.List; +import java.util.*; -@Service("fileMessageListener") -public class FileMessageListener extends MessageListener { +@Service("fileChecker") +public class FileChecker { private final Logger log = LoggerFactory.getLogger(getClass()); private final AProperties properties; - public FileMessageListener(AProperties properties) { + public FileChecker(AProperties properties) { this.properties = properties; } - @Override - public void checkMessage() { + public Map> checkNewFiles() { + Map> newFiles = new HashMap<>(); + String srcDir = properties.getSrcDir(); List dbfFiles = lsDBF(srcDir); - if (dbfFiles.isEmpty()) return; + if (dbfFiles.isEmpty()) return newFiles; for (File dbfFile : dbfFiles) { + if (dbfFile.isDirectory()) continue; + Table currTable = Table.getTableForFilename(dbfFile.getName()); if (currTable == null) continue; - byte[] fileBytes; - try { - fileBytes = Files.readAllBytes(Paths.get(dbfFile.getPath())); - if (properties.deleteSrcFiles()) { - boolean deleteOk = dbfFile.delete(); - if (!deleteOk) { - log.warn("Can't remove source file {}. File was skipped.", dbfFile.getPath()); - continue; - } - } - } catch (IOException e) { - log.error("Can't read file {}. File was skipped.", dbfFile.getPath()); - continue; - } - List currList = sources.computeIfAbsent(currTable, k -> new LinkedList<>()); - currList.add(fileBytes); + + List currList = newFiles.computeIfAbsent(currTable, list -> new LinkedList<>()); + currList.add(dbfFile); } + + return newFiles; } private List lsDBF(String dbfDirPath) { diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/MessageListener.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/MessageListener.java deleted file mode 100644 index 3ce245ee2..000000000 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/MessageListener.java +++ /dev/null @@ -1,35 +0,0 @@ -package ru.spcex.clearing.dbf.importer.services; - -import org.springframework.stereotype.Service; -import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer; -import ru.spcex.clearing.dbf.importer.logic.data.enums.Table; - -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -@Service -public abstract class MessageListener { - protected Map> sources = new ConcurrentHashMap<>(); - - public List getTasks() { - List taskList = new LinkedList<>(); - for (Map.Entry> sourceEntry : sources.entrySet()) { - Table currTable = sourceEntry.getKey(); - List sourceList = sourceEntry.getValue(); - for (byte[] sourceBytes : sourceList) { - ResultContainer currResultContainer = ResultContainer.createNewTask(currTable, sourceBytes); - taskList.add(currResultContainer); - } - sources.remove(currTable); - } - return taskList; - } - - /** - * Перегрузить для поставки sources в очередь - */ - public abstract void checkMessage(); - -} diff --git a/clearing-parent/dbf-importer/src/main/resources/application.properties b/clearing-parent/dbf-importer/src/main/resources/application.properties index 926bc6ebc..7ba588894 100644 --- a/clearing-parent/dbf-importer/src/main/resources/application.properties +++ b/clearing-parent/dbf-importer/src/main/resources/application.properties @@ -7,10 +7,11 @@ db.driver=org.postgresql.Driver db.login=clearing db.password=Aa111111 -dbf.check-source-timeout=300000000 -dbf.execute-timeout=300000000 +dbf.check-src-dir-cron=* * * * 1 ? dbf.encoding-source=cp866 dbf.insert-batch-size=100 dbf.delete-src-files=false dbf.src-dir=D:\\dbf\\ + +dbf.threads-count=1 \ No newline at end of file