Fix bean creating; fix has fid logic (NO_FID status)

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}

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;

src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java

@@ -47,11 +47,12 @@ public class LpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
         log.debug("LpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size());
         //найдем inn + regNum в табличном кэше
         Map<Pair<Long, Long>, LegalsMap> cachedRecords = packageRecordsService.findCachedLegalsByInnAndRegNums(subjects);
-        processByNumber(subjects, cachedRecords, ctx);
+        boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
+        taskPayload.setHasFids(hasFids);
         return taskPayload;
     }
 
-    private void processByNumber(List<SubjectProfile> subjects,
+    private boolean processByNumber(List<SubjectProfile> subjects,
                                  Map<Pair<Long, Long>, LegalsMap> cachedRecords,
                                  PipelineContext<PackageVariable> context) {
         List<SubjectProfile> noFidsData = subjects.stream()
@@ -125,7 +126,10 @@ public class LpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
             .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) {

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);
+        super(creditTrackerSettings, userPackageMapper, lpPipelineExecutorService, lpSaverExecutorService);
-        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java

@@ -10,12 +10,9 @@ import java.util.concurrent.ExecutorService;
 @Service
 public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator {
 
-    private final CreditTrackerSettings creditTrackerSettings;
-
     public LpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
                                         @Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService) {
-        super(lpPipelineExecutorService);
+        super(creditTrackerSettings, lpPipelineExecutorService);
-        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

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);
+        super(creditTrackerSettings, userPackageMapper, npPipelineExecutorService, npSaverExecutorService);
-        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

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);
+        super(creditTrackerSettings, npPipelineExecutorService);
-        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

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<PackageOrchestrator.Launch> 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;

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);
     }

src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidEnricher.java

@@ -46,11 +46,12 @@ public class NpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
         log.debug("NpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size());
         //найдем телефоны в табличном кэше
         Map<Long, PhonesMap> cachedRecords = packageRecordsService.findCachedPhones(subjects);
-        processByNumber(subjects, cachedRecords, ctx);
+        boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
+        taskPayload.setHasFids(hasFids);
         return taskPayload;
     }
 
-    private void processByNumber(List<SubjectProfile> subjects,
+    private boolean processByNumber(List<SubjectProfile> subjects,
                                  Map<Long, PhonesMap> cachedRecords,
                                  PipelineContext<PackageVariable> context) {
         List<SubjectProfile> noFidsData = subjects.stream()
@@ -97,6 +98,8 @@ public class NpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
             .toList();
 
         packageRecordsService.saveToCache(subjectForSave);
+
+        return subjectForSave.stream().anyMatch(s -> s.getFid() != null);
     }
 
     private void fillByCache(SubjectProfile subjectProfile, PhonesMap cachedRecord) {

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");