LP: added build for fids loading
This commit is contained in:
Mikhail Trofimov
parent
e94df35c76
commit
012d7085ad
6 changed files with 105 additions and 28 deletions
|
|
@ -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()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,25 +8,47 @@ import ru.nbch.credit_tracker.model.impl.VoidPayload;
|
|||
|
||||
@Configuration
|
||||
public class StageConfig {
|
||||
// NP
|
||||
|
||||
@Bean("nameEnricherQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> nameEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
@Bean("npFidEnricherQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> npFidEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
|
||||
}
|
||||
|
||||
@Bean("nameUpdaterQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
@Bean("npFidUpdaterQueueFactoryUpdater")
|
||||
public StageQueueFactory<SubjectPayload> npFidUpdaterQueueFactoryUpdater(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameUpdaterQueueCapacity());
|
||||
}
|
||||
|
||||
@Bean("noOpQueueFactory")
|
||||
public StageQueueFactory<VoidPayload> noOpQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
@Bean("npNoOpQueueFactory")
|
||||
public StageQueueFactory<VoidPayload> npNoOpQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
|
||||
}
|
||||
|
||||
@Bean("partFileQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
@Bean("npFinishSetupQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> npFinishSetupQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
|
||||
}
|
||||
|
||||
// LP
|
||||
|
||||
@Bean("lpFidEnricherQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> lpFidEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().nameEnricherQueueCapacity());
|
||||
}
|
||||
|
||||
@Bean("lpFidUpdaterQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> lpFidUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().nameUpdaterQueueCapacity());
|
||||
}
|
||||
|
||||
@Bean("lpNoOpQueueFactory")
|
||||
public StageQueueFactory<VoidPayload> lpNoOpQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().mergerQueueCapacity());
|
||||
}
|
||||
|
||||
@Bean("lpFinishSetupQueueFactory")
|
||||
public StageQueueFactory<SubjectPayload> lpFinishSetupQueueFactory(CreditTrackerSettings creditTrackerSettings) {
|
||||
return new StageQueueFactory<>(creditTrackerSettings.getLpPipeline().mergerQueueCapacity());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<SubjectPayload> lpFidEnricherQueueFactory;
|
||||
private final StageQueueFactory<SubjectPayload> lpFidUpdaterQueueFactory;
|
||||
private final StageQueueFactory<SubjectPayload> lpFinishSetupQueueFactory;
|
||||
private final StageQueueFactory<VoidPayload> noOpQueueFactory;
|
||||
private final LpFidEnricher lpFidEnricher;
|
||||
private final Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory;
|
||||
private final Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory;
|
||||
private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
|
||||
|
||||
public LpPipelinePackageFabric(CreditTrackerSettings creditTrackerSettings) {
|
||||
private final CreditTrackerSettings creditTrackerSettings;
|
||||
private final PipelineBackoff pipelineBackoff;
|
||||
|
||||
public LpPipelinePackageFabric(@Qualifier("lpFidEnricherQueueFactory") StageQueueFactory<SubjectPayload> lpFidEnricherQueueFactory,
|
||||
@Qualifier("lpFidUpdaterQueueFactory") StageQueueFactory<SubjectPayload> lpFidUpdaterQueueFactory,
|
||||
@Qualifier("lpFinishSetupQueueFactory") StageQueueFactory<SubjectPayload> lpFinishSetupQueueFactory,
|
||||
@Qualifier("lpNoOpQueueFactory") StageQueueFactory<VoidPayload> lpNoOpQueueFactory,
|
||||
LpFidEnricher lpFidEnricher,
|
||||
Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory,
|
||||
Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory,
|
||||
Function<SetupParam, IPipeLineStarter<SubjectPayload>> 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
|
||||
|
|
|
|||
|
|
@ -21,9 +21,9 @@ import java.util.function.Supplier;
|
|||
@Component
|
||||
@Slf4j
|
||||
public class NpPipelinePackageFabric implements InitializingBean {
|
||||
private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory;
|
||||
private final StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory;
|
||||
private final StageQueueFactory<SubjectPayload> sendSignalQueueFactory;
|
||||
private final StageQueueFactory<SubjectPayload> npFidEnricherQueueFactory;
|
||||
private final StageQueueFactory<SubjectPayload> npFidUpdaterQueueFactoryUpdater;
|
||||
private final StageQueueFactory<SubjectPayload> npFinishSetupQueueFactory;
|
||||
private final StageQueueFactory<VoidPayload> 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<SubjectPayload> nameEnricherQueueFactory,
|
||||
@Qualifier("nameUpdaterQueueFactory") StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory,
|
||||
@Qualifier("partFileQueueFactory") StageQueueFactory<SubjectPayload> sendSignalQueueFactory,
|
||||
@Qualifier("noOpQueueFactory") StageQueueFactory<VoidPayload> noOpQueueFactory,
|
||||
public NpPipelinePackageFabric(@Qualifier("npFidEnricherQueueFactory") StageQueueFactory<SubjectPayload> npFidEnricherQueueFactory,
|
||||
@Qualifier("npFidUpdaterQueueFactoryUpdater") StageQueueFactory<SubjectPayload> npFidUpdaterQueueFactoryUpdater,
|
||||
@Qualifier("npFinishSetupQueueFactory") StageQueueFactory<SubjectPayload> npFinishSetupQueueFactory,
|
||||
@Qualifier("npNoOpQueueFactory") StageQueueFactory<VoidPayload> noOpQueueFactory,
|
||||
NpFidUpdater npFidUpdater,
|
||||
Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory,
|
||||
Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory,
|
||||
Function<SetupParam, IPipeLineStarter<SubjectPayload>> 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))
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
}
|
||||
|
|
|
|||
|
|
@ -98,6 +98,7 @@ public class FlagsEnricher {
|
|||
}).toList();
|
||||
|
||||
packageRecordsService.updateNpFlagCache(flagCacheOnUpdate);
|
||||
packageRecordsService.updateLpFlagCache(lpFlagCacheOnUpdate);
|
||||
}
|
||||
|
||||
public void accept(List<PhonesMap> phonesMaps, List<LegalsMap> legalsMaps) {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue