diff --git a/credit-tracker.properties b/credit-tracker.properties index 36d5e9d..1f10554 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -22,9 +22,9 @@ ct.indic-data-source-settings.leak-detection-threshold=${LEAK_DETECTION_THRESHOL # signal ct.signal-data-source-settings.driver=org.postgresql.Driver -ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.230.227.125:5432/credit_tracker?currentSchema=test} +ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.200.200.133:5432/credit_tracker?currentSchema=test_3} ct.signal-data-source-settings.username=${USERNAME_SIGNAL:credit_tracker_user} -ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:password} +ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:-} ct.signal-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SIGNAL:40} ct.signal-data-source-settings.minimum-idle=${MINIMUM_IDLE_SIGNAL:10} ct.signal-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_SIGNAL:10000} diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java index 3d7b2f9..ba07179 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java @@ -17,6 +17,8 @@ public class SubjectPayload implements StagePayload { public static final SubjectPayload POISON = new SubjectPayload(null, PoisonType.NORMALLY); private final PoisonType poison; + private boolean hasFids = false; + public SubjectPayload(Long packageId, PoisonType poison) { this.memberCode = null; this.packageName = null; @@ -59,6 +61,14 @@ public class SubjectPayload implements StagePayload { return packageName; } + public boolean isHasFids() { + return hasFids; + } + + public void setHasFids(boolean hasFids) { + this.hasFids = hasFids; + } + @Override public boolean isPoison() { return poison == PoisonType.NORMALLY; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java index 1992010..0944163 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java @@ -47,11 +47,12 @@ public class LpFidEnricher implements Handler { log.debug("LpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size()); //найдем inn + regNum в табличном кэше Map, LegalsMap> cachedRecords = packageRecordsService.findCachedLegalsByInnAndRegNums(subjects); - processByNumber(subjects, cachedRecords, ctx); + boolean hasFids = processByNumber(subjects, cachedRecords, ctx); + taskPayload.setHasFids(hasFids); return taskPayload; } - private void processByNumber(List subjects, + private boolean processByNumber(List subjects, Map, LegalsMap> cachedRecords, PipelineContext context) { List noFidsData = subjects.stream() @@ -125,7 +126,10 @@ public class LpFidEnricher implements Handler { .filter(legalsMap -> seen.add(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()))) .toList(); + packageRecordsService.saveLegalsToCache(subjectForSave); + + return subjectForSave.stream().anyMatch(s -> s.getFid() != null); } private void fillByCache(SubjectProfile subjectProfile, LegalsMap cachedRecord) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java index 9354aad..20ba280 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java @@ -10,14 +10,12 @@ import java.util.concurrent.ExecutorService; @Service public class LpPipelinePackageOrchestrator extends PackageOrchestrator { - private final CreditTrackerSettings creditTrackerSettings; public LpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings, UserPackageMapper userPackageMapper, @Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService, @Qualifier("lpSaverExecutorService") ExecutorService lpSaverExecutorService) { - super(userPackageMapper, lpPipelineExecutorService, lpSaverExecutorService); - this.creditTrackerSettings = creditTrackerSettings; + super(creditTrackerSettings, userPackageMapper, lpPipelineExecutorService, lpSaverExecutorService); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java index 419fc51..f098c93 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java @@ -8,14 +8,11 @@ import ru.nbch.credit_tracker.config.settings.PipelineSettings; import java.util.concurrent.ExecutorService; @Service -public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator{ - - private final CreditTrackerSettings creditTrackerSettings; +public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator { public LpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings, @Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService) { - super(lpPipelineExecutorService); - this.creditTrackerSettings = creditTrackerSettings; + super(creditTrackerSettings, lpPipelineExecutorService); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java index 0387a59..48706eb 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java @@ -10,14 +10,12 @@ import java.util.concurrent.ExecutorService; @Service public class NpPipelinePackageOrchestrator extends PackageOrchestrator { - private final CreditTrackerSettings creditTrackerSettings; public NpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings, UserPackageMapper userPackageMapper, @Qualifier("npPipelineExecutorService") ExecutorService npPipelineExecutorService, @Qualifier("npSaverExecutorService") ExecutorService npSaverExecutorService) { - super(userPackageMapper, npPipelineExecutorService, npSaverExecutorService); - this.creditTrackerSettings = creditTrackerSettings; + super(creditTrackerSettings, userPackageMapper, npPipelineExecutorService, npSaverExecutorService); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java index d55bbf1..b938733 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java @@ -10,11 +10,9 @@ import java.util.concurrent.ExecutorService; @Service public class NpPipelineRecalcOrchestrator extends RecalcOrchestrator { - private final CreditTrackerSettings creditTrackerSettings; public NpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings, @Qualifier("npPipelineExecutorService")ExecutorService npPipelineExecutorService) { - super(npPipelineExecutorService); - this.creditTrackerSettings = creditTrackerSettings; + super(creditTrackerSettings, npPipelineExecutorService); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PackageOrchestrator.java index da16fb5..4f24844 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PackageOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PackageOrchestrator.java @@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.service.task; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.settings.PipelineSettings; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.PackageVariable; @@ -20,6 +21,7 @@ import java.util.concurrent.atomic.AtomicInteger; public abstract class PackageOrchestrator { private final Logger log = LoggerFactory.getLogger(getClass()); + protected final CreditTrackerSettings creditTrackerSettings; private final UserPackageMapper userPackageMapper; private final ExecutorService pipelineExecutorService; private final ExecutorService saverExecutorService; @@ -27,9 +29,11 @@ public abstract class PackageOrchestrator { private final java.util.Deque pending = new java.util.ArrayDeque<>(); private final Semaphore semaphore; - public PackageOrchestrator(UserPackageMapper userPackageMapper, + public PackageOrchestrator(CreditTrackerSettings creditTrackerSettings, + UserPackageMapper userPackageMapper, ExecutorService pipelineExecutorService, ExecutorService saverExecutorService) { + this.creditTrackerSettings = creditTrackerSettings; this.userPackageMapper = userPackageMapper; this.pipelineExecutorService = pipelineExecutorService; this.saverExecutorService = saverExecutorService; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/RecalcOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/RecalcOrchestrator.java index 8881082..b3a1d63 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/RecalcOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/RecalcOrchestrator.java @@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.service.task; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.settings.PipelineSettings; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; import ru.nbch.pipeline.engine.core.Pipeline; @@ -13,11 +14,14 @@ import java.util.concurrent.TimeUnit; public abstract class RecalcOrchestrator { private final Logger log = LoggerFactory.getLogger(getClass()); + protected final CreditTrackerSettings creditTrackerSettings; private final ExecutorService pipelineExecutorService; private final Object lock = new Object(); private final Semaphore semaphore; - public RecalcOrchestrator(ExecutorService pipelineExecutorService) { + public RecalcOrchestrator(CreditTrackerSettings creditTrackerSettings, + ExecutorService pipelineExecutorService) { + this.creditTrackerSettings = creditTrackerSettings; this.pipelineExecutorService = pipelineExecutorService; this.semaphore = new Semaphore(1); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java index 91a7cb1..011c9b2 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java @@ -46,11 +46,12 @@ public class NpFidEnricher implements Handler { log.debug("NpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size()); //найдем телефоны в табличном кэше Map cachedRecords = packageRecordsService.findCachedPhones(subjects); - processByNumber(subjects, cachedRecords, ctx); + boolean hasFids = processByNumber(subjects, cachedRecords, ctx); + taskPayload.setHasFids(hasFids); return taskPayload; } - private void processByNumber(List subjects, + private boolean processByNumber(List subjects, Map cachedRecords, PipelineContext context) { List noFidsData = subjects.stream() @@ -97,6 +98,8 @@ public class NpFidEnricher implements Handler { .toList(); packageRecordsService.saveToCache(subjectForSave); + + return subjectForSave.stream().anyMatch(s -> s.getFid() != null); } private void fillByCache(SubjectProfile subjectProfile, PhonesMap cachedRecord) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java index 46a57cd..adfc3bb 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SetupMonitoringPipelineTerminator.java @@ -34,7 +34,7 @@ public final class SetupMonitoringPipelineTerminator implements IPipeLineTermina continue; } if (!part.isPoison() && !hasFids.get()) { - hasFids.set(part.getSubjects().stream().anyMatch(s -> s.getFid() != null)); + hasFids.set(part.isHasFids()); } if (part.isPoison()) { log.debug("Got poison");