diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java index 97b975d..2e54a54 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java @@ -13,6 +13,8 @@ public class CreditTrackerApplication { ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args); UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class); MonitoringService monitoringService = ctx.getBean(MonitoringService.class); +// UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class); +// updatingFlagsService.update(); updateProcessingPackagesService.run(); monitoringService.run(); } diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java index fa434e0..d5172d9 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java @@ -5,6 +5,7 @@ import java.util.function.Supplier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload; import ru.nbch.credit_tracker.model.impl.PartFilePayload; @@ -15,6 +16,7 @@ import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineDefaultImpl; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRepeatedRestoreImpl; +import ru.nbch.credit_tracker.service.task.impl.KickRecalcPipeLine; import ru.nbch.credit_tracker.service.task.impl.NoOpTerminator; import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; @@ -25,7 +27,8 @@ public class PipelineStarterConfig { @Bean public Function> pipelineStarterFactory(CreditTrackerSettings creditTrackerSettings, UserPackageMapper userPackageMapper, - PackageRecordsService packageRecordsService) { + PackageRecordsService packageRecordsService, + PhonesMapMapper phonesMapMapper) { return (setupParam) -> switch (setupParam.mode()) { case DEFAULT -> new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); @@ -33,7 +36,8 @@ public class PipelineStarterConfig { new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); case REPEATED_RESTORE -> new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - case RECALC_PERSONAL_DATA -> null; + case RECALC_PERSONAL_DATA -> + new KickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings); }; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index 9365d12..f84f93b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -70,7 +70,7 @@ public class UpdatingFlagsService { } // @Scheduled(cron = "${ct.update-flags-cron-expression}") - private void update() { + public void update() { log.info("Start updating flags"); try { Pipeline pipeline = pipelinePackageManager.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA)); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java index 6b9a5aa..896eda6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java @@ -62,7 +62,7 @@ public class FidIdUpdater implements Handler { }) .toList(); - packageRecordsService.updateIdCache(idCacheOnUpdate); + packageRecordsService.updateIdCache(personOnProcessing); log.debug("personOnProcessing size: {}", personOnProcessing.size()); return new VoidPayload(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java index 88216cf..a507bf9 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java @@ -43,7 +43,7 @@ public class FidNameUpdater implements Handler { Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone().toString())); if (name == null) { notFoundFidPhones.add(subjectProfile.getPhone()); - } else if (subjectProfile.getFid().equals(name.getFid())) { + } else if (!subjectProfile.getFid().equals(name.getFid())) { changingFidPhones.add(subjectProfile); } }); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java index e586504..078b5f6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java @@ -2,16 +2,16 @@ package ru.nbch.credit_tracker.service.task.impl; import java.util.List; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhonesMap; -import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.enums.PoisonType; import ru.nbch.credit_tracker.exceptions.SignalClientException; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.IdData; +import ru.nbch.credit_tracker.model.NameData; import ru.nbch.credit_tracker.model.PipelineContext; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.PackageVariable; @@ -21,7 +21,7 @@ import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; -public abstract class KickRecalcPipeLine implements IPipeLineStarter { +public class KickRecalcPipeLine implements IPipeLineStarter { private final Logger log = LoggerFactory.getLogger(getClass()); protected final Long packageId; protected final UserPackageMapper userPackageMapper; @@ -31,7 +31,6 @@ public abstract class KickRecalcPipeLine implements IPipeLineStarter context; @@ -64,21 +63,28 @@ public abstract class KickRecalcPipeLine implements IPipeLineStarter firstQueue) { try { - log.info("Start kicker: mode:{}, packageId: {};", mode(), packageId); - context.store(PackageVariable.packageId, packageId); - context.store(PackageVariable.cacheMisses, new AtomicInteger()); - context.store(PackageVariable.cacheHits, new AtomicInteger()); - int limit = BATCH_SIZE; - UserPackage userPackage = userPackageMapper.findPackage(packageId); - List page = nextPage(limit); + log.info("Start kicker: mode:{};", mode()); + int offset = 0; + List page = nextPage(offset); while (!page.isEmpty()) { log.debug("Get page {}", page.size()); List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() .mapId(phoneFidMap.getId()) .phone(phoneFidMap.getPhone()) + .fid(phoneFidMap.getFid()) + .nameData(NameData.builder() + .middleName(phoneFidMap.getMiddle()) + .firstName(phoneFidMap.getFirst()) + .lastName(phoneFidMap.getName1()) + .birthDate(phoneFidMap.getBirthDt()) + .build()) + .idData(IdData.builder() + .idNum(phoneFidMap.getSerNum()) + .idType(phoneFidMap.getIdType()) + .build()) .build()).toList(); try { - SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles); + SubjectPayload payload = new SubjectPayload(null, null, null, subjectProfiles); payload.setContext(context); firstQueue.put(payload); } catch (InterruptedException e) { @@ -86,8 +92,8 @@ public abstract class KickRecalcPipeLine implements IPipeLineStarter nextPage(Integer limit) { - return phonesMapMapper.searchedAtLessThan(daysToUpdateFlags.intValue(), limit, BATCH_SIZE); + protected List nextPage(int offset) { + return phonesMapMapper.searchedAtLessThan(daysToUpdateFlags.intValue(), BATCH_SIZE, offset); } } diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index 425abf6..86cae50 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -6,6 +6,7 @@ + @@ -21,7 +22,8 @@