some refactoring
This commit is contained in:
parent
66cbf4e6b2
commit
2d446f5239
18 changed files with 181 additions and 111 deletions
|
|
@ -92,6 +92,7 @@ public class StatementService extends QueueConsumer implements InitializingBean
|
||||||
.sorted(Comparator.comparing(SpcexObjectBase::getId))
|
.sorted(Comparator.comparing(SpcexObjectBase::getId))
|
||||||
.collect(Collectors.toList());
|
.collect(Collectors.toList());
|
||||||
List<AccountSdf01RequestPart> accountRequests = new ArrayList<>();
|
List<AccountSdf01RequestPart> accountRequests = new ArrayList<>();
|
||||||
|
Long generationIdForGroup = imdgProvider.getImdgIdGenerator().nextId();
|
||||||
for (SDf01 sdf01 : sdf01Group) {
|
for (SDf01 sdf01 : sdf01Group) {
|
||||||
IValidator validator = sDf01Validator.apply(sdf01);
|
IValidator validator = sDf01Validator.apply(sdf01);
|
||||||
Optional<EnumMessage> error = validator.tillFirstError();
|
Optional<EnumMessage> error = validator.tillFirstError();
|
||||||
|
|
@ -109,7 +110,7 @@ public class StatementService extends QueueConsumer implements InitializingBean
|
||||||
}
|
}
|
||||||
if (error.isPresent()) {
|
if (error.isPresent()) {
|
||||||
errorLogger.logError("sdf01.id={}", error.get(), sdf01.getId());
|
errorLogger.logError("sdf01.id={}", error.get(), sdf01.getId());
|
||||||
sdf02Imdg.insert(createErrorSdf02(sdf01, error.get()));
|
sdf02Imdg.insert(createErrorSdf02(sdf01, error.get(), generationIdForGroup));
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
Statement statement = statementImdg.getSingleObjectByFieldValues(Map.of("account", sdf01.getAccount()));
|
Statement statement = statementImdg.getSingleObjectByFieldValues(Map.of("account", sdf01.getAccount()));
|
||||||
|
|
@ -121,7 +122,7 @@ public class StatementService extends QueueConsumer implements InitializingBean
|
||||||
updateFlow(statement, sdf01,
|
updateFlow(statement, sdf01,
|
||||||
validator.getStored(ValidationStored.Sdf01Account));
|
validator.getStored(ValidationStored.Sdf01Account));
|
||||||
}
|
}
|
||||||
SDf02 sdf02New = createSuccessSdf02(sdf01);
|
SDf02 sdf02New = createSuccessSdf02(sdf01, generationIdForGroup);
|
||||||
sdf02Imdg.insert(sdf02New);
|
sdf02Imdg.insert(sdf02New);
|
||||||
statement.setOutSDfId(sdf02New.getId());
|
statement.setOutSDfId(sdf02New.getId());
|
||||||
AccountBalance accountBalance = createSuccessAccountBalance();
|
AccountBalance accountBalance = createSuccessAccountBalance();
|
||||||
|
|
@ -183,7 +184,7 @@ public class StatementService extends QueueConsumer implements InitializingBean
|
||||||
return r;
|
return r;
|
||||||
}
|
}
|
||||||
|
|
||||||
private SDf02 createErrorSdf02(SDf01 sdf01, EnumMessage error) {
|
private SDf02 createErrorSdf02(SDf01 sdf01, EnumMessage error, Long generationIdForGroup) {
|
||||||
SDf02 sDf02 = new SDf02();
|
SDf02 sDf02 = new SDf02();
|
||||||
sDf02.setCurr_code(sdf01.getCurr_code());
|
sDf02.setCurr_code(sdf01.getCurr_code());
|
||||||
sDf02.setAccount(sdf01.getAccount());
|
sDf02.setAccount(sdf01.getAccount());
|
||||||
|
|
@ -200,10 +201,12 @@ public class StatementService extends QueueConsumer implements InitializingBean
|
||||||
sDf02.setInSDf01Id(sdf01.getId());
|
sDf02.setInSDf01Id(sdf01.getId());
|
||||||
String errorId = error.getSubject().getId().toString();
|
String errorId = error.getSubject().getId().toString();
|
||||||
sDf02.setResult(errorId.substring(errorId.length() - 3));
|
sDf02.setResult(errorId.substring(errorId.length() - 3));
|
||||||
|
sDf02.setGenerationId(generationIdForGroup);
|
||||||
|
sDf02.setGenerationTime(Instant.now());
|
||||||
return sDf02;
|
return sDf02;
|
||||||
}
|
}
|
||||||
|
|
||||||
private SDf02 createSuccessSdf02(SDf01 sdf01) {
|
private SDf02 createSuccessSdf02(SDf01 sdf01, Long generationIdForGroup) {
|
||||||
SDf02 sDf02 = new SDf02();
|
SDf02 sDf02 = new SDf02();
|
||||||
sDf02.setCurr_code(sdf01.getCurr_code());
|
sDf02.setCurr_code(sdf01.getCurr_code());
|
||||||
sDf02.setAccount(sdf01.getAccount());
|
sDf02.setAccount(sdf01.getAccount());
|
||||||
|
|
@ -218,6 +221,8 @@ public class StatementService extends QueueConsumer implements InitializingBean
|
||||||
sDf02.setSumunblock(sdf01.getSumunblock());
|
sDf02.setSumunblock(sdf01.getSumunblock());
|
||||||
sDf02.setFile_type(sdf01.getFile_type());
|
sDf02.setFile_type(sdf01.getFile_type());
|
||||||
sDf02.setInSDf01Id(sdf01.getId());
|
sDf02.setInSDf01Id(sdf01.getId());
|
||||||
|
sDf02.setGenerationId(generationIdForGroup);
|
||||||
|
sDf02.setGenerationTime(Instant.now());
|
||||||
sDf02.setResult("OK!");
|
sDf02.setResult("OK!");
|
||||||
return sDf02;
|
return sDf02;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -5,13 +5,13 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||||
|
|
||||||
@SpringBootApplication
|
@SpringBootApplication
|
||||||
public class DBFLoaderApplication {
|
public class DBFImporterApplication {
|
||||||
public static void main(String[] args) {
|
public static void main(String[] args) {
|
||||||
try {
|
try {
|
||||||
SpringApplicationBuilder builder = new SpringApplicationBuilder(DBFLoaderApplication.class);
|
SpringApplicationBuilder builder = new SpringApplicationBuilder(DBFImporterApplication.class);
|
||||||
builder.run(args);
|
builder.run(args);
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
LoggerFactory.getLogger(DBFLoaderApplication.class).error("DBF-Loader start failed: {} -> {}", e.getClass().getSimpleName(), e.getMessage());
|
LoggerFactory.getLogger(DBFImporterApplication.class).error("DBF-Loader start failed: {} -> {}", e.getClass().getSimpleName(), e.getMessage());
|
||||||
System.exit(-1);
|
System.exit(-1);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -1,16 +1,15 @@
|
||||||
package ru.spcex.clearing.dbf.importer.config;
|
package ru.spcex.clearing.dbf.importer.config;
|
||||||
|
|
||||||
import org.springframework.beans.factory.annotation.Qualifier;
|
|
||||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||||
import org.springframework.context.ApplicationContext;
|
import org.springframework.context.ApplicationContext;
|
||||||
import org.springframework.context.annotation.Bean;
|
import org.springframework.context.annotation.Bean;
|
||||||
import org.springframework.context.annotation.ComponentScan;
|
import org.springframework.context.annotation.ComponentScan;
|
||||||
import org.springframework.context.annotation.Configuration;
|
import org.springframework.context.annotation.Configuration;
|
||||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||||
|
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
|
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.tables.*;
|
import ru.spcex.clearing.dbf.importer.logic.data.tables.*;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.stages.*;
|
import ru.spcex.clearing.dbf.importer.logic.stages.*;
|
||||||
import ru.spcex.clearing.dbf.importer.properties.AProperties;
|
|
||||||
|
|
||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.LinkedList;
|
import java.util.LinkedList;
|
||||||
|
|
@ -21,12 +20,12 @@ import java.util.Map;
|
||||||
@EnableConfigurationProperties
|
@EnableConfigurationProperties
|
||||||
@ComponentScan(basePackages = {"ru.spcex.clearing.dbf.importer"})
|
@ComponentScan(basePackages = {"ru.spcex.clearing.dbf.importer"})
|
||||||
public class DBFImporterConfig {
|
public class DBFImporterConfig {
|
||||||
private final AProperties properties;
|
private final ImportDBFServiceSettings settings;
|
||||||
private final ApplicationContext context;
|
private final ApplicationContext context;
|
||||||
|
|
||||||
|
|
||||||
public DBFImporterConfig(@Qualifier("dbfImporterProperties") AProperties properties, ApplicationContext context) {
|
public DBFImporterConfig(ImportDBFServiceSettings settings, ApplicationContext context) {
|
||||||
this.properties = properties;
|
this.settings = settings;
|
||||||
this.context = context;
|
this.context = context;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -45,8 +44,8 @@ public class DBFImporterConfig {
|
||||||
@Bean("executor")
|
@Bean("executor")
|
||||||
public ThreadPoolTaskExecutor executor() {
|
public ThreadPoolTaskExecutor executor() {
|
||||||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||||||
executor.setMaxPoolSize(properties.getThreadsCount());
|
executor.setMaxPoolSize(settings.getCommon().getThreadsCount());
|
||||||
executor.setCorePoolSize(properties.getThreadsCount());
|
executor.setCorePoolSize(settings.getCommon().getThreadsCount());
|
||||||
executor.setThreadNamePrefix("dbf-importer-thread-");
|
executor.setThreadNamePrefix("dbf-importer-thread-");
|
||||||
executor.setWaitForTasksToCompleteOnShutdown(true);
|
executor.setWaitForTasksToCompleteOnShutdown(true);
|
||||||
executor.setAwaitTerminationSeconds(300);
|
executor.setAwaitTerminationSeconds(300);
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,32 @@
|
||||||
|
package ru.spcex.clearing.dbf.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;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,14 @@
|
||||||
|
package ru.spcex.clearing.dbf.importer.config.settings;
|
||||||
|
|
||||||
|
public class Cron {
|
||||||
|
|
||||||
|
private String checkSrcDirCron;
|
||||||
|
|
||||||
|
public String getCheckSrcDirCron() {
|
||||||
|
return checkSrcDirCron;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void setCheckSrcDirCron(String checkSrcDirCron) {
|
||||||
|
this.checkSrcDirCron = checkSrcDirCron;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -12,6 +12,9 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
|
||||||
public class ImportDBFServiceSettings {
|
public class ImportDBFServiceSettings {
|
||||||
private HazelcastClientParams hazelcast;
|
private HazelcastClientParams hazelcast;
|
||||||
private KafkaProducerSettings kafkaProducer;
|
private KafkaProducerSettings kafkaProducer;
|
||||||
|
private Common common;
|
||||||
|
private Store store;
|
||||||
|
private Cron cron;
|
||||||
|
|
||||||
public HazelcastClientParams getHazelcast() {
|
public HazelcastClientParams getHazelcast() {
|
||||||
return hazelcast;
|
return hazelcast;
|
||||||
|
|
@ -28,4 +31,28 @@ public class ImportDBFServiceSettings {
|
||||||
public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
|
public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
|
||||||
this.kafkaProducer = kafkaProducer;
|
this.kafkaProducer = kafkaProducer;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,32 @@
|
||||||
|
package ru.spcex.clearing.dbf.importer.config.settings;
|
||||||
|
|
||||||
|
public class Store {
|
||||||
|
|
||||||
|
private String srcDir;
|
||||||
|
private String outDir;
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -2,9 +2,9 @@ package ru.spcex.clearing.dbf.importer.logic.stages;
|
||||||
|
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
|
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.data.enums.StageResult;
|
||||||
import ru.spcex.clearing.dbf.importer.properties.AProperties;
|
|
||||||
import ru.spcex.platform.utils.time.TimeUtil;
|
import ru.spcex.platform.utils.time.TimeUtil;
|
||||||
|
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
|
|
@ -20,18 +20,18 @@ import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
|
||||||
public class ChangeDirOfFileStage extends Stage {
|
public class ChangeDirOfFileStage extends Stage {
|
||||||
|
|
||||||
private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy.MM.dd HH.mm.ss");
|
private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy.MM.dd HH.mm.ss");
|
||||||
private final AProperties properties;
|
private final ImportDBFServiceSettings settings;
|
||||||
|
|
||||||
@Autowired
|
@Autowired
|
||||||
public ChangeDirOfFileStage(AProperties properties) {
|
public ChangeDirOfFileStage(ImportDBFServiceSettings settings) {
|
||||||
this.properties = properties;
|
this.settings = settings;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public StageResult process(ResultContainer resultContainer) {
|
public StageResult process(ResultContainer resultContainer) {
|
||||||
File srcDir = new File(properties.getSrcDir());
|
File srcDir = new File(settings.getStore().getSrcDir());
|
||||||
File outDir = new File(properties.getOutDir());
|
File outDir = new File(settings.getStore().getOutDir());
|
||||||
File dbfFile = resultContainer.getDbfFile();
|
File dbfFile = resultContainer.getDbfFile();
|
||||||
|
|
||||||
if (!srcDir.exists()) {
|
if (!srcDir.exists()) {
|
||||||
|
|
|
||||||
|
|
@ -3,11 +3,11 @@ package ru.spcex.clearing.dbf.importer.logic.stages;
|
||||||
import com.linuxense.javadbf.DBFReader;
|
import com.linuxense.javadbf.DBFReader;
|
||||||
import org.springframework.beans.factory.annotation.Qualifier;
|
import org.springframework.beans.factory.annotation.Qualifier;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
|
import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
|
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.enums.StageResult;
|
import ru.spcex.clearing.dbf.importer.logic.data.enums.StageResult;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.tables.AbstractTable;
|
import ru.spcex.clearing.dbf.importer.logic.data.tables.AbstractTable;
|
||||||
import ru.spcex.clearing.dbf.importer.properties.AProperties;
|
|
||||||
import ru.spcex.clearing.platform.messaging.domain.Consts;
|
import ru.spcex.clearing.platform.messaging.domain.Consts;
|
||||||
import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
|
import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
|
||||||
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
|
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
|
||||||
|
|
@ -25,15 +25,16 @@ import java.util.function.Supplier;
|
||||||
*/
|
*/
|
||||||
@Component
|
@Component
|
||||||
public class ImportToDB extends Stage {
|
public class ImportToDB extends Stage {
|
||||||
private final AProperties properties;
|
private final ImportDBFServiceSettings settings;
|
||||||
private final HazelcastService hazelcastService;
|
private final HazelcastService hazelcastService;
|
||||||
private final Map<ETable, AbstractTable> mappingEnumTableObjectTable;
|
private final Map<ETable, AbstractTable> mappingEnumTableObjectTable;
|
||||||
private final Supplier<KafkaSender> kafka;
|
private final Supplier<KafkaSender> kafka;
|
||||||
|
|
||||||
public ImportToDB(@Qualifier("dbfImporterProperties") AProperties properties,
|
public ImportToDB(ImportDBFServiceSettings settings,
|
||||||
HazelcastService hazelcastService, @Qualifier("mapOfTable") Map<ETable, AbstractTable> mappingEnumTableObjectTable,
|
HazelcastService hazelcastService,
|
||||||
|
@Qualifier("mapOfTable") Map<ETable, AbstractTable> mappingEnumTableObjectTable,
|
||||||
Supplier<KafkaSender> kafka) {
|
Supplier<KafkaSender> kafka) {
|
||||||
this.properties = properties;
|
this.settings = settings;
|
||||||
this.hazelcastService = hazelcastService;
|
this.hazelcastService = hazelcastService;
|
||||||
this.mappingEnumTableObjectTable = mappingEnumTableObjectTable;
|
this.mappingEnumTableObjectTable = mappingEnumTableObjectTable;
|
||||||
this.kafka = kafka;
|
this.kafka = kafka;
|
||||||
|
|
@ -43,7 +44,7 @@ public class ImportToDB extends Stage {
|
||||||
public StageResult process(ResultContainer resultContainer) {
|
public StageResult process(ResultContainer resultContainer) {
|
||||||
ETable currTable = resultContainer.getDbfTable();
|
ETable currTable = resultContainer.getDbfTable();
|
||||||
byte[] source = resultContainer.getDbfSource();
|
byte[] source = resultContainer.getDbfSource();
|
||||||
Charset sourceCharset = Charset.forName(properties.getDbfEncoding());
|
Charset sourceCharset = Charset.forName(settings.getCommon().getEncodingSource());
|
||||||
|
|
||||||
try (InputStream is = new ByteArrayInputStream(source);
|
try (InputStream is = new ByteArrayInputStream(source);
|
||||||
DBFReader dbfReader = new DBFReader(is, sourceCharset)) {
|
DBFReader dbfReader = new DBFReader(is, sourceCharset)) {
|
||||||
|
|
|
||||||
|
|
@ -1,10 +1,9 @@
|
||||||
package ru.spcex.clearing.dbf.importer.logic.stages;
|
package ru.spcex.clearing.dbf.importer.logic.stages;
|
||||||
|
|
||||||
import org.springframework.beans.factory.annotation.Qualifier;
|
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
|
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.data.enums.StageResult;
|
||||||
import ru.spcex.clearing.dbf.importer.properties.AProperties;
|
|
||||||
|
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
|
@ -15,10 +14,10 @@ import java.util.UUID;
|
||||||
|
|
||||||
@Component
|
@Component
|
||||||
public class LoadFileFromDisk extends Stage {
|
public class LoadFileFromDisk extends Stage {
|
||||||
private final AProperties properties;
|
private final ImportDBFServiceSettings settings;
|
||||||
|
|
||||||
public LoadFileFromDisk(@Qualifier("dbfImporterProperties") AProperties properties) {
|
public LoadFileFromDisk(ImportDBFServiceSettings settings) {
|
||||||
this.properties = properties;
|
this.settings = settings;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
@ -32,7 +31,7 @@ public class LoadFileFromDisk extends Stage {
|
||||||
log.debug("uuid {}. Read all bytes from source file {}", taskUuid, resultContainer.getDbfFile().getName());
|
log.debug("uuid {}. Read all bytes from source file {}", taskUuid, resultContainer.getDbfFile().getName());
|
||||||
fileBytes = Files.readAllBytes(Paths.get(dbfFile.getAbsolutePath()));
|
fileBytes = Files.readAllBytes(Paths.get(dbfFile.getAbsolutePath()));
|
||||||
if (fileBytes.length == 0) throw new IOException("Empty file");
|
if (fileBytes.length == 0) throw new IOException("Empty file");
|
||||||
if (properties.isDeleteSrcFiles()) Files.delete(dbfFile.toPath());
|
if (settings.getStore().isDeleteSrcFiles()) Files.delete(dbfFile.toPath());
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
log.error(String.format("uuid %s. Can't read file %s", taskUuid, resultContainer.getDbfFile().getName()), e);
|
log.error(String.format("uuid %s. Can't read file %s", taskUuid, resultContainer.getDbfFile().getName()), e);
|
||||||
return StageResult.ERROR;
|
return StageResult.ERROR;
|
||||||
|
|
|
||||||
|
|
@ -2,10 +2,10 @@ package ru.spcex.clearing.dbf.importer.logic.stages;
|
||||||
|
|
||||||
import org.springframework.beans.factory.InitializingBean;
|
import org.springframework.beans.factory.InitializingBean;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
|
||||||
import ru.spcex.clearing.dbf.importer.exceptions.ConfigException;
|
import ru.spcex.clearing.dbf.importer.exceptions.ConfigException;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
|
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.data.enums.StageResult;
|
||||||
import ru.spcex.clearing.dbf.importer.properties.AProperties;
|
|
||||||
|
|
||||||
import java.nio.charset.Charset;
|
import java.nio.charset.Charset;
|
||||||
|
|
||||||
|
|
@ -14,11 +14,11 @@ import java.nio.charset.Charset;
|
||||||
*/
|
*/
|
||||||
@Component
|
@Component
|
||||||
public class ValidateFields extends Stage implements InitializingBean {
|
public class ValidateFields extends Stage implements InitializingBean {
|
||||||
private final AProperties properties;
|
private final ImportDBFServiceSettings settings;
|
||||||
private Charset dbfCharset;
|
private Charset dbfCharset;
|
||||||
|
|
||||||
public ValidateFields(AProperties properties) {
|
public ValidateFields(ImportDBFServiceSettings settings) {
|
||||||
this.properties = properties;
|
this.settings = settings;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
@ -36,9 +36,9 @@ public class ValidateFields extends Stage implements InitializingBean {
|
||||||
|
|
||||||
private void initDBFCharset() {
|
private void initDBFCharset() {
|
||||||
try {
|
try {
|
||||||
dbfCharset = Charset.forName(properties.getDbfEncoding());
|
dbfCharset = Charset.forName(settings.getCommon().getEncodingSource());
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
throw new ConfigException("Unknown encoding from properties: " + properties.getDbfEncoding(), e);
|
throw new ConfigException("Unknown encoding from properties: " + settings.getCommon().getEncodingSource(), e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,45 +0,0 @@
|
||||||
package ru.spcex.clearing.dbf.importer.properties;
|
|
||||||
|
|
||||||
import org.springframework.beans.factory.annotation.Value;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
@Component("dbfImporterProperties")
|
|
||||||
public class AProperties {
|
|
||||||
|
|
||||||
@Value("${dbf.src-dir}")
|
|
||||||
private String srcDir;
|
|
||||||
@Value("${dbf.out-dir}")
|
|
||||||
private String outDir;
|
|
||||||
@Value("${dbf.delete-src-files}")
|
|
||||||
private boolean deleteSrcFiles = true;
|
|
||||||
@Value("${dbf.encoding-source}")
|
|
||||||
private String dbfEncoding;
|
|
||||||
@Value("${dbf.insert-batch-size}")
|
|
||||||
private int insertBatchSize;
|
|
||||||
@Value("${dbf.threads-count}")
|
|
||||||
private int threadsCount;
|
|
||||||
|
|
||||||
public String getOutDir() {
|
|
||||||
return outDir;
|
|
||||||
}
|
|
||||||
|
|
||||||
public String getSrcDir() {
|
|
||||||
return srcDir;
|
|
||||||
}
|
|
||||||
|
|
||||||
public boolean isDeleteSrcFiles() {
|
|
||||||
return deleteSrcFiles;
|
|
||||||
}
|
|
||||||
|
|
||||||
public String getDbfEncoding() {
|
|
||||||
return dbfEncoding;
|
|
||||||
}
|
|
||||||
|
|
||||||
public int getInsertBatchSize() {
|
|
||||||
return insertBatchSize;
|
|
||||||
}
|
|
||||||
|
|
||||||
public int getThreadsCount() {
|
|
||||||
return threadsCount;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -28,7 +28,7 @@ public class DBFImporterService {
|
||||||
this.processor = processor;
|
this.processor = processor;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Scheduled(cron = "${dbf.check-src-dir-cron}")
|
@Scheduled(cron = "${import-dbf-service.scheduler.check-src-dir-cron}")
|
||||||
public void run() {
|
public void run() {
|
||||||
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles();
|
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles();
|
||||||
for (Map.Entry<ETable, List<File>> newFilesEntry : newFiles.entrySet()) {
|
for (Map.Entry<ETable, List<File>> newFilesEntry : newFiles.entrySet()) {
|
||||||
|
|
|
||||||
|
|
@ -1,24 +1,24 @@
|
||||||
package ru.spcex.clearing.dbf.importer.services;
|
package ru.spcex.clearing.dbf.importer.services;
|
||||||
|
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
|
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
|
||||||
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
|
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
|
||||||
import ru.spcex.clearing.dbf.importer.properties.AProperties;
|
|
||||||
|
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
import java.util.*;
|
import java.util.*;
|
||||||
|
|
||||||
@Service("fileChecker")
|
@Service("fileChecker")
|
||||||
public class FileChecker {
|
public class FileChecker {
|
||||||
private final AProperties properties;
|
private final ImportDBFServiceSettings settings;
|
||||||
|
|
||||||
public FileChecker(AProperties properties) {
|
public FileChecker(ImportDBFServiceSettings settings) {
|
||||||
this.properties = properties;
|
this.settings = settings;
|
||||||
}
|
}
|
||||||
|
|
||||||
public Map<ETable, List<File>> checkNewFiles() {
|
public Map<ETable, List<File>> checkNewFiles() {
|
||||||
Map<ETable, List<File>> newFiles = new EnumMap<>(ETable.class);
|
Map<ETable, List<File>> newFiles = new EnumMap<>(ETable.class);
|
||||||
|
|
||||||
String srcDir = properties.getSrcDir();
|
String srcDir = settings.getStore().getSrcDir();
|
||||||
List<File> dbfFiles = lsDBF(srcDir);
|
List<File> dbfFiles = lsDBF(srcDir);
|
||||||
if (dbfFiles.isEmpty()) return newFiles;
|
if (dbfFiles.isEmpty()) return newFiles;
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,14 +1,18 @@
|
||||||
server.port=8080
|
server.port=8080
|
||||||
server.servlet.context-path=/importer
|
server.servlet.context-path=/importer
|
||||||
spring.main.web-application-type=servlet
|
spring.main.web-application-type=servlet
|
||||||
dbf.check-src-dir-cron=* * * * 1 ?
|
|
||||||
dbf.encoding-source=cp866
|
import-dbf-service.scheduler.check-src-dir-cron=* * * * 1 ?
|
||||||
dbf.insert-batch-size=100
|
|
||||||
dbf.delete-src-files=false
|
import-dbf-service.store.delete-src-files=false
|
||||||
dbf.src-dir=/opt/clearing/file/importer/
|
import-dbf-service.store.src-dir=/opt/clearing/file/importer/
|
||||||
dbf.out-dir=/opt/clearing/file/importer/loaded/
|
import-dbf-service.store.out-dir=/opt/clearing/file/importer/loaded/
|
||||||
dbf.threads-count=10
|
|
||||||
import-dbf-service.hazelcast.cluster-members=127.0.0.1:5701
|
import-dbf-service.common.encoding-source=cp866
|
||||||
|
import-dbf-service.common.insert-batch-size=100
|
||||||
|
import-dbf-service.common.threads-count=10
|
||||||
|
|
||||||
|
import-dbf-service.hazelcast.cluster-members=10.200.200.181:5071
|
||||||
import-dbf-service.hazelcast.login=dev
|
import-dbf-service.hazelcast.login=dev
|
||||||
import-dbf-service.hazelcast.password=dev-pass
|
import-dbf-service.hazelcast.password=dev-pass
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -47,14 +47,16 @@ public class CompanyMapStore extends BusinessObjectMapStore<Company> {
|
||||||
return namedParameterJdbcTemplate.query("select * from " + getTableName() + " where id in (:ids)", paramMap,
|
return namedParameterJdbcTemplate.query("select * from " + getTableName() + " where id in (:ids)", paramMap,
|
||||||
(resultSet, i) -> {
|
(resultSet, i) -> {
|
||||||
Company company = new Company();
|
Company company = new Company();
|
||||||
fillBusinessObjectFields(company, resultSet);
|
company.setId(resultSet.getObject("ID", Long.class));
|
||||||
|
company.setCreated(getInstantFromTimestamp(resultSet, "CREATED_AT"));
|
||||||
|
company.setUpdated(getInstantFromTimestamp(resultSet, "UPDATED_AT"));
|
||||||
company.setTradingCode(resultSet.getObject("TRADING_CODE", String.class));
|
company.setTradingCode(resultSet.getObject("TRADING_CODE", String.class));
|
||||||
company.setClearingCode(resultSet.getObject("CLEARING_CODE", String.class));
|
company.setClearingCode(resultSet.getObject("CLEARING_CODE", String.class));
|
||||||
company.setRegistrationCode(resultSet.getObject("REGISTRATION_CODE", String.class));
|
company.setRegistrationCode(resultSet.getObject("REGISTRATION_CODE", String.class));
|
||||||
company.setWorkflowStatus(resultSet.getObject("WORKFLOW_STATUS", String.class));
|
company.setWorkflowStatus(resultSet.getObject("WORKFLOW_STATUS", String.class));
|
||||||
company.setShortName(resultSet.getObject("SHORT_NAME", String.class));
|
company.setShortName(resultSet.getObject("SHORT_NAME", String.class));
|
||||||
company.setFullName(resultSet.getObject("FULL_NAME", String.class));
|
company.setFullName(resultSet.getObject("FULL_NAME", String.class));
|
||||||
company.setProfile(getCompanyInfo(company.getId()));
|
// company.setProfile(getCompanyInfo(company.getId()));
|
||||||
return company;
|
return company;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
@ -89,7 +91,7 @@ public class CompanyMapStore extends BusinessObjectMapStore<Company> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
batchInsertUpdate(insertStatement, batchArgs);
|
batchInsertUpdate(insertStatement, batchArgs);
|
||||||
batchInsertUpdate(insertCompanyInfoStatement, batchCompanyInfoArgs);
|
// batchInsertUpdate(insertCompanyInfoStatement, batchCompanyInfoArgs);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -30,7 +30,7 @@ public class RelationMapStore extends TemplateMapStore<Relation> {
|
||||||
@Override
|
@Override
|
||||||
public String[] getFields() {
|
public String[] getFields() {
|
||||||
return new String[]{
|
return new String[]{
|
||||||
"ID", "CREATED_AT", "UPDATED_AT", "CONSUMER_ID", "SUPPLIER_ID", "SERVICE_STATUS", "SERVICE", "SERVICE_PRODUCT", "COMMENT"
|
"ID", "CREATED_AT", "UPDATED_AT", "CONSUMER_ID", "SUPPLIER_ID", "SERVICE_STATUS", "SERVICE", "SERVICE_PRODUCT", "COMMENT"
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -38,14 +38,14 @@ public class RelationMapStore extends TemplateMapStore<Relation> {
|
||||||
public Relation objectReader(ResultSet resultSet) throws SQLException {
|
public Relation objectReader(ResultSet resultSet) throws SQLException {
|
||||||
Relation object = new Relation();
|
Relation object = new Relation();
|
||||||
object.setId(resultSet.getObject("ID", Long.class));
|
object.setId(resultSet.getObject("ID", Long.class));
|
||||||
object.setCreated(getInstantFromTimestamp(resultSet, "CREATED_AT"));
|
object.setCreated(getInstantFromTimestamp(resultSet, "CREATED_AT"));
|
||||||
object.setUpdated(getInstantFromTimestamp(resultSet, "UPDATED_AT"));
|
object.setUpdated(getInstantFromTimestamp(resultSet, "UPDATED_AT"));
|
||||||
object.setConsumerId(resultSet.getObject("CONSUMER_ID", Long.class));
|
object.setConsumerId(resultSet.getObject("CONSUMER_ID", Long.class));
|
||||||
object.setSupplierId(resultSet.getObject("SUPPLIER_ID", Long.class));
|
object.setSupplierId(resultSet.getObject("SUPPLIER_ID", Long.class));
|
||||||
object.setServiceStatus(resultSet.getObject("SERVICE_STATUS", String.class));
|
object.setServiceStatus(resultSet.getObject("SERVICE_STATUS", String.class));
|
||||||
object.setService(resultSet.getObject("SERVICE", Long.class));
|
object.setService(resultSet.getObject("SERVICE", Long.class));
|
||||||
object.setServiceProduct(resultSet.getObject("SERVICE_PRODUCT", Long.class));
|
object.setServiceProduct(resultSet.getObject("SERVICE_PRODUCT", Long.class));
|
||||||
object.setComment(resultSet.getObject("COMMENT", String.class));
|
object.setComment(resultSet.getObject("COMMENT", String.class));
|
||||||
return object;
|
return object;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -211,8 +211,8 @@ public class PoolMapConfigs {
|
||||||
private CompanyMapStore companyMapStore;
|
private CompanyMapStore companyMapStore;
|
||||||
|
|
||||||
public MapConfig map_Company() {
|
public MapConfig map_Company() {
|
||||||
return makeDefaultMapConfig(IMDGDistributedNames.Map_Company, companyMapStore)
|
return makeDefaultMapConfig(IMDGDistributedNames.Map_Company, companyMapStore);
|
||||||
.addMapIndexConfig(makeMapIndexConfig("companyId"));
|
// .addMapIndexConfig(makeMapIndexConfig("companyId"));
|
||||||
}
|
}
|
||||||
|
|
||||||
// @Autowired
|
// @Autowired
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue