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

add dbf loader module

Browse source
This commit is contained in:
akulikov 2022-08-01 12:27:38 +03:00
parent f968c21992
commit 04087dd313
21 changed files with 1027 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
clearing-parent/dbf-loader/pom.xml Normal file
View file

@ -0,0 +1,87 @@
<?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">
<modelVersion>4.0.0</modelVersion>
<artifactId>dbf-loader</artifactId>
<name>dbf-loader</name>
<description>DBF loader for DBF files</description>
<version>SPCEX-1.0.0.0</version>
<parent>
<artifactId>clearing-parent</artifactId>
<groupId>ru.spcex.clearing</groupId>
<version>SPCEX-1.0.0.0</version>
</parent>
<dependencies>
<!-- Spring boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</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>
<version>0.9.5.2</version>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
</dependency>
<!-- DBF files -->
<dependency>
<groupId>com.github.albfernandez</groupId>
<artifactId>javadbf</artifactId>
<version>1.13.1</version>
</dependency>
</dependencies>
<build>
<finalName>jar/${project.artifactId}</finalName>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<archive>
<manifest>
<!--<classpathPrefix>../lib/</classpathPrefix>-->
<mainClass>ru.spcex.clearing.dbf.loader.DBFLoaderApplication</mainClass>
</manifest>
</archive>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
<configuration>
<addResources>true</addResources>
<classifier>exec</classifier>
</configuration>
</plugin>
</plugins>
</build>
</project>

18
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/DBFLoaderApplication.java Normal file
View file

@ -0,0 +1,18 @@
package ru.spcex.clearing.dbf.loader;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
@SpringBootApplication
public class DBFLoaderApplication {
public static void main(String[] args) {
try {
SpringApplicationBuilder builder = new SpringApplicationBuilder(DBFLoaderApplication.class);
builder.run(args);
} catch (Throwable e) {
LoggerFactory.getLogger(DBFLoaderApplication.class).error("DBF-Loader start failed: {} -> {}", e.getClass().getSimpleName(), e.getMessage());
System.exit(-1);
}
}
}

56
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/config/DBFLoaderConfig.java Normal file
View file

@ -0,0 +1,56 @@
package ru.spcex.clearing.dbf.loader.config;
import com.mchange.v2.c3p0.ComboPooledDataSource;
import org.springframework.beans.factory.annotation.Qualifier;
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.jdbc.core.JdbcTemplate;
import ru.spcex.clearing.dbf.loader.logic.stages.ImportToDB;
import ru.spcex.clearing.dbf.loader.logic.stages.Stage;
import ru.spcex.clearing.dbf.loader.logic.stages.ValidateFields;
import ru.spcex.clearing.dbf.loader.properties.AProperties;
import javax.sql.DataSource;
import java.beans.PropertyVetoException;
import java.util.LinkedList;
import java.util.List;
@Configuration
@ComponentScan(basePackages = {"ru.spcex.clearing.dbf.loader"})
public class DBFLoaderConfig {
private final AProperties properties;
private final ApplicationContext context;
public DBFLoaderConfig(@Qualifier("dbfLoaderProperties") AProperties properties, ApplicationContext context) {
this.properties = properties;
this.context = context;
}
@Bean("dbfDataSource")
public DataSource dataSource() throws PropertyVetoException {
ComboPooledDataSource dataSource = new ComboPooledDataSource();
dataSource.setDriverClass(properties.getDbDriver());
dataSource.setJdbcUrl(properties.getJdbcUrl());
dataSource.setUser(properties.getDbLogin());
dataSource.setPassword(properties.getDbPassword());
return dataSource;
}
@Bean("dbfJdbcTemplate")
public JdbcTemplate jdbcTemplate(@Qualifier("dbfDataSource") DataSource dataSource) {
return new JdbcTemplate(dataSource);
}
@Bean("pipeline")
public List<Stage> pipeline() {
List<Stage> pipeline = new LinkedList<>();
pipeline.add(context.getBean(ValidateFields.class));
pipeline.add(context.getBean(ImportToDB.class));
return pipeline;
}
}

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

@ -0,0 +1,6 @@
package ru.spcex.clearing.dbf.loader.exceptions;
public class ConfigException extends RuntimeException {
public ConfigException(String msg) { super(msg); }
public ConfigException(String msg, Throwable cause) { super(msg, cause); }
}

51
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/data/ColumnStructureDBF.java Normal file
View file

@ -0,0 +1,51 @@
package ru.spcex.clearing.dbf.loader.logic.data;
import ru.spcex.clearing.dbf.loader.logic.data.enums.ColumnType;
public class ColumnStructureDBF {
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/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/data/DBStructureDBF.java Normal file
View file

@ -0,0 +1,18 @@
package ru.spcex.clearing.dbf.loader.logic.data;
import ru.spcex.clearing.dbf.loader.logic.data.enums.Table;
import java.util.HashMap;
import java.util.Map;
public class DBStructureDBF {
private Map<Table, TableStructureDBF> tables = new HashMap<>();
public Map<Table, TableStructureDBF> getTables() {
return tables;
}
public void setTables(Map<Table, TableStructureDBF> tables) {
this.tables = tables;
}
}

36
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/data/ResultContainer.java Normal file
View file

@ -0,0 +1,36 @@
package ru.spcex.clearing.dbf.loader.logic.data;
import ru.spcex.clearing.dbf.loader.logic.data.enums.Table;
/**
* Контейнер для передачи результата между стадиями
*/
public class ResultContainer {
private Table dbfTable;
private byte[] dbfSource;
protected ResultContainer() {}
public static ResultContainer createNewTask(Table dbfTable, byte[] dbfSource) {
ResultContainer container = new ResultContainer();
container.dbfTable = dbfTable;
container.dbfSource = dbfSource;
return container;
}
public Table getDbfTable() {
return dbfTable;
}
public void setDbfTable(Table dbfTable) {
this.dbfTable = dbfTable;
}
public byte[] getDbfSource() {
return dbfSource;
}
public void setDbfSource(byte[] dbfSource) {
this.dbfSource = dbfSource;
}
}

25
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/data/TableStructureDBF.java Normal file
View file

@ -0,0 +1,25 @@
package ru.spcex.clearing.dbf.loader.logic.data;
import java.util.HashMap;
import java.util.Map;
public class TableStructureDBF {
private Map<String, ColumnStructureDBF> columns = new HashMap<>();
public Map<String, ColumnStructureDBF> getColumns() {
return columns;
}
public void setColumns(Map<String, ColumnStructureDBF> columns) {
this.columns = columns;
}
public ColumnStructureDBF getIgnoreCase(String columnName) {
for (Map.Entry<String, ColumnStructureDBF> entry : columns.entrySet()) {
String currColumnName = entry.getKey();
if (currColumnName.equalsIgnoreCase(columnName)) return entry.getValue();
}
return null;
}
}

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

@ -0,0 +1,46 @@
package ru.spcex.clearing.dbf.loader.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;
}
}

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

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

35
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/data/enums/Table.java Normal file
View file

@ -0,0 +1,35 @@
package ru.spcex.clearing.dbf.loader.logic.data.enums;
public enum Table {
DF_01("DF-01"),
DF_02("DF-02"),
DF_03("DF-03"),
DF_04("DF-04"),
DF_05("DF-05"),
DF_06("DF-06");
private final String prefix;
Table(String prefix) {
this.prefix = prefix;
}
public boolean fileForThisTable(String filename) {
return filename != null && filename.startsWith(prefix);
}
public static Table getTableForFilename(String filename) {
for (Table table : Table.values()) {
if (table.fileForThisTable(filename)) return table;
}
return null;
}
public static Table tableForName(String name) {
for (Table table : values()) {
if (name.equalsIgnoreCase(table.name()))
return table;
}
return null;
}
}

107
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/stages/ImportToDB.java Normal file
View file

@ -0,0 +1,107 @@
package ru.spcex.clearing.dbf.loader.logic.stages;
import com.linuxense.javadbf.DBFField;
import com.linuxense.javadbf.DBFReader;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.dbf.loader.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.loader.logic.data.enums.ColumnType;
import ru.spcex.clearing.dbf.loader.logic.data.enums.StageResult;
import ru.spcex.clearing.dbf.loader.logic.data.enums.Table;
import ru.spcex.clearing.dbf.loader.properties.AProperties;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
/**
* Заливка проверенных данных в базу
*/
@Component
public class ImportToDB extends Stage {
private final AProperties properties;
private final JdbcTemplate dbfJdbcTemplate;
public ImportToDB(@Qualifier("dbfLoaderProperties") AProperties properties,
@Qualifier("dbfJdbcTemplate") JdbcTemplate dbfJdbcTemplate) {
this.properties = properties;
this.dbfJdbcTemplate = dbfJdbcTemplate;
}
@Override
public StageResult process(ResultContainer resultContainer) {
Objects.requireNonNull(resultContainer.getDbfTable());
Objects.requireNonNull(resultContainer.getDbfSource());
Table currTable = resultContainer.getDbfTable();
byte[] source = resultContainer.getDbfSource();
Charset sourceCharset = Charset.forName(properties.getDbfEncoding());
try (InputStream is = new ByteArrayInputStream(source);
DBFReader dbfReader = new DBFReader(is, sourceCharset)) {
int columnCount = dbfReader.getFieldCount();
List<String> columnNames = new ArrayList<>(columnCount);
List<Integer> columnTypes = new ArrayList<>(columnCount);
for (int columnIdx = 0; columnIdx < columnCount; columnIdx++) {
DBFField currField = dbfReader.getField(columnIdx);
String fieldName = currField.getName();
ColumnType columnType = ColumnType.getForDBFType(currField.getType());
columnNames.add(fieldName);
columnTypes.add(columnType.getSqlType());
}
int recordCount = dbfReader.getRecordCount();
List<Object[]> values = new ArrayList<>(recordCount);
for (int rowIdx = 0; rowIdx < recordCount; rowIdx++) {
Object[] dbfRow = dbfReader.nextRecord();
values.add(dbfRow);
}
int maxBatchSize = properties.getInsertBatchSize();
int batchCount = recordCount / maxBatchSize;
int batchRem = recordCount % maxBatchSize;
for (int batchIdx = 0; batchIdx < batchCount + 1; batchIdx++) {
if (batchIdx == batchCount && batchRem == 0) continue;
final Integer currentBatchIdx = batchIdx;
int[] rowsInserted = dbfJdbcTemplate.batchUpdate(
"insert into " + currTable.name() +
" (" + String.join(",", columnNames) + ")" +
" values (" + String.join(",", Collections.nCopies(columnCount, "?")) + ")",
new BatchPreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps, int i) throws SQLException {
Object[] currValuesRow = values.get(currentBatchIdx * maxBatchSize + i);
int parameterIndex = 1;
for (Object object : currValuesRow) {
ps.setObject(parameterIndex, object, columnTypes.get(parameterIndex - 1));
parameterIndex++;
}
}
@Override
public int getBatchSize() {
return currentBatchIdx == batchCount ? batchRem : maxBatchSize;
}
});
log.info("Imported records: " + rowsInserted.length);
}
} catch (IOException e) {
e.printStackTrace();
}
return StageResult.OK;
}
}

12
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/stages/Stage.java Normal file
View file

@ -0,0 +1,12 @@
package ru.spcex.clearing.dbf.loader.logic.stages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.spcex.clearing.dbf.loader.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.loader.logic.data.enums.StageResult;
public abstract class Stage {
protected Logger log = LoggerFactory.getLogger(getClass());
public abstract StageResult process(ResultContainer resultContainer);
}

189
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/logic/stages/ValidateFields.java Normal file
View file

@ -0,0 +1,189 @@
package ru.spcex.clearing.dbf.loader.logic.stages;
import com.linuxense.javadbf.DBFDataType;
import com.linuxense.javadbf.DBFField;
import com.linuxense.javadbf.DBFReader;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.jdbc.support.MetaDataAccessException;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.dbf.loader.exceptions.ConfigException;
import ru.spcex.clearing.dbf.loader.logic.data.ColumnStructureDBF;
import ru.spcex.clearing.dbf.loader.logic.data.DBStructureDBF;
import ru.spcex.clearing.dbf.loader.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.loader.logic.data.TableStructureDBF;
import ru.spcex.clearing.dbf.loader.logic.data.enums.ColumnType;
import ru.spcex.clearing.dbf.loader.logic.data.enums.StageResult;
import ru.spcex.clearing.dbf.loader.logic.data.enums.Table;
import ru.spcex.clearing.dbf.loader.properties.AProperties;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.sql.ResultSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
/**
* Фильтрация источника на предмет соответствия полей
*/
@Component
public class ValidateFields extends Stage implements InitializingBean {
private final JdbcTemplate dbfJdbcTemplate;
private final AProperties properties;
private DBStructureDBF dbStructureDBF;
private Charset dbfCharset;
public ValidateFields(@Qualifier("dbfJdbcTemplate") JdbcTemplate dbfJdbcTemplate, AProperties properties) {
this.dbfJdbcTemplate = dbfJdbcTemplate;
this.properties = properties;
}
@Override
public StageResult process(ResultContainer resultContainer) {
Objects.requireNonNull(resultContainer.getDbfTable());
Objects.requireNonNull(resultContainer.getDbfSource());
Table currTable = resultContainer.getDbfTable();
byte[] source = resultContainer.getDbfSource();
TableStructureDBF currTableStructure = dbStructureDBF.getTables().get(currTable);
if (currTableStructure == null) {
log.warn("Unknown table {}, table was skipped.", currTable);
return StageResult.COMPLETE;
}
log.info("Check source " + currTable);
try (InputStream is = new ByteArrayInputStream(source);
DBFReader dbfReader = new DBFReader(is, dbfCharset)) {
int columnCount = dbfReader.getFieldCount();
if (columnCount != currTableStructure.getColumns().size()) {
log.warn("Source for table {} doesn't match DB columns count. Source column count: {}. DB column count: {}. Source was skipped.",
currTable,
columnCount,
currTableStructure.getColumns().size());
return StageResult.COMPLETE;
}
boolean valid = true;
for (int columnIdx = 0; columnIdx < columnCount; columnIdx++) {
DBFField currField = dbfReader.getField(columnIdx);
String currFieldName = currField.getName();
ColumnStructureDBF currColumnStructure = currTableStructure.getIgnoreCase(currFieldName);
if (currColumnStructure == null) {
log.warn("Source for table {} contain unknown field {}. ", currTable, currFieldName);
valid = false;
continue;
}
DBFDataType dbfDataType = currField.getType();
if (!currColumnStructure.getType().getDbfType().equals(dbfDataType)) {
log.warn("Field {} from source {} has wrong data type (actual {}, expected {}).",
currFieldName,
currTable,
dbfDataType.name(),
currColumnStructure.getType());
valid = false;
continue;
}
int fieldLength = currField.getLength();
if (fieldLength != currColumnStructure.getLength()) {
if (!dbfDataType.equals(DBFDataType.DATE)) {
log.warn("Field {} from source {} has wrong length (actual {}, expected {}).",
currFieldName,
currTable,
fieldLength,
currColumnStructure.getLength());
valid = false;
continue;
} else {
if (fieldLength > currColumnStructure.getLength()) {
log.warn("...");
}
}
}
int decimalCount = currField.getDecimalCount();
if (decimalCount != currColumnStructure.getDecimalDigits()) {
log.warn("Field {} from source {} has wrong decimal count (actual {}, expected {}).",
currFieldName,
currTable,
decimalCount,
currColumnStructure.getDecimalDigits());
valid = false;
}
if (!valid) {
log.warn("Source was skipped.");
return StageResult.COMPLETE;
}
}
} catch (IOException e) {
log.error("Can't read source for table {}. Source was skipped.", currTable);
return StageResult.ERROR;
}
return StageResult.OK;
}
/**
* 1. Собирает из БД названия столбцов для дальнейшей валидации
* 2. Проверяет кодировку из настройки dbf.encoding-source
*/
@Override
public void afterPropertiesSet() throws Exception {
initDBStructure();
initDBFCharset();
}
private void initDBStructure() throws MetaDataAccessException {
dbStructureDBF = JdbcUtils.extractDatabaseMetaData(Objects.requireNonNull(dbfJdbcTemplate.getDataSource()),
dbMeta -> {
DBStructureDBF dbStructureDBF = new DBStructureDBF();
List<String> tableNamesFromResultSet = new LinkedList<>();
ResultSet tableNamesRS = dbMeta.getTables(null, null, "%", new String[]{"TABLE"});
while (tableNamesRS.next()) {
tableNamesFromResultSet.add(tableNamesRS.getString("TABLE_NAME"));
}
for (String tableNameFromResultSet : tableNamesFromResultSet) {
Table currTable = Table.tableForName(tableNameFromResultSet);
if (currTable == null) continue;
TableStructureDBF tableStructureDBF = new TableStructureDBF();
ResultSet columnNamesRS = dbMeta.getColumns(null, null, tableNameFromResultSet, null);
while (columnNamesRS.next()) {
ColumnStructureDBF columnStructureDBF = new ColumnStructureDBF();
columnStructureDBF.setName(columnNamesRS.getString("COLUMN_NAME"));
columnStructureDBF.setType(ColumnType.getForSQLType(columnNamesRS.getInt("DATA_TYPE")));
columnStructureDBF.setLength(columnNamesRS.getInt("COLUMN_SIZE"));
columnStructureDBF.setDecimalDigits(columnNamesRS.getInt("DECIMAL_DIGITS"));
columnStructureDBF.setComment(columnNamesRS.getString("REMARKS"));
tableStructureDBF.getColumns().put(columnStructureDBF.getName(), columnStructureDBF);
}
assert !dbStructureDBF.getTables().containsKey(currTable);
dbStructureDBF.getTables().put(currTable, tableStructureDBF);
}
return dbStructureDBF;
});
}
private void initDBFCharset() {
try {
dbfCharset = Charset.forName(properties.getDbfEncoding());
} catch (Exception e) {
throw new ConfigException("В properties файле содержится неизвестная кодировка: " + properties.getDbfEncoding(), e);
}
}
}

121
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/properties/AProperties.java Normal file
View file

@ -0,0 +1,121 @@
package ru.spcex.clearing.dbf.loader.properties;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;
@Component("dbfLoaderProperties")
@PropertySource(value = {"classpath:dbf_loader.properties"})
@PropertySource(value = {"file:dbf_loader.properties"}, ignoreResourceNotFound = true)
public class AProperties {
@Value("${db.jdbc-url}")
private String jdbcUrl;
@Value("${db.driver}")
private String dbDriver;
@Value("${db.login}")
private String dbLogin;
@Value("${db.password}")
private String dbPassword;
@Value("${dbf.src-dir}")
private String srcDir;
@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;
public String getJdbcUrl() {
return jdbcUrl;
}
public void setJdbcUrl(String jdbcUrl) {
this.jdbcUrl = jdbcUrl;
}
public String getDbLogin() {
return dbLogin;
}
public void setDbLogin(String dbLogin) {
this.dbLogin = dbLogin;
}
public String getDbPassword() {
return dbPassword;
}
public void setDbPassword(String dbPassword) {
this.dbPassword = dbPassword;
}
public String getDbDriver() {
return dbDriver;
}
public void setDbDriver(String dbDriver) {
this.dbDriver = dbDriver;
}
public String getSrcDir() {
return srcDir;
}
public void setSrcDir(String srcDir) {
this.srcDir = srcDir;
}
public long getCheckSourceTimeout() {
return checkSourceTimeout;
}
public void setCheckSourceTimeout(long checkSourceTimeout) {
this.checkSourceTimeout = checkSourceTimeout;
}
public boolean deleteSrcFiles() {
return deleteSrcFiles;
}
public void setDeleteSrcFiles(boolean deleteSrcFiles) {
this.deleteSrcFiles = deleteSrcFiles;
}
public String getDbfEncoding() {
return dbfEncoding;
}
public void setDbfEncoding(String dbfEncoding) {
this.dbfEncoding = dbfEncoding;
}
public long getExecuteTimeout() {
return executeTimeout;
}
public void setExecuteTimeout(long executeTimeout) {
this.executeTimeout = executeTimeout;
}
public int getInsertBatchSize() {
return insertBatchSize;
}
public void setInsertBatchSize(int insertBatchSize) {
this.insertBatchSize = insertBatchSize;
}
}

41
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/services/DBFLoaderService.java Normal file
View file

@ -0,0 +1,41 @@
package ru.spcex.clearing.dbf.loader.services;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.dbf.loader.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.loader.logic.data.enums.StageResult;
import ru.spcex.clearing.dbf.loader.logic.stages.Stage;
import ru.spcex.clearing.dbf.loader.properties.AProperties;
import java.util.Arrays;
import java.util.List;
@Service
@EnableScheduling
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;
}
@Scheduled(fixedDelayString = "${dbf.execute-timeout}")
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;
}
}
}
}

87
clearing-parent/dbf-loader/src/main/java/ru/spcex/clearing/dbf/loader/services/FileMessageListener.java Normal file
View file

@ -0,0 +1,87 @@
package ru.spcex.clearing.dbf.loader.services;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.dbf.loader.exceptions.ConfigException;
import ru.spcex.clearing.dbf.loader.logic.data.enums.Table;
import ru.spcex.clearing.dbf.loader.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;
@Service("fileMessageListener")
@EnableScheduling
public class FileMessageListener extends MessageListener implements InitializingBean {
private final Logger log = LoggerFactory.getLogger(getClass());
private final AProperties properties;
public FileMessageListener(AProperties properties) {
this.properties = properties;
}
@Override
@Scheduled(fixedDelayString = "${dbf.check-source-timeout}")
public void checkMessage() {
String srcDir = properties.getSrcDir();
List<File> dbfFiles = lsDBF(srcDir);
if (dbfFiles.isEmpty()) return;
for (File dbfFile : dbfFiles) {
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);
}
}
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;
}
@Override
public void afterPropertiesSet() {
String dbfSourceDirPath = properties.getSrcDir();
File dbfSourceDir = new File(dbfSourceDirPath);
if (!dbfSourceDir.exists()) throw new ConfigException(String.format("DBF directory (%s) doesn't exist", dbfSourceDirPath));
if (!dbfSourceDir.isDirectory()) throw new ConfigException(String.format("DBF path (%s) isn't directory", dbfSourceDirPath));
}
}

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

@ -0,0 +1,35 @@
package ru.spcex.clearing.dbf.loader.services;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.dbf.loader.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.loader.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();
}

12
clearing-parent/dbf-loader/src/main/resources/dbf_loader.properties Normal file
View file

@ -0,0 +1,12 @@
db.jdbc-url=jdbc:postgresql://10.200.200.133:5432/postgres
db.driver=org.postgresql.Driver
db.login=clearing
db.password=Aa111111
dbf.check-source-timeout=300000000
dbf.execute-timeout=300000000
dbf.encoding-source=cp866
dbf.insert-batch-size=100
dbf.delete-src-files=false
dbf.src-dir=D:\\dbf\\

37
clearing-parent/dbf-loader/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>%d{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/dbf_loader.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/dbf_loader.%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>

1
clearing-parent/pom.xml
View file

@ -21,6 +21,7 @@
<module>backend-api</module>
<module>storage</module>
<module>db-scripts</module>
<module>dbf-loader</module>
</modules>
<properties>