From 012d7085ad832009a248f0c53377941e17c7b156 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 2 Jul 2026 16:11:37 +0300 Subject: [PATCH] LP: added build for fids loading --- .../config/task/PipelineStarterConfig.java | 12 ++++- .../config/task/StageConfig.java | 38 +++++++++++--- .../service/task/LpPipelinePackageFabric.java | 51 +++++++++++++++++-- .../service/task/NpPipelinePackageFabric.java | 28 +++++----- .../service/task/SetupParam.java | 3 +- .../service/task/impl/FlagsEnricher.java | 1 + 6 files changed, 105 insertions(+), 28 deletions(-) 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 a96126b..96f74db 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 @@ -26,6 +26,7 @@ public class PipelineStarterConfig { 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); @@ -47,11 +48,20 @@ public class PipelineStarterConfig { } @Bean - public PipelineBackoff pipelineBackoff(CreditTrackerSettings creditTrackerSettings) { + public PipelineBackoff npPipelineBackoff(CreditTrackerSettings creditTrackerSettings) { return new PipelineBackoff( creditTrackerSettings.getPipeline().errorAttemptCount(), creditTrackerSettings.getPipeline().errorAttemptDelay(), creditTrackerSettings.getPipeline().retryableExceptions() ); } + + @Bean + public PipelineBackoff lpPipelineBackoff(CreditTrackerSettings creditTrackerSettings) { + return new PipelineBackoff( + creditTrackerSettings.getLpPipeline().errorAttemptCount(), + creditTrackerSettings.getLpPipeline().errorAttemptDelay(), + creditTrackerSettings.getLpPipeline().retryableExceptions() + ); + } } 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 b8844f6..e980133 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 @@ -8,25 +8,47 @@ import ru.nbch.credit_tracker.model.impl.VoidPayload; @Configuration public class StageConfig { + // NP - @Bean("nameEnricherQueueFactory") - public StageQueueFactory nameEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) { + @Bean("npFidEnricherQueueFactory") + public StageQueueFactory npFidEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); } - @Bean("nameUpdaterQueueFactory") - public StageQueueFactory nameUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) { + @Bean("npFidUpdaterQueueFactoryUpdater") + public StageQueueFactory npFidUpdaterQueueFactoryUpdater(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity()); } - @Bean("noOpQueueFactory") - public StageQueueFactory noOpQueueFactory(CreditTrackerSettings creditTrackerSettings) { + @Bean("npNoOpQueueFactory") + public StageQueueFactory npNoOpQueueFactory(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); } - @Bean("partFileQueueFactory") - public StageQueueFactory partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) { + @Bean("npFinishSetupQueueFactory") + public StageQueueFactory npFinishSetupQueueFactory(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); } + // LP + + @Bean("lpFidEnricherQueueFactory") + public StageQueueFactory lpFidEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().nameEnricherQueueCapacity()); + } + + @Bean("lpFidUpdaterQueueFactory") + public StageQueueFactory lpFidUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().nameUpdaterQueueCapacity()); + } + + @Bean("lpNoOpQueueFactory") + public StageQueueFactory lpNoOpQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().mergerQueueCapacity()); + } + + @Bean("lpFinishSetupQueueFactory") + public StageQueueFactory lpFinishSetupQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().mergerQueueCapacity()); + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java index 004f3ca..6ddbbcb 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java @@ -2,22 +2,67 @@ package ru.nbch.credit_tracker.service.task; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.config.task.StageQueueFactory; +import ru.nbch.credit_tracker.model.impl.*; import ru.nbch.pipeline.engine.core.Pipeline; +import ru.nbch.pipeline.engine.core.PipelineBackoff; +import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter; +import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator; +import ru.nbch.pipeline.engine.core.setup.PipeLineBuilderV2; + +import java.util.function.Function; +import java.util.function.Supplier; @Slf4j @Component public class LpPipelinePackageFabric implements InitializingBean { - private final CreditTrackerSettings creditTrackerSettings; + private final StageQueueFactory lpFidEnricherQueueFactory; + private final StageQueueFactory lpFidUpdaterQueueFactory; + private final StageQueueFactory lpFinishSetupQueueFactory; + private final StageQueueFactory noOpQueueFactory; + private final LpFidEnricher lpFidEnricher; + private final Supplier> pipelineTerminatorFactory; + private final Supplier> noOpTerminatorFactory; + private final Function> pipelineStarterFactory; - public LpPipelinePackageFabric(CreditTrackerSettings creditTrackerSettings) { + private final CreditTrackerSettings creditTrackerSettings; + private final PipelineBackoff pipelineBackoff; + + public LpPipelinePackageFabric(@Qualifier("lpFidEnricherQueueFactory") StageQueueFactory lpFidEnricherQueueFactory, + @Qualifier("lpFidUpdaterQueueFactory") StageQueueFactory lpFidUpdaterQueueFactory, + @Qualifier("lpFinishSetupQueueFactory") StageQueueFactory lpFinishSetupQueueFactory, + @Qualifier("lpNoOpQueueFactory") StageQueueFactory lpNoOpQueueFactory, + LpFidEnricher lpFidEnricher, + Supplier> pipelineTerminatorFactory, + Supplier> noOpTerminatorFactory, + Function> pipelineStarterFactory, + CreditTrackerSettings creditTrackerSettings, + PipelineBackoff lpPipelineBackoff) { + this.lpFidEnricherQueueFactory = lpFidEnricherQueueFactory; + this.lpFidUpdaterQueueFactory = lpFidUpdaterQueueFactory; + this.lpFinishSetupQueueFactory = lpFinishSetupQueueFactory; + this.noOpQueueFactory = lpNoOpQueueFactory; + this.lpFidEnricher = lpFidEnricher; + this.pipelineTerminatorFactory = pipelineTerminatorFactory; + this.noOpTerminatorFactory = noOpTerminatorFactory; + this.pipelineStarterFactory = pipelineStarterFactory; this.creditTrackerSettings = creditTrackerSettings; + this.pipelineBackoff = lpPipelineBackoff; } public Pipeline build(SetupParam setupParam) throws Exception { - return null; + return PipeLineBuilderV2.first(lpFidEnricherQueueFactory.getObject(), lpFinishSetupQueueFactory.getObject()) + .handler(lpFidEnricher) + .andThen() + .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) + .pollFinishingQueue(pipelineTerminatorFactory.get()) + .backoff(pipelineBackoff) + .context(PackageVariable.class) + .build(); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageFabric.java index a22ed4d..6dc22ae 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageFabric.java @@ -21,9 +21,9 @@ import java.util.function.Supplier; @Component @Slf4j public class NpPipelinePackageFabric implements InitializingBean { - private final StageQueueFactory nameEnricherQueueFactory; - private final StageQueueFactory nameUpdaterQueueFactory; - private final StageQueueFactory sendSignalQueueFactory; + private final StageQueueFactory npFidEnricherQueueFactory; + private final StageQueueFactory npFidUpdaterQueueFactoryUpdater; + private final StageQueueFactory npFinishSetupQueueFactory; private final StageQueueFactory noOpQueueFactory; private final NpFidEnricher npFidEnricher; private final NpFidUpdater npFidUpdater; @@ -33,20 +33,20 @@ public class NpPipelinePackageFabric implements InitializingBean { private final CreditTrackerSettings creditTrackerSettings; private final PipelineBackoff pipelineBackoff; - public NpPipelinePackageFabric(@Qualifier("nameEnricherQueueFactory") StageQueueFactory nameEnricherQueueFactory, - @Qualifier("nameUpdaterQueueFactory") StageQueueFactory nameUpdaterQueueFactory, - @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, - @Qualifier("noOpQueueFactory") StageQueueFactory noOpQueueFactory, + public NpPipelinePackageFabric(@Qualifier("npFidEnricherQueueFactory") StageQueueFactory npFidEnricherQueueFactory, + @Qualifier("npFidUpdaterQueueFactoryUpdater") StageQueueFactory npFidUpdaterQueueFactoryUpdater, + @Qualifier("npFinishSetupQueueFactory") StageQueueFactory npFinishSetupQueueFactory, + @Qualifier("npNoOpQueueFactory") StageQueueFactory noOpQueueFactory, NpFidUpdater npFidUpdater, Supplier> pipelineTerminatorFactory, Supplier> noOpTerminatorFactory, Function> pipelineStarterFactory, NpFidEnricher npFidEnricher, CreditTrackerSettings creditTrackerSettings, - PipelineBackoff pipelineBackoff) { - this.nameEnricherQueueFactory = nameEnricherQueueFactory; - this.nameUpdaterQueueFactory = nameUpdaterQueueFactory; - this.sendSignalQueueFactory = sendSignalQueueFactory; + PipelineBackoff npPipelineBackoff) { + this.npFidEnricherQueueFactory = npFidEnricherQueueFactory; + this.npFidUpdaterQueueFactoryUpdater = npFidUpdaterQueueFactoryUpdater; + this.npFinishSetupQueueFactory = npFinishSetupQueueFactory; this.noOpQueueFactory = noOpQueueFactory; this.npFidUpdater = npFidUpdater; this.pipelineTerminatorFactory = pipelineTerminatorFactory; @@ -54,11 +54,11 @@ public class NpPipelinePackageFabric implements InitializingBean { this.pipelineStarterFactory = pipelineStarterFactory; this.npFidEnricher = npFidEnricher; this.creditTrackerSettings = creditTrackerSettings; - this.pipelineBackoff = pipelineBackoff; + this.pipelineBackoff = npPipelineBackoff; } public Pipeline build(SetupParam setupParam) throws Exception { - return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), sendSignalQueueFactory.getObject()) + return PipeLineBuilderV2.first(npFidEnricherQueueFactory.getObject(), npFinishSetupQueueFactory.getObject()) .handler(npFidEnricher) .andThen() .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) @@ -69,7 +69,7 @@ public class NpPipelinePackageFabric implements InitializingBean { } public Pipeline buildRecalculationPD(SetupParam setupParam) throws Exception { - return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), noOpQueueFactory.getObject()) + return PipeLineBuilderV2.first(npFidUpdaterQueueFactoryUpdater.getObject(), noOpQueueFactory.getObject()) .handler(npFidUpdater) .andThen() .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/SetupParam.java b/src/main/java/ru/nbch/credit_tracker/service/task/SetupParam.java index fe2dc38..2e7dc6a 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/SetupParam.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/SetupParam.java @@ -1,10 +1,9 @@ package ru.nbch.credit_tracker.service.task; -import java.nio.file.Path; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; public record SetupParam( Long packageId, - Path xmlReqPath, + int subjectType, PipelineMode mode) { } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java index 0bbf704..2c8ed1f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java @@ -98,6 +98,7 @@ public class FlagsEnricher { }).toList(); packageRecordsService.updateNpFlagCache(flagCacheOnUpdate); + packageRecordsService.updateLpFlagCache(lpFlagCacheOnUpdate); } public void accept(List phonesMaps, List legalsMaps) {