some refactoring :]

src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java

@@ -1,77 +0,0 @@
-package ru.nbch.credit_tracker.config.task;
-
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import ru.nbch.credit_tracker.model.*;
-import ru.nbch.credit_tracker.model.impl.SubjectPayload;
-import ru.nbch.credit_tracker.service.task.*;
-
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import ru.nbch.credit_tracker.service.task.impl.dprctd.FidIdEnricherV2;
-import ru.nbch.credit_tracker.service.task.impl.dprctd.FidNameEnricherV2;
-import ru.nbch.credit_tracker.service.task.impl.dprctd.FlagsEnricher;
-
-@Configuration
-public class ConsumerConfigV2 {
-
-    @Bean
-    public BlockingQueue<PhoneFidPayloadV2> savingRequest2() {
-        return new LinkedBlockingQueue<>();
-    }
-
-    @Bean
-    public BlockingQueue<FlagsPayload> flagsPayload() {
-        return new LinkedBlockingQueue<>();
-    }
-
-    @Bean
-    public BlockingQueue<SubjectsProfilePayloadV2> enricherQueue2() {
-        return new LinkedBlockingQueue<>();
-    }
-
-    @Bean
-    public BlockingQueue<SendSignalsPayload> enricherIdQueue2() {
-        return new LinkedBlockingQueue<>();
-    }
-
-    @Bean
-    public StagePayload POISON() {
-        return () -> true;
-    }
-
-    @SuppressWarnings("unchecked")
-    @Bean
-    public Integer consumerConfigurer(BlockingQueue<PhoneFidPayloadV2> savingRequest2,
-                                      BlockingQueue<SubjectsProfilePayloadV2> enricherQueue2,
-                                      BlockingQueue<SendSignalsPayload> enricherIdQueue2,
-                                      BlockingQueue<FlagsPayload> flagsPayloadQueue,
-                                      FidNameEnricherV2 fidNameEnricherV2,
-                                      FidIdEnricherV2 fidIdEnricherV2,
-                                      FlagsEnricher flagsEnricher) {
-
-        SubjectsProfilePayloadV2 POISON = new SubjectsProfilePayloadV2(true);
-        SendSignalsPayload FID_ID_ENRICHER_POISON = new SendSignalsPayload(true);
-        DummyPayload FLAGS_POISON = new DummyPayload(true);
-        Stage fidNameEnricher = new Stage(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON);
-        Stage fidIdEnricher = new Stage(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON);
-        Stage flagsEnricherConsumer = new Stage(flagsEnricher, flagsPayloadQueue, new LinkedBlockingQueue(), FLAGS_POISON);
-        Thread thread1 = new Thread(fidNameEnricher::start);
-        Thread thread2 = new Thread(fidIdEnricher::start);
-        Thread thread3 = new Thread(flagsEnricherConsumer::start);
-        thread1.start();
-        thread2.start();
-        thread3.start();
-        return 0;
-    }
-
-    @Bean
-    public StageQueueFactory<SubjectPayload> saveSubjectQueueFactory() {
-        return new StageQueueFactory<>();
-    }
-
-    @Bean
-    public StageQueueFactory<SubjectPayload> idEnricherQueueFactory() {
-        return new StageQueueFactory<>();
-    }
-}

src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java

@@ -0,0 +1,19 @@
+package ru.nbch.credit_tracker.config.task;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+
+@Configuration
+public class StageConfig {
+
+    @Bean("saveSubjectQueueFactory")
+    public StageQueueFactory<SubjectPayload> saveSubjectQueueFactory() {
+        return new StageQueueFactory<>();
+    }
+
+    @Bean("idEnricherQueueFactory")
+    public StageQueueFactory<SubjectPayload> idEnricherQueueFactory() {
+        return new StageQueueFactory<>();
+    }
+}

src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java

@@ -16,29 +16,29 @@ import ru.nbch.credit_tracker.entities.user_response.success.MonitoringSetupResp
 import ru.nbch.credit_tracker.enums.SignalError;
 import ru.nbch.credit_tracker.enums.XmlErrorFormat;
 import ru.nbch.credit_tracker.exceptions.SignalException;
-import ru.nbch.credit_tracker.facade.SignalFacadeV2;
+import ru.nbch.credit_tracker.facade.SignalFacadeV3;
 import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp;
 
 @RestController
 @RequestMapping("/api/v3/signal")
 public class SignalControllerV3 {
 
-    private final SignalFacadeV2 signalFacadeV2;
+    private final SignalFacadeV3 signalFacadeV3;
 
-    public SignalControllerV3(SignalFacadeV2 signalFacadeV2) {
+    public SignalControllerV3(SignalFacadeV3 signalFacadeV3) {
-        this.signalFacadeV2 = signalFacadeV2;
+        this.signalFacadeV3 = signalFacadeV3;
     }
 
     @PostMapping(value = "/set", produces = MediaType.APPLICATION_XML_VALUE)
     public ResponseEntity<MonitoringSetupResponse> setupMonitoring(@RequestParam("file") MultipartFile file) {
-        signalFacadeV2.setupMonitoringRequest(file);
+        signalFacadeV3.setupMonitoringRequest(file);
         return ResponseEntity.ok().body(new MonitoringSetupResponse());
     }
 
     @PostMapping(value = "/active_packages", consumes = MediaType.APPLICATION_XML_VALUE, produces = MediaType.APPLICATION_XML_VALUE)
     public ResponseEntity<MonitoringActivePackagesResponse> findActivePackages(@RequestBody MonitoringActivePackagesRequest activePackagesRequest) {
         try {
-            return signalFacadeV2.findActivePackages(activePackagesRequest);
+            return signalFacadeV3.findActivePackages(activePackagesRequest);
         } catch (Exception e) {
             throwOrWrap(SignalError.ERROR_013, XmlErrorFormat.MonitoringActivePackagesError, e);
             return null;
@@ -48,7 +48,7 @@ public class SignalControllerV3 {
     @PostMapping(value = "/all_packages", consumes = MediaType.APPLICATION_XML_VALUE, produces = MediaType.APPLICATION_XML_VALUE)
     public ResponseEntity<MonitoringAllPackagesResponse> findAllPackages(@RequestBody MonitoringAllPackagesRequest allPackagesRequest) {
         try {
-            return signalFacadeV2.findAllPackages(allPackagesRequest);
+            return signalFacadeV3.findAllPackages(allPackagesRequest);
         } catch (Exception e) {
             throwOrWrap(SignalError.ERROR_013, XmlErrorFormat.MonitoringAllPackagesError, e);
             return null;

src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java

@@ -1,31 +0,0 @@
-package ru.nbch.credit_tracker.model;
-
-public class PackageToSavePayloadV2 implements StagePayload {
-    private final Long packageId;
-    private final String filePath;
-    private final boolean poison;
-
-    public PackageToSavePayloadV2(boolean poison) {
-        this.packageId = null;
-        this.filePath = null;
-        this.poison = true;
-    }
-    public PackageToSavePayloadV2(Long packageId, String filePath) {
-        this.packageId = packageId;
-        this.filePath = filePath;
-        this.poison = false;
-    }
-
-    public Long getPackageId() {
-        return packageId;
-    }
-
-    public String getFilePath() {
-        return filePath;
-    }
-
-    @Override
-    public boolean isPoison() {
-        return poison;
-    }
-}

src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java

@@ -1,36 +0,0 @@
-package ru.nbch.credit_tracker.model;
-
-import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-
-import java.util.List;
-
-public class PhoneFidPayloadV2 implements StagePayload {
-    private final Long packageId;
-    private final List<PhoneFidMap> phoneFidsBatch;
-    private final boolean poison;
-
-    public PhoneFidPayloadV2(boolean poison) {
-        this.packageId = null;
-        this.phoneFidsBatch = null;
-        this.poison = poison;
-    }
-
-    public PhoneFidPayloadV2(Long packageId, List<PhoneFidMap> phoneFidsBatch) {
-        this.packageId = packageId;
-        this.phoneFidsBatch = phoneFidsBatch;
-        this.poison = false;
-    }
-
-    public Long getPackageId() {
-        return packageId;
-    }
-
-    public List<PhoneFidMap> getPhoneFidsBatch() {
-        return phoneFidsBatch;
-    }
-
-    @Override
-    public boolean isPoison() {
-        return poison;
-    }
-}

src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java

@@ -46,12 +46,7 @@ public class PhoneRepository {
         FROM PHONE_NORM pn
                  JOIN NAME nf
                       ON nf.fid = pn.fid
-        WHERE number in (:phones)
+        WHERE number in (:phones) WITH UR
-        ORDER BY
-            pn.number ASC,
-            pn.file_since_dt DESC,
-            pn.fid DESC
-        WITH UR
         """;
 
     private final NamedParameterJdbcTemplate jdbcTemplate;

src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java

@@ -119,8 +119,8 @@ public class PhoneService2 {
         log.debug("End merging fids and names");
     }
 
-    Comparator<Name> byDateThenId =
+    Comparator<Name> byDateThenIdNullSafe =
-        Comparator.comparing(Name::getFileSinceDate)   // LocalDate natural order
+        Comparator.comparing(Name::getFileSinceDate, Comparator.nullsFirst(Comparator.naturalOrder()))
             .thenComparing(Name::getFid);
 
     /**
@@ -140,7 +140,7 @@ public class PhoneService2 {
             .collect(Collectors.toMap(
                 Name::getNumber,
                 Function.identity(),
-                BinaryOperator.maxBy(byDateThenId)
+                BinaryOperator.maxBy(byDateThenIdNullSafe)
             ));
     }
     /**

src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java

@@ -14,6 +14,7 @@ import ru.nbch.credit_tracker.entities.indic.Name;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject;
 import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.utils.CTUtil;
 
 @Service
 public class PhoneFidMapService {
@@ -41,7 +42,7 @@ public class PhoneFidMapService {
     public void register(List<SubjectProfile> subjects, Map<String, Name> nameByPhone, Long packageId) {
         List<PhoneFidMap> phoneFidMaps = subjects.stream().map(subjectProfile -> {
                 Integer fid = Optional
-                    .of(nameByPhone.get(subjectProfile.getPhone()))
+                    .ofNullable(nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone())))
                     .map(Name::getFid)
                     .orElse(null);
 

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

@@ -9,9 +9,9 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Component;
 import ru.nbch.credit_tracker.config.task.StageQueueFactory;
@@ -33,8 +33,8 @@ public class PipelinePackageManager {
 
     public PipelinePackageManager(ApplicationContext applicationContext,
                                   StaxXmlProcessor xmlProcessor,
-                                  StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
+                                  @Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
-                                  StageQueueFactory<SubjectPayload> idEnricherQueueFactory) {
+                                  @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory) {
         this.applicationContext = applicationContext;
         this.xmlProcessor = xmlProcessor;
         this.saveSubjectQueueFactory = saveSubjectQueueFactory;
@@ -46,7 +46,7 @@ public class PipelinePackageManager {
         BlockingQueue<SubjectPayload> saveSubjectQueue = saveSubjectQueueFactory.getObject();
         BlockingQueue<SubjectPayload> idEnricherQueue = idEnricherQueueFactory.getObject();
 
-        StageRunnable<SubjectPayload, SubjectPayload> saveSubjectStage = new StageRunnable<>(saveSubjectHandler,
+        Stage<SubjectPayload, SubjectPayload> saveSubjectStage = new Stage<>(saveSubjectHandler,
             saveSubjectQueue,
             idEnricherQueue,
             SubjectPayload.POISON
@@ -80,7 +80,7 @@ public class PipelinePackageManager {
                 saveSubjectQueue.add(new SubjectPayload(true));
             }
         });
-        future.get(10_000, TimeUnit.SECONDS);
+//        future.get(10_000, TimeUnit.SECONDS);
         log.info("Subjects were registered. PackageId: {}", packageId);
     }
 }

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

@@ -15,8 +15,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils;
 import ru.nbch.credit_tracker.model.StagePayload;
 
 @Slf4j
-public class Stage<X extends StagePayload, Z extends StagePayload>
+public class Stage<X extends StagePayload, Z extends StagePayload> implements Runnable {
-{
     private final Handler<X, Z> dispatcher;
     private final BlockingQueue<X> in;
     private final BlockingQueue<Z> out;
@@ -34,7 +33,8 @@ public class Stage<X extends StagePayload, Z extends StagePayload>
         this.POISON = POISON;
     }
 
-    public void start() {
+    @Override
+    public void run() {
         List<Future<Z>> inflight = new ArrayList<>();
         AtomicInteger inFlightCount = new AtomicInteger();
         try {

src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidNameEnricherV2.java

@@ -1,67 +0,0 @@
-package ru.nbch.credit_tracker.service.task.impl.dprctd;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.model.NameData;
-import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.entities.dto.SubjectProfileForSignal;
-import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
-import ru.nbch.credit_tracker.model.FlagsPayload;
-import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
-import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2;
-import ru.nbch.credit_tracker.service.indic.PhoneService2;
-
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import ru.nbch.credit_tracker.service.task.Handler;
-
-@Service
-@Slf4j
-public class FidNameEnricherV2 implements Handler<PhoneFidPayloadV2, SubjectsProfilePayloadV2> {
-    private final PhoneService2 phoneService2;
-    private final PhoneFidMapMapper mapper;
-    private final BlockingQueue<FlagsPayload> flagsPayloadQueue;
-
-    public FidNameEnricherV2(PhoneService2 phoneService2,
-                             PhoneFidMapMapper mapper,
-                             BlockingQueue<FlagsPayload> flagsPayloadQueue) {
-        this.phoneService2 = phoneService2;
-        this.mapper = mapper;
-        this.flagsPayloadQueue = flagsPayloadQueue;
-    }
-
-    @Override
-    public SubjectsProfilePayloadV2 handle(PhoneFidPayloadV2 taskPayload) {
-        List<PhoneFidMap> phoneFidMap = taskPayload.getPhoneFidsBatch();
-        log.debug("FidNameEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMap.size());
-        mapper.registerWithId(phoneFidMap); // надо зарегистрировать phoneFidMap'ы т.к далее id участвует в формированиии запроса в оф. сервис uid = id
-        phoneService2.defineFidsAndNameByPhone2(phoneFidMap);
-        log.debug("Found {} fids by batch", phoneFidMap.stream().filter(s -> s.getFid() != null).count());
-        flagsPayloadQueue.add(new FlagsPayload(taskPayload.getPackageId(), phoneFidMap)); // TODO find out how to use it in other way
-        //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать
-
-        List<SubjectProfileForSignal> subjectProfiles = phoneFidMap.stream()
-                .filter(subject -> subject.getFid() != null)
-                .filter(this::requiredFieldPresent) // TODO log.warn("Person does not have required fields. Fid {}", fid);
-                .map(pf -> {
-                    SubjectProfileForSignal subjectProfile = new SubjectProfileForSignal();
-                    subjectProfile.setId(pf.getId());
-                    subjectProfile.setFid(pf.getFid());
-                    subjectProfile.setNameProfile(NameData.builder()
-                            .firstName(pf.getFirstName())
-                            .middleName(pf.getMiddleName())
-                            .lastName(pf.getLastName())
-                            .birthDate(pf.getBirthDate())
-                            .fileSinceDate(pf.getFileSinceDate())
-                            .build());
-                    return subjectProfile;
-                }).toList();
-
-        return new SubjectsProfilePayloadV2(taskPayload.getPackageId(), subjectProfiles);
-    }
-
-    private boolean requiredFieldPresent(PhoneFidMap phoneFidMap) {
-        return !StringUtils.isAllBlank(phoneFidMap.getFirstName(), phoneFidMap.getLastName()) && phoneFidMap.getBirthDate() != null;
-    }
-}

src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java

@@ -28,7 +28,7 @@ public class FlagsEnricher implements Handler<FlagsPayload, DummyPayload> {
     public DummyPayload handle(FlagsPayload taskPayload) {
         List<PhoneFidMap> phoneFidMaps = taskPayload.getPhoneFidMaps();
         log.debug("FlagsEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMaps.size());
-        flagsService.fillFlagsData(phoneFidMaps);
+//        flagsService.fillFlagsData(phoneFidMaps);
 
         log.debug("Start updating subject's batch");
         if (!phoneFidMaps.isEmpty()) {

src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/StageForDeletion.java

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service.task;
+package ru.nbch.credit_tracker.service.task.impl.dprctd;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -13,19 +13,21 @@ import java.util.concurrent.atomic.AtomicInteger;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import ru.nbch.credit_tracker.model.StagePayload;
+import ru.nbch.credit_tracker.service.task.Handler;
 
 @Slf4j
-public class StageRunnable<X extends StagePayload, Z extends StagePayload> implements Runnable {
+public class StageForDeletion<X extends StagePayload, Z extends StagePayload>
+{
     private final Handler<X, Z> dispatcher;
     private final BlockingQueue<X> in;
     private final BlockingQueue<Z> out;
     private final Z POISON;
     private final ExecutorService ioPool;
 
-    public StageRunnable(Handler<X, Z> dispatcher,
+    public StageForDeletion(Handler<X, Z> dispatcher,
-                         BlockingQueue<X> in,
+                            BlockingQueue<X> in,
-                         BlockingQueue<Z> out,
+                            BlockingQueue<Z> out,
-                         Z POISON) {
+                            Z POISON) {
         this.dispatcher = dispatcher;
         this.in = in;
         this.out = out;
@@ -33,8 +35,7 @@ public class StageRunnable<X extends StagePayload, Z extends StagePayload> imple
         this.POISON = POISON;
     }
 
-    @Override
+    public void start() {
-    public void run() {
         List<Future<Z>> inflight = new ArrayList<>();
         AtomicInteger inFlightCount = new AtomicInteger();
         try {

src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml

@@ -25,7 +25,7 @@
         </foreach>
     </insert>
 
-    <insert id="register" parameterType="java.util.List" useGeneratedKeys="true" keyColumn="id" keyProperty="id">
+    <insert id="register" parameterType="java.util.List">
         INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id)
         values
         <foreach collection="phoneFidMap" item="phoneFid" separator=",">