diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java index 368087d..872a783 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java @@ -2,7 +2,7 @@ package ru.nbch.credit_tracker.config.task; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import ru.nbch.credit_tracker.model.SendSignalsPayload; +import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; @Configuration @@ -18,8 +18,8 @@ public class StageConfig { return new StageQueueFactory<>(); } - @Bean("signalsQueueFactory") - public StageQueueFactory sendSignalsQueueFactory() { + @Bean("partFileQueueFactory") + public StageQueueFactory partFileQueueFactory() { return new StageQueueFactory<>(10); } diff --git a/src/main/java/ru/nbch/credit_tracker/model/PartFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/PartFilePayload.java deleted file mode 100644 index 9e7c498..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/PartFilePayload.java +++ /dev/null @@ -1,25 +0,0 @@ -package ru.nbch.credit_tracker.model; - -import ru.nbch.credit_tracker.entities.external_request.signals.Signals; - -import java.nio.file.Path; - -public class PartFilePayload { - private final Path partPath; - private final Long packageId; - private final Signals signals; - - public PartFilePayload(Path partPath, Long packageId, Signals signals) { - this.partPath = partPath; - this.packageId = packageId; - this.signals = signals; - } - - public Path getPartPath() { - return partPath; - } - - public Long getPackageId() { - return packageId; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java new file mode 100644 index 0000000..4897a42 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java @@ -0,0 +1,30 @@ +package ru.nbch.credit_tracker.model.impl; + +import java.nio.file.Path; +import lombok.Getter; +import ru.nbch.credit_tracker.model.StagePayload; + +@Getter +public class PartFilePayload implements StagePayload { + private final Path partPath; + private final Long packageId; + private final boolean poison; + public static final PartFilePayload POISON = new PartFilePayload(true); + + public PartFilePayload(boolean poison) { + this.poison = poison; + this.partPath = null; + this.packageId = null; + } + + public PartFilePayload(Path partPath, Long packageId) { + this.partPath = partPath; + this.packageId = packageId; + this.poison = false; + } + + @Override + public boolean isPoison() { + return false; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java index 7cdef61..5997ed3 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java @@ -14,6 +14,7 @@ import java.sql.SQLException; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.*; +import ru.nbch.credit_tracker.model.SubjectProfile; @Slf4j @Service @@ -40,20 +41,20 @@ public class FlagsService2 { } // TODO optimize - public void fillFlagsData(List subjects) { - List fids = subjects.stream().map(PhoneFidMap::getFid).filter(Objects::nonNull).toList(); + public void fillFlagsData(List subjects) { + List fids = subjects.stream().map(SubjectProfile::getFid).filter(Objects::nonNull).toList(); Map fidsflagsDataMap = findFlags(fids); fidsflagsDataMap.forEach((fid, data) -> { subjects.stream().filter(s-> s.getFid() != null && s.getFid().equals(fid)).forEach(phoneFidMap -> { - phoneFidMap.setFlagDebt5000(data.getFlagDebt5000()); - phoneFidMap.setFlag90days(data.getFlag90days()); - phoneFidMap.setCalculatedAt(LocalDateTime.now()); +// phoneFidMap.setFlagDebt5000(data.getFlagDebt5000()); +// phoneFidMap.setFlag90days(data.getFlag90days()); +// phoneFidMap.setCalculatedAt(LocalDateTime.now()); }); }); subjects.stream().filter(s-> s.getFid() == null).forEach(phoneFidMap -> { - phoneFidMap.setCalculatedAt(LocalDateTime.now()); +// phoneFidMap.setCalculatedAt(LocalDateTime.now()); }); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java index 0471ea6..bf9ae73 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java @@ -10,10 +10,10 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.task.StageQueueFactory; -import ru.nbch.credit_tracker.model.SendSignalsPayload; +import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler; -import ru.nbch.credit_tracker.service.task.impl.dprctd.FidIdEnricher; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @@ -26,7 +26,7 @@ public class PipelinePackageManager { private final StaxXmlProcessor xmlProcessor; private final StageQueueFactory saveSubjectQueueFactory; private final StageQueueFactory idEnricherQueueFactory; - private final StageQueueFactory sendSignalQueueFactory; + private final StageQueueFactory sendSignalQueueFactory; private final SaveSubjectHandler saveSubjectHandler; private final FidIdEnricher fidIdEnricherHandler; private final BiFunction> pipelineStarterFactory; @@ -39,7 +39,7 @@ public class PipelinePackageManager { StaxXmlProcessor xmlProcessor, @Qualifier("saveSubjectQueueFactory") StageQueueFactory saveSubjectQueueFactory, @Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, - @Qualifier("signalsQueueFactory") StageQueueFactory sendSignalQueueFactory, + @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, BiFunction> pipelineStarterFactory) { this.applicationContext = applicationContext; @@ -65,7 +65,7 @@ public class PipelinePackageManager { .andThen() .stage(sendSignalQueueFactory.getObject()) .handler(fidIdEnricherHandler) - .poison(SendSignalsPayload.POISON) + .poison(PartFilePayload.POISON) .andThen() .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)); executorService.submit(pipeline.getStarter()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java index fbc2991..2ec31a1 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java @@ -23,12 +23,15 @@ import javax.xml.stream.XMLStreamWriter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.springframework.beans.factory.InitializingBean; +import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.model.PartFilePayload; +import ru.nbch.credit_tracker.model.impl.PartFilePayload; @Slf4j -//@Service -public final class XmlMergeCoordinator implements InitializingBean { +@Service +public final class XmlMergeCoordinator implements InitializingBean +// implements IPipeLineTerminator +{ private final BlockingQueue mergerQueue; private final Path mergedDir; private final Charset encoding = Charset.forName("windows-1251"); @@ -155,4 +158,9 @@ public final class XmlMergeCoordinator implements InitializingBean { public void afterPropertiesSet() throws Exception { mergerExec.submit(this::runLoop); } +// +// @Override +// public void startPollingQueue(BlockingQueue queue) { +// +// } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java similarity index 85% rename from src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidIdEnricher.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java index 5faabc6..b55949b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidIdEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.service.task.impl.dprctd; +package ru.nbch.credit_tracker.service.task.impl; import java.io.IOException; import java.io.OutputStream; @@ -14,7 +14,7 @@ import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.builder.PersonFactoryBuilderV2; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.external_request.signals.Person; -import ru.nbch.credit_tracker.model.SendSignalsPayload; +import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.indic.PersonServiceV2; @@ -23,9 +23,10 @@ import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2; @Service @Slf4j -public class FidIdEnricher implements Handler { +public class FidIdEnricher implements Handler { private final PersonServiceV2 personService; private final CreditTrackerSettings config; + public FidIdEnricher(PersonServiceV2 personService, CreditTrackerSettings config) { this.personService = personService; @@ -33,7 +34,7 @@ public class FidIdEnricher implements Handler subjectsProfile = taskPayload.getSubjects(); log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); personService.filledPassportsData(subjectsProfile); @@ -42,11 +43,12 @@ public class FidIdEnricher implements Handler PersonFactoryBuilderV2.builder().personData(subjectProfile).build()) .toList(); log.debug("personOnProcessing size: {}", personOnProcessing.size()); - writeSignals(personOnProcessing, taskPayload.getPackageId()); - return new SendSignalsPayload(false); + Long packageId = taskPayload.getPackageId(); + Path partPath = writeSignals(personOnProcessing, packageId); + return new PartFilePayload(partPath, packageId); } - private void writeSignals(List persons, Long packageId){ + private Path writeSignals(List persons, Long packageId) { Path dir = Paths.get(config.getLoadDir(), "tmp_folder"); try { Files.createDirectories(dir); @@ -70,6 +72,7 @@ public class FidIdEnricher implements Handler { +public class FlagsEnricher implements Handler { private final FlagsService2 flagsService; private final PhoneFidMapMapper mapper; @@ -25,21 +24,22 @@ public class FlagsEnricher implements Handler { } @Override - public DummyPayload handle(FlagsPayload taskPayload) { - List phoneFidMaps = taskPayload.getPhoneFidMaps(); + public DummyPayload handle(SubjectPayload taskPayload) { + List phoneFidMaps = taskPayload.getSubjects(); log.debug("FlagsEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMaps.size()); -// flagsService.fillFlagsData(phoneFidMaps); + flagsService.fillFlagsData(phoneFidMaps); - log.debug("Start updating subject's batch"); - if (!phoneFidMaps.isEmpty()) { - int updateBatchSize = 7000; - for (int j = 0; j < phoneFidMaps.size(); j += updateBatchSize) { - int endIndex = Math.min(j + updateBatchSize, phoneFidMaps.size()); - mapper.updatePhoneFidMaps(phoneFidMaps.subList(j, endIndex)); - log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, phoneFidMaps.size(), taskPayload.getPackageId()); - } - } - log.debug("Complete updating subject's batch"); + //todo по какому ключу апдейтить записи, потому что id у нас тут нет, по subject_id? или все же сохранять id при вставке в SaveSubjectHandler +// log.debug("Start updating subject's batch"); +// if (!phoneFidMaps.isEmpty()) { +// int updateBatchSize = 7000; +// for (int j = 0; j < phoneFidMaps.size(); j += updateBatchSize) { +// int endIndex = Math.min(j + updateBatchSize, phoneFidMaps.size()); +// mapper.updatePhoneFidMaps(phoneFidMaps.subList(j, endIndex)); +// log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, phoneFidMaps.size(), taskPayload.getPackageId()); +// } +// } +// log.debug("Complete updating subject's batch"); return new DummyPayload(false); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java new file mode 100644 index 0000000..f548630 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java @@ -0,0 +1,8 @@ +package ru.nbch.credit_tracker.service.task.setup; + +import java.util.concurrent.BlockingQueue; +import ru.nbch.credit_tracker.model.StagePayload; + +public interface IPipeLineTerminator { + void startPollingQueue(BlockingQueue queue); +}