From bc78b389e721d56c7caaeba182a5d41929bf8a08 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 2 Jul 2026 17:09:39 +0300 Subject: [PATCH] NP: get rid of unused kick pipelines; LP: added lp kick pipelines with data loading --- .../config/task/PipelineStarterConfig.java | 28 +++-- .../mapper/signal/LegalsFidMapMapper.java | 4 + .../mapper/signal/LegalsMapMapper.java | 4 + .../mode/common/StartupRestoreService.java | 4 +- .../mode/common/UpdatingFidService.java | 6 +- .../service/signal/PackageRecordsService.java | 9 ++ .../impl/KickPipeLineRepeatedRestoreImpl.java | 42 ------- .../task/impl/KickPipeLineRestoreImpl.java | 42 ------- .../service/task/impl/LpKickPipeLine.java | 111 +++++++++++++++++ .../task/impl/LpKickPipelineDefaultImpl.java | 26 ++++ .../task/impl/LpKickRecalcPipeLine.java | 112 ++++++++++++++++++ ...{KickPipeLine.java => NpKickPipeLine.java} | 10 +- ...pl.java => NpKickPipeLineDefaultImpl.java} | 14 +-- ...ipeLine.java => NpKickRecalcPipeLine.java} | 14 +-- .../service/task/impl/PipelineMode.java | 5 +- .../mapper/signal/LegalsFidMapMapper.xml | 20 ++++ .../mybatis/mapper/signal/LegalsMapMapper.xml | 15 +++ .../mapper/signal/PackageRecordsMapper.xml | 13 +- 18 files changed, 349 insertions(+), 130 deletions(-) delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipeLine.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipelineDefaultImpl.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java rename src/main/java/ru/nbch/credit_tracker/service/task/impl/{KickPipeLine.java => NpKickPipeLine.java} (93%) rename src/main/java/ru/nbch/credit_tracker/service/task/impl/{KickPipeLineDefaultImpl.java => NpKickPipeLineDefaultImpl.java} (65%) rename src/main/java/ru/nbch/credit_tracker/service/task/impl/{KickRecalcPipeLine.java => NpKickRecalcPipeLine.java} (90%) 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 96f74db..34f6846 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 @@ -3,6 +3,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.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload; @@ -25,15 +26,24 @@ public class PipelineStarterConfig { public Function> pipelineStarterFactory(CreditTrackerSettings creditTrackerSettings, UserPackageMapper userPackageMapper, PackageRecordsService packageRecordsService, - PhonesMapMapper phonesMapMapper) { - // TODO separate for subject type - return (setupParam) -> switch (setupParam.mode()) { - case DEFAULT, RESEND, RESTORE -> - new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - case REPEATED_RESTORE, MAKE_XML_BY_REPEATED -> - new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - case RECALC_PERSONAL_DATA -> - new KickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings); + PhonesMapMapper phonesMapMapper, + LegalsMapMapper legalsMapMapper) { + return (setupParam) -> { + if (setupParam.subjectType() == 1) { + return switch (setupParam.mode()) { + case DEFAULT, RESEND -> + new NpKickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); + case RECALC_FID -> + new NpKickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings); + }; + } else { + return switch (setupParam.mode()) { + case DEFAULT, RESEND -> + new LpKickPipelineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); + case RECALC_FID -> + new LpKickRecalcPipeLine(setupParam, userPackageMapper, legalsMapMapper, packageRecordsService, creditTrackerSettings); + }; + } }; } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsFidMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsFidMapMapper.java index 6b952d1..18884c3 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsFidMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsFidMapMapper.java @@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.mapper.signal; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; +import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap; import java.util.List; @@ -10,4 +11,7 @@ public interface LegalsFidMapMapper { List registerJson(@Param("json") String json); + List findFullLegalFidMap(@Param("packageId") Long packageId, + @Param("limit") int limit, + @Param("offset") int offset); } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java index 4b44176..05cfab5 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java @@ -22,5 +22,9 @@ public interface LegalsMapMapper { List findEligibleForFlagsInsert(@Param("lastId") Long lastId, @Param("batchSize") Integer batchSize); + List searchedAtLessThan(@Param("daysCount") Integer daysCount, + @Param("batchSize") int batchSize, + @Param("lastId") Long lastId); + List findCachedByInnAndRegNUm(@Param("json") String json); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java index fd4ebf5..31583f9 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java @@ -64,7 +64,7 @@ public class StartupRestoreService { log.trace("Deleted {} rows", count); Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage); log.debug("Try to read and build pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath); - Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT)); + Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT)); pipelineOrchestrator.startOrDelay( () -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath), pipeline, userPackage.getId() @@ -81,7 +81,7 @@ public class StartupRestoreService { for (UserPackage userPackage : packages) { try { log.debug("Build pipeline for packageId {}", userPackage.getId()); - Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT)); + Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT)); pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); } catch (Exception e) { log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java index 0bcbf9b..6986743 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java @@ -51,7 +51,7 @@ public class UpdatingFidService { phonesMapService.deleteUnusedPhones(); log.info("Start updating fids and PD"); try { - Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA)); + Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, 1, PipelineMode.RECALC_FID)); CompletableFuture future = pipelineRecalcPDOrchestrator.startOrReject(pipeline); future.whenComplete((o, throwable) -> { try { @@ -63,7 +63,7 @@ public class UpdatingFidService { List packagesRepeated = userPackageMapper.findForRepeatedPackages(); packagesRepeated.forEach(userPackage -> { Long packageId = userPackage.getId(); - SetupParam setupParam = new SetupParam(packageId, null, PipelineMode.RESEND); + SetupParam setupParam = new SetupParam(packageId, 1, PipelineMode.RESEND); try { Pipeline makeXmlByCache = npPipelinePackageFabric.build(setupParam); pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId); @@ -82,7 +82,7 @@ public class UpdatingFidService { } } - @Scheduled(cron = "${ct.update-fid-cron-expression}") + @Scheduled(cron = "${ct.update-fid-cron-expression}") // TODO another property public void updateLp() { // TODO } diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java index 3126937..92a4258 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java @@ -8,6 +8,8 @@ import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.app.LegalsMap; import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; import ru.nbch.credit_tracker.entities.app.PhonesMap; +import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap; +import ru.nbch.credit_tracker.mapper.signal.LegalsFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; @@ -33,15 +35,18 @@ public class PackageRecordsService { private final PhonesMapMapper phonesMapMapper; private final LegalsMapMapper legalsMapMapper; private final PackagesRecordsMapper packagesRecordsMapper; + private final LegalsFidMapMapper legalsFidMapMapper; private final ObjectMapper entityToJsonMapper; public PackageRecordsService(PhonesMapMapper phonesMapMapper, LegalsMapMapper legalsMapMapper, PackagesRecordsMapper packagesRecordsMapper, + LegalsFidMapMapper legalsFidMapMapper, ObjectMapper entityToJsonMapper) { this.phonesMapMapper = phonesMapMapper; this.legalsMapMapper = legalsMapMapper; this.packagesRecordsMapper = packagesRecordsMapper; + this.legalsFidMapMapper = legalsFidMapMapper; this.entityToJsonMapper = entityToJsonMapper; } @@ -75,6 +80,10 @@ public class PackageRecordsService { return packagesRecordsMapper.findFullPhoneFidMap(packageId, limit, offset); } + public List findFullLegalFidMap(Long packageId, Integer limit, int offset) { + return legalsFidMapMapper.findFullLegalFidMap(packageId, limit, offset); + } + public void saveToCache(List phonesMaps) { try { String json = entityToJsonMapper.writeValueAsString(phonesMaps.stream() diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java deleted file mode 100644 index 366be10..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRepeatedRestoreImpl.java +++ /dev/null @@ -1,42 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import java.util.ArrayList; -import java.util.List; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.signal.PackageRecordsService; -import ru.nbch.credit_tracker.service.task.SetupParam; - -public class KickPipeLineRepeatedRestoreImpl extends KickPipeLine { - - private boolean oldRowsWereFound = false; - private final Integer daysToUpdateFlags; - public KickPipeLineRepeatedRestoreImpl(SetupParam setupParam, - UserPackageMapper userPackageMapper, - PackageRecordsService packageRecordsService, - CreditTrackerSettings creditTrackerSettings) { - super(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - this.daysToUpdateFlags = creditTrackerSettings.getDaysToUpdateFlags(); - } - - @Override - protected PipelineMode mode() { - return PipelineMode.RESTORE; - } - - @Override - protected List nextPage(Integer limit) { - List pageWithNullFids = new ArrayList<>(); - //fixme здесь должен быть запрос к таблице signals_phones_map, см. xml маппер -// if (!oldRowsWereFound) { -// pageWithNullFids = phoneFidMapMapper.pageByPackageIdAndOlderThanN(packageId, limit, offset, daysToUpdateFlags); -// } -// if (pageWithNullFids.isEmpty()) { -// oldRowsWereFound = true; -// offset = 0; -// return phoneFidMapMapper.pageByPackageIdAndNewerThanN(packageId, limit, offset, daysToUpdateFlags); -// } - return pageWithNullFids; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java deleted file mode 100644 index 1508792..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java +++ /dev/null @@ -1,42 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import java.util.ArrayList; -import java.util.List; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.service.signal.PackageRecordsService; -import ru.nbch.credit_tracker.service.task.SetupParam; - -public class KickPipeLineRestoreImpl extends KickPipeLine { - - private boolean nullFidsWereFound = false; - public KickPipeLineRestoreImpl(SetupParam setupParam, - UserPackageMapper userPackageMapper, - PackageRecordsService packageRecordsService, - CreditTrackerSettings creditTrackerSettings) { - super(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); - } - - @Override - protected PipelineMode mode() { - return PipelineMode.REPEATED_RESTORE; - } - - @Override - protected List nextPage(Integer limit) { - //todo рассмотреть к удалению - List page = new ArrayList<>(); -// if (!nullFidsWereFound) { -// page = phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset); -// } -// if (page.isEmpty()) { -// if (!nullFidsWereFound) { -// nullFidsWereFound = true; -// offset = 0; -// } -// return phoneFidMapMapper.findNotNullFidPhoneFidMap(packageId, limit, offset); -// } - return page; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipeLine.java new file mode 100644 index 0000000..58c5d7f --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipeLine.java @@ -0,0 +1,111 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap; +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.UserPackageMapper; +import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.PackageVariable; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; +import ru.nbch.pipeline.engine.core.PipelineContext; +import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter; + +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.atomic.AtomicInteger; + +public abstract class LpKickPipeLine implements IPipeLineStarter { + private final Logger log = LoggerFactory.getLogger(getClass()); + protected final Long packageId; + protected final PipelineMode mode; + protected final UserPackageMapper userPackageMapper; + protected final PackageRecordsService packageRecordsService; + protected final CreditTrackerSettings creditTrackerSettings; + private final int BATCH_SIZE; + private boolean isInterrupted = false; + protected int offset = 0; + private PipelineContext context; + + + public LpKickPipeLine(SetupParam setupParam, + UserPackageMapper userPackageMapper, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { + this.packageId = setupParam.packageId(); + this.userPackageMapper = userPackageMapper; + this.packageRecordsService = packageRecordsService; + this.creditTrackerSettings = creditTrackerSettings; + this.BATCH_SIZE = creditTrackerSettings.getLpPipeline().subjectBatchSize(); + this.mode = setupParam.mode(); + } + + private static class PipelineStarterException extends SignalClientException { + public PipelineStarterException() { + super(null, null, null); + } + } + + @SuppressWarnings("unchecked") + @Override + public void setContext(PipelineContext context) { + this.context = (PipelineContext) context; + } + + @Override + public void startFillingQueue(BlockingQueue 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()); + context.store(PackageVariable.mode, mode); + int limit = BATCH_SIZE; + UserPackage userPackage = userPackageMapper.findPackage(packageId); + List page = nextPage(limit); + while (!page.isEmpty()) { + log.debug("Get page {}", page.size()); + List subjectProfiles = page.stream().map(legalFidMap -> SubjectProfile.builder() + .mapId(legalFidMap.getId()) + .identifier(legalFidMap.getSubjectId()) + .inn(legalFidMap.getInn()) + .regNum(legalFidMap.getRegNum()) + .build()).toList(); + try { + SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles); + payload.setContext(context); + firstQueue.put(payload); + } catch (InterruptedException e) { + isInterrupted = true; + log.error("Pipeline starter was interrupted!"); + throw new PipelineStarterException(); + } + offset += limit; + page = nextPage(limit); + } + } catch (PipelineStarterException e) { + log.info("Pipeline starter shutting down..."); + } finally { + if (isInterrupted) { + //noinspection ReturnInsideFinallyBlock + return; + } + log.debug("Send poison pill"); + try { + SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); + firstQueue.put(poison); + } catch (InterruptedException e) { + log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); + } + } + } + + protected abstract PipelineMode mode(); + protected abstract List nextPage(Integer limit); +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipelineDefaultImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipelineDefaultImpl.java new file mode 100644 index 0000000..ca5c0dd --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickPipelineDefaultImpl.java @@ -0,0 +1,26 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.service.task.SetupParam; + +import java.util.List; + +public class LpKickPipelineDefaultImpl extends LpKickPipeLine{ + + public LpKickPipelineDefaultImpl(SetupParam setupParam, UserPackageMapper userPackageMapper, PackageRecordsService packageRecordsService, CreditTrackerSettings creditTrackerSettings) { + super(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); + } + + @Override + protected PipelineMode mode() { + return PipelineMode.DEFAULT; + } + + @Override + protected List nextPage(Integer limit) { + return packageRecordsService.findFullLegalFidMap(packageId, limit, offset); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java new file mode 100644 index 0000000..494c2bd --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java @@ -0,0 +1,112 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.LegalsMap; +import ru.nbch.credit_tracker.enums.PoisonType; +import ru.nbch.credit_tracker.exceptions.SignalClientException; +import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.PackageVariable; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; +import ru.nbch.pipeline.engine.core.PipelineContext; +import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter; + +import java.util.List; +import java.util.concurrent.BlockingQueue; + +public class LpKickRecalcPipeLine implements IPipeLineStarter { + private final Logger log = LoggerFactory.getLogger(getClass()); + protected final Long packageId; + protected final UserPackageMapper userPackageMapper; + protected final LegalsMapMapper legalsMapMapper; + protected final PackageRecordsService packageRecordsService; + protected final CreditTrackerSettings creditTrackerSettings; + private final int BATCH_SIZE; + private final Long daysToUpdateFids; + private boolean isInterrupted = false; + private PipelineContext context; + + + public LpKickRecalcPipeLine(SetupParam setupParam, + UserPackageMapper userPackageMapper, + LegalsMapMapper legalsMapMapper, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { + this.packageId = setupParam.packageId(); + this.userPackageMapper = userPackageMapper; + this.legalsMapMapper = legalsMapMapper; + this.packageRecordsService = packageRecordsService; + this.creditTrackerSettings = creditTrackerSettings; + this.BATCH_SIZE = creditTrackerSettings.getLpPipeline().subjectBatchSize(); + this.daysToUpdateFids = creditTrackerSettings.getDaysToUpdateFids() != null ? creditTrackerSettings.getDaysToUpdateFids() : 1; + } + + private static class PipelineStarterException extends SignalClientException { + public PipelineStarterException() { + super(null, null, null); + } + } + + @SuppressWarnings("unchecked") + @Override + public void setContext(PipelineContext context) { + this.context = (PipelineContext) context; + } + + @Override + public void startFillingQueue(BlockingQueue firstQueue) { + try { + log.info("Start kicker: mode:{};", mode()); + long lastId = 0; + List page = nextPage(lastId); + while (!page.isEmpty()) { + log.debug("Get page {}", page.size()); + List subjectProfiles = page.stream().map(legalsMap -> SubjectProfile.builder() + .mapId(legalsMap.getId()) + .inn(legalsMap.getInn()) + .regNum(legalsMap.getRegNum()) + .fid(legalsMap.getFid()) + .build()).toList(); + try { + SubjectPayload payload = new SubjectPayload(null, null, null, subjectProfiles); + payload.setContext(context); + firstQueue.put(payload); + } catch (InterruptedException e) { + isInterrupted = true; + log.error("Pipeline starter was interrupted!"); + throw new PipelineStarterException(); + } + lastId = page.getLast().getId(); + page = nextPage(lastId); + } + } catch (PipelineStarterException e) { + log.info("Pipeline starter shutting down..."); + } finally { + if (isInterrupted) { + //noinspection ReturnInsideFinallyBlock + return; + } + log.debug("Send poison pill"); + try { + SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); + firstQueue.put(poison); + } catch (InterruptedException e) { + log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); + } + } + } + + protected PipelineMode mode() { + return PipelineMode.RECALC_FID; + } + + protected List nextPage(long lastId) { + return legalsMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, lastId); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickPipeLine.java similarity index 93% rename from src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickPipeLine.java index 8ab5563..eaa2488 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickPipeLine.java @@ -21,7 +21,7 @@ import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.atomic.AtomicInteger; -public abstract class KickPipeLine implements IPipeLineStarter { +public abstract class NpKickPipeLine implements IPipeLineStarter { private final Logger log = LoggerFactory.getLogger(getClass()); protected final Long packageId; protected final PipelineMode mode; @@ -34,10 +34,10 @@ public abstract class KickPipeLine implements IPipeLineStarter { private PipelineContext context; - public KickPipeLine(SetupParam setupParam, - UserPackageMapper userPackageMapper, - PackageRecordsService packageRecordsService, - CreditTrackerSettings creditTrackerSettings) { + public NpKickPipeLine(SetupParam setupParam, + UserPackageMapper userPackageMapper, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { this.packageId = setupParam.packageId(); this.userPackageMapper = userPackageMapper; this.packageRecordsService = packageRecordsService; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickPipeLineDefaultImpl.java similarity index 65% rename from src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickPipeLineDefaultImpl.java index f4681f2..6b77d18 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickPipeLineDefaultImpl.java @@ -1,19 +1,19 @@ package ru.nbch.credit_tracker.service.task.impl; -import java.util.List; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.task.SetupParam; -public class KickPipeLineDefaultImpl extends KickPipeLine { - private boolean cachedPhonesWereFound = false; +import java.util.List; - public KickPipeLineDefaultImpl(SetupParam setupParam, - UserPackageMapper userPackageMapper, - PackageRecordsService packageRecordsService, - CreditTrackerSettings creditTrackerSettings) { +public class NpKickPipeLineDefaultImpl extends NpKickPipeLine { + + public NpKickPipeLineDefaultImpl(SetupParam setupParam, + UserPackageMapper userPackageMapper, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { super(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); } 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/NpKickRecalcPipeLine.java similarity index 90% rename from src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickRecalcPipeLine.java index 5737040..6c35b9d 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/NpKickRecalcPipeLine.java @@ -20,7 +20,7 @@ import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter; import java.util.List; import java.util.concurrent.BlockingQueue; -public class KickRecalcPipeLine implements IPipeLineStarter { +public class NpKickRecalcPipeLine implements IPipeLineStarter { private final Logger log = LoggerFactory.getLogger(getClass()); protected final Long packageId; protected final UserPackageMapper userPackageMapper; @@ -33,11 +33,11 @@ public class KickRecalcPipeLine implements IPipeLineStarter { private PipelineContext context; - public KickRecalcPipeLine(SetupParam setupParam, - UserPackageMapper userPackageMapper, - PhonesMapMapper phonesMapMapper, - PackageRecordsService packageRecordsService, - CreditTrackerSettings creditTrackerSettings) { + public NpKickRecalcPipeLine(SetupParam setupParam, + UserPackageMapper userPackageMapper, + PhonesMapMapper phonesMapMapper, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { this.packageId = setupParam.packageId(); this.userPackageMapper = userPackageMapper; this.phonesMapMapper = phonesMapMapper; @@ -102,7 +102,7 @@ public class KickRecalcPipeLine implements IPipeLineStarter { } protected PipelineMode mode() { - return PipelineMode.REPEATED_RESTORE; + return PipelineMode.RECALC_FID; } protected List nextPage(long lastId) { return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, lastId); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java index 7b7959f..5d89ce6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java @@ -8,8 +8,5 @@ package ru.nbch.credit_tracker.service.task.impl; public enum PipelineMode { DEFAULT, RESEND, - RESTORE, - REPEATED_RESTORE, - RECALC_PERSONAL_DATA, - MAKE_XML_BY_REPEATED + RECALC_FID } diff --git a/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml index 9bca8c9..effa12e 100644 --- a/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/LegalsFidMapMapper.xml @@ -5,6 +5,14 @@ + + + + + + + + + + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml index a0a5a4c..cfd28b3 100644 --- a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml @@ -126,4 +126,19 @@ AND dst.id IN (SELECT id FROM to_lock); + + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml index 0adbf9f..3f2c880 100644 --- a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml @@ -10,15 +10,13 @@ - @@ -27,8 +25,7 @@ SELECT id, package_id, subject_id, - phone_l, - error_text + phone_l FROM signals_phone_fid_map WHERE phone_l = #{phone} @@ -37,8 +34,7 @@ SELECT id, package_id, subject_id, - phone_l, - error_text + phone_l FROM signals_phone_fid_map WHERE phone_l IN @@ -114,8 +110,7 @@ SELECT id, package_id, subject_id, - phone_l, - error_text + phone_l FROM signals_phone_fid_map WHERE package_id = #{packageId} ORDER BY id