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

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

Browse source
This commit is contained in:
ialbert 2022-12-12 11:04:48 +03:00
parent e0dcd24604
commit 70bfd94b64
2 changed files with 43 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -27,6 +27,7 @@ public class Processor {
for (Stage currStage : pipeline) {
log.info("uuid {}. Stage: {}", task.getUuid(), currStage.getClass().getSimpleName());
result = currStage.process(task);
log.info("uuid {}. Stage {} finished", task.getUuid(), currStage.getClass().getSimpleName());
if (Arrays.asList(StageResult.ERROR, StageResult.COMPLETE).contains(result)) break;
}
long endMills = System.currentTimeMillis();

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

@ -12,8 +12,12 @@ import ru.spcex.clearing.dbf.importer.logic.data.ResultContainer;
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
import java.io.File;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@Service("dbfImporterService")
@EnableScheduling
@ -37,15 +41,45 @@ public class DBFImporterService {
}
public void run(ETable specificTable) {
log.info("performing import {}", specificTable != null ? specificTable.name() : "");
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles(specificTable);
for (Map.Entry<ETable, List<File>> newFilesEntry : newFiles.entrySet()) {
ETable currTable = newFilesEntry.getKey();
List<File> fileList = newFilesEntry.getValue();
for (File dbfFile : fileList) {
executorService.execute(() -> processor.process(ResultContainer.createNewTask(currTable, dbfFile)));
log.info("adding import task {}", specificTable != null ? specificTable.name() : "");
executorService.execute(() -> {
log.info("checking new files... {}", specificTable != null ? specificTable.name() : "");
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles(specificTable);
if (log.isDebugEnabled()) {
log.debug("following files will be processed {}", forLogging(newFiles));
} else {
log.info("following files will be processed {}", forLoggingSizeOnly(newFiles));
}
}
for (Map.Entry<ETable, List<File>> newFilesEntry : newFiles.entrySet()) {
ETable currTable = newFilesEntry.getKey();
List<File> fileList = newFilesEntry.getValue();
for (File dbfFile : fileList) {
processor.process(ResultContainer.createNewTask(currTable, dbfFile));
}
}
});
}
private static String forLogging(Map<ETable, List<File>> files) {
return files
.entrySet()
.stream()
.flatMap((Function<Map.Entry<ETable, List<File>>, Stream<String>>) entry -> {
List<String> r = new LinkedList<>();
for (File file : entry.getValue()) {
r.add(entry.getKey().name() + " " + file.toPath());
}
return r.stream();
})
.collect(Collectors.joining(";", "[", "]"));
}
private static String forLoggingSizeOnly(Map<ETable, List<File>> files) {
return files
.entrySet()
.stream()
.map(entry -> entry.getKey().name() + " " + entry.getValue().size())
.collect(Collectors.joining(";", "[", "]"));
}
}