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

dbf importer (in progress)

Browse source 
(cherry picked from commit 6cd07131b209f002d3fa32ba850c7371b1700179)
This commit is contained in:
akulikov 2022-08-10 16:09:53 +03:00 • committed by AKurakin
parent ed8b6dba91
commit 0d607ebf52
13 changed files with 202 additions and 129 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -1,3 +1,4 @@
target
/.idea/
*.iml
logs

14
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFLoaderConfig.java → clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/DBFImporterConfig.java
View file

@ -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<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));
return pipeline;
}
@Bean("executor")
public ThreadPoolTaskExecutor executor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setMaxPoolSize(properties.getThreadsCount());
executor.setThreadNamePrefix("dbf-loader");
return executor;
}
}

13
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/controller/DefaultController.java
View file

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

27
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/data/ResultContainer.java
View file

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

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

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

30
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/Processor.java Normal file
View file

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

1
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ValidateFields.java
View file

@ -33,7 +33,6 @@ import java.util.Objects;
*/
@Component
public class ValidateFields extends Stage implements InitializingBean {
private final JdbcTemplate dbfJdbcTemplate;
private final AProperties properties;

33
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/properties/AProperties.java
View file

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

44
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFImporterService.java Normal file
View file

@ -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<Table, List<File>> newFiles = fileChecker.checkNewFiles();
for (Map.Entry<Table, List<File>> newFilesEntry : newFiles.entrySet()) {
Table currTable = newFilesEntry.getKey();
List<File> fileList = newFilesEntry.getValue();
for (File dbfFile : fileList) {
executorService.execute(() -> processor.process(ResultContainer.createNewTask(currTable, dbfFile)));
}
}
}
}

37
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFLoaderService.java
View file

@ -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<Stage> pipeline;
public DBFLoaderService(AProperties properties,
@Qualifier("fileMessageListener") MessageListener messageListener,
@Qualifier("pipeline") List<Stage> pipeline) {
this.properties = properties;
this.messageListener = messageListener;
this.pipeline = pipeline;
}
public void run() {
List<ResultContainer> 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;
}
}
}
}

43
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileMessageListener.java → clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileChecker.java
View file

@ -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<Table, List<File>> checkNewFiles() {
Map<Table, List<File>> newFiles = new HashMap<>();
String srcDir = properties.getSrcDir();
List<File> 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<byte[]> currList = sources.computeIfAbsent(currTable, k -> new LinkedList<>());
currList.add(fileBytes);
List<File> currList = newFiles.computeIfAbsent(currTable, list -> new LinkedList<>());
currList.add(dbfFile);
}
return newFiles;
}
private List<File> lsDBF(String dbfDirPath) {

35
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/MessageListener.java
View file

@ -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<Table, List<byte[]>> sources = new ConcurrentHashMap<>();
public List<ResultContainer> getTasks() {
List<ResultContainer> taskList = new LinkedList<>();
for (Map.Entry<Table, List<byte[]>> sourceEntry : sources.entrySet()) {
Table currTable = sourceEntry.getKey();
List<byte[]> 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();
}

5
clearing-parent/dbf-importer/src/main/resources/application.properties
View file

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