some refactoring :/

src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java

@@ -2,9 +2,9 @@ package ru.nbch.credit_tracker.builder;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import ru.nbch.credit_tracker.entities.dto.IdProfile;
+import ru.nbch.credit_tracker.model.IdData;
-import ru.nbch.credit_tracker.entities.dto.NameProfile;
+import ru.nbch.credit_tracker.model.NameData;
-import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+import ru.nbch.credit_tracker.entities.dto.SubjectProfileForSignal;
 import ru.nbch.credit_tracker.entities.external_request.signals.Person;
 import ru.nbch.credit_tracker.utils.CTUtil;
 
@@ -31,14 +31,14 @@ public class PersonFactoryBuilderV2 {
     private final static String CONSENT_ISSUE_AUTHORITY = "Кем выдан";
 
 
-    private SubjectProfile personData;
+    private SubjectProfileForSignal personData;
     private Long uid;
 
     public static PersonFactoryBuilderV2 builder() {
         return new PersonFactoryBuilderV2();
     }
 
-    public PersonFactoryBuilderV2 personData(SubjectProfile personData) {
+    public PersonFactoryBuilderV2 personData(SubjectProfileForSignal personData) {
         this.personData = personData;
         this.uid = personData.getId();
         return this;
@@ -48,8 +48,8 @@ public class PersonFactoryBuilderV2 {
         if (personData == null) {
             return null;
         }
-        NameProfile nameData = personData.getNameProfile();
+        NameData nameData = personData.getNameProfile();
-        IdProfile passportData = personData.getIdProfile();
+        IdData passportData = personData.getIdProfile();
         Person person = new Person();
         person.setS(nameData.getLastName());
         person.setN(nameData.getFirstName());

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

@@ -3,10 +3,14 @@ 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 {
@@ -32,7 +36,7 @@ public class ConsumerConfigV2 {
     }
 
     @Bean
-    public TaskPayloadV2 POISON() {
+    public StagePayload POISON() {
         return () -> true;
     }
 
@@ -49,9 +53,9 @@ public class ConsumerConfigV2 {
         SubjectsProfilePayloadV2 POISON = new SubjectsProfilePayloadV2(true);
         SendSignalsPayload FID_ID_ENRICHER_POISON = new SendSignalsPayload(true);
         DummyPayload FLAGS_POISON = new DummyPayload(true);
-        TaskConsumerV2 fidNameEnricher = new TaskConsumerV2(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON);
+        Stage fidNameEnricher = new Stage(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON);
-        TaskConsumerV2 fidIdEnricher = new TaskConsumerV2(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON);
+        Stage fidIdEnricher = new Stage(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON);
-        TaskConsumerV2 flagsEnricherConsumer = new TaskConsumerV2(flagsEnricher, flagsPayloadQueue, new LinkedBlockingQueue(), FLAGS_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);
@@ -60,4 +64,14 @@ public class ConsumerConfigV2 {
         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/StageQueueFactory.java

@@ -0,0 +1,24 @@
+package ru.nbch.credit_tracker.config.task;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.springframework.beans.factory.FactoryBean;
+import ru.nbch.credit_tracker.model.StagePayload;
+
+public class StageQueueFactory<T extends StagePayload> implements FactoryBean<BlockingQueue<T>> {
+
+    @Override
+    public BlockingQueue<T> getObject() throws Exception {
+        return new LinkedBlockingQueue<>();
+    }
+
+    @Override
+    public Class<?> getObjectType() {
+        return BlockingQueue.class;
+    }
+
+    @Override
+    public boolean isSingleton() {
+        return false;
+    }
+}

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

@@ -0,0 +1,65 @@
+package ru.nbch.credit_tracker.controller;
+
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.multipart.MultipartFile;
+import ru.nbch.credit_tracker.entities.user_request.active_packages.MonitoringActivePackagesRequest;
+import ru.nbch.credit_tracker.entities.user_request.all_packages.MonitoringAllPackagesRequest;
+import ru.nbch.credit_tracker.entities.user_response.active_packages.MonitoringActivePackagesResponse;
+import ru.nbch.credit_tracker.entities.user_response.all_packages.MonitoringAllPackagesResponse;
+import ru.nbch.credit_tracker.entities.user_response.success.MonitoringSetupResponse;
+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 static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp;
+
+@RestController
+@RequestMapping("/api/v3/signal")
+public class SignalControllerV3 {
+
+    private final SignalFacadeV2 signalFacadeV2;
+
+    public SignalControllerV3(SignalFacadeV2 signalFacadeV2) {
+        this.signalFacadeV2 = signalFacadeV2;
+    }
+
+    @PostMapping(value = "/set", produces = MediaType.APPLICATION_XML_VALUE)
+    public ResponseEntity<MonitoringSetupResponse> setupMonitoring(@RequestParam("file") MultipartFile file) {
+        signalFacadeV2.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);
+        } catch (Exception e) {
+            throwOrWrap(SignalError.ERROR_013, XmlErrorFormat.MonitoringActivePackagesError, e);
+            return null;
+        }
+    }
+
+    @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);
+        } catch (Exception e) {
+            throwOrWrap(SignalError.ERROR_013, XmlErrorFormat.MonitoringAllPackagesError, e);
+            return null;
+        }
+    }
+
+    private static void throwOrWrap(SignalError error, XmlErrorFormat specific, Exception ex) {
+        if (ex instanceof SignalException) {
+            throw (SignalException) ex;
+        }
+        throw serverExcp(error, specific, ex);
+    }
+}
+

src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java

@@ -1,14 +0,0 @@
-package ru.nbch.credit_tracker.entities.dto;
-
-import lombok.Getter;
-import lombok.Setter;
-
-@Getter
-@Setter
-public class SubjectProfile {
-    private Long id;
-    private Integer fid;
-    private IdProfile idProfile;
-    private NameProfile nameProfile;
-    private boolean isFilled = false;
-}

src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfileForSignal.java

@@ -0,0 +1,17 @@
+package ru.nbch.credit_tracker.entities.dto;
+
+import lombok.Getter;
+import lombok.Setter;
+import ru.nbch.credit_tracker.model.IdData;
+import ru.nbch.credit_tracker.model.NameData;
+
+@Getter
+@Setter
+@Deprecated
+public class SubjectProfileForSignal {
+    private Long id;
+    private Integer fid;
+    private IdData idProfile;
+    private NameData nameProfile;
+    private boolean isFilled = false;
+}

src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java

@@ -0,0 +1,20 @@
+package ru.nbch.credit_tracker.entities.indic;
+
+import java.time.LocalDate;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+@Builder
+public class Name {
+    private Integer fid;
+    private String number;
+    private LocalDate fileSinceDate;
+    private String first;
+    private String last;
+    private String middle;
+    private LocalDate birthDate;
+
+}

src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java

@@ -0,0 +1,172 @@
+package ru.nbch.credit_tracker.facade;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import javax.xml.stream.XMLStreamException;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.FileUtils;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Service;
+import org.springframework.web.multipart.MultipartFile;
+import ru.nbch.credit_tracker.component.MonitoringRequestValidator;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.dto.UserPackage;
+import ru.nbch.credit_tracker.entities.user_request.active_packages.MonitoringActivePackagesRequest;
+import ru.nbch.credit_tracker.entities.user_request.all_packages.MonitoringAllPackagesRequest;
+import ru.nbch.credit_tracker.entities.user_response.active_packages.MonitoringActivePackagesResponse;
+import ru.nbch.credit_tracker.entities.user_response.active_packages.Package;
+import ru.nbch.credit_tracker.entities.user_response.active_packages.Packages;
+import ru.nbch.credit_tracker.entities.user_response.all_packages.MonitoringAllPackagesResponse;
+import ru.nbch.credit_tracker.entities.user_response.all_packages.PackageAll;
+import ru.nbch.credit_tracker.entities.user_response.all_packages.PackagesAll;
+import ru.nbch.credit_tracker.enums.SignalError;
+import ru.nbch.credit_tracker.enums.Status;
+import ru.nbch.credit_tracker.enums.XmlErrorFormat;
+import ru.nbch.credit_tracker.exceptions.SignalClientException;
+import ru.nbch.credit_tracker.model.MonitoringValidationError;
+import ru.nbch.credit_tracker.model.ValidationResult;
+import ru.nbch.credit_tracker.service.SignalService;
+import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
+import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
+import ru.nbch.credit_tracker.utils.CTUtil;
+import ru.nbch.credit_tracker.utils.ZipUtils;
+import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp;
+import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp;
+
+@Service
+@Slf4j
+public class SignalFacadeV3 {
+
+    private final MonitoringRequestValidator monitoringRequestValidator;
+    private final CreditTrackerSettings config;
+    private final SignalService signalService;
+    private final AuthorizationService authorizationService;
+    private final PipelinePackageManager pipelinePackageManager;
+    public SignalFacadeV3(MonitoringRequestValidator monitoringRequestValidator,
+                          CreditTrackerSettings config,
+                          SignalService signalService,
+                          AuthorizationService authorizationService,
+                          PipelinePackageManager pipelinePackageManager) {
+        this.monitoringRequestValidator = monitoringRequestValidator;
+        this.config = config;
+        this.signalService = signalService;
+        this.authorizationService = authorizationService;
+        this.pipelinePackageManager = pipelinePackageManager;
+    }
+
+    public void setupMonitoringRequest(MultipartFile file) {
+        log.debug("Start processing monitoring request");
+        monitoringRequestValidator.validateZipContent(file);
+
+        log.trace("Saving monitoring request to tmp file...");
+        String filePath;
+        try {
+            String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator;
+            filePath = ZipUtils.writeTmpFile(folderPath, file);
+        } catch (IOException e) {
+            throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError, e);
+        }
+
+        log.debug("File saved to {}", filePath);
+        ValidationResult validationResult = null;
+        try {
+            validationResult = monitoringRequestValidator.readAndValidateMonitoringRequest(filePath);
+            if (!validationResult.getValidationErrors().isEmpty()) {
+                MonitoringValidationError firstError = validationResult.getValidationErrors().get(0);
+                signalService.rejectPackage(validationResult.getUserId(), validationResult.getPackageName());
+                FileUtils.delete(new File(filePath));
+                throw firstError.getSignalException();
+            }
+        } catch (RuntimeException e) {
+            if (e.getCause() instanceof XMLStreamException) {
+                throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError);
+            }
+            throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError, e);
+        } catch (IOException e) {
+            log.warn("Error deleting tmp file", e);
+        }
+
+        String packageName = validationResult.getPackageName();
+        String userId = validationResult.getUserId();
+        String password = validationResult.getPassword();
+        try {
+            // проверка авторизации
+            AuthorizationService.AuthResult authResult = authorizationService.findAuthInfo(userId, password);
+            if (!authResult.isSuccess()) {
+                signalService.rejectPackage(userId, packageName);
+                throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringError);
+            }
+
+            // проверка на уникальность пакета
+//            if (signalService.isPackageNotUnique(userId, packageName)) {
+//                signalService.rejectPackage(userId, packageName);
+//                throw clientExcp(SignalError.ERROR_007, XmlErrorFormat.MonitoringError);
+//            }
+
+            Long packageId = signalService.registerPackage(userId, packageName);
+            log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name());
+            pipelinePackageManager.buildAndRun(packageId, filePath);
+        } catch (Exception e) {
+            CTUtil.removeFile(filePath);
+            if (e instanceof SignalClientException) {
+                throw (SignalClientException) e;
+            }
+            throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError, e);
+        }
+    }
+
+    public ResponseEntity<MonitoringActivePackagesResponse> findActivePackages(MonitoringActivePackagesRequest activePackagesRequest) throws Exception {
+        AuthorizationService.AuthResult authResult = authorizationService.findAuthInfo(activePackagesRequest.getUserId(), activePackagesRequest.getPassword());
+        if (!authResult.isSuccess()) {
+            throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringActivePackagesError);
+        }
+
+        List<UserPackage> activePackages = signalService.findActivePackages(activePackagesRequest.getUserId());
+        if (activePackages.isEmpty()) {
+            throw clientExcp(SignalError.ERROR_015, XmlErrorFormat.MonitoringActivePackagesError);
+        }
+
+        MonitoringActivePackagesResponse response = new MonitoringActivePackagesResponse();
+        response.setStatus("OK");
+        response.setMembercode(activePackagesRequest.getUserId().substring(0, 6));
+        Packages packages = new Packages();
+        activePackages.forEach(data -> {
+            Package pack = new Package();
+            pack.setPackageName(data.getPackageName());
+            pack.setCreatedAt(data.getCreatedAt() != null ? data.getCreatedAt().toString() : null);
+            packages.getPackages().add(pack);
+        });
+        response.setPackages(packages);
+        return ResponseEntity.ok(response);
+    }
+
+    public ResponseEntity<MonitoringAllPackagesResponse> findAllPackages(MonitoringAllPackagesRequest allPackagesRequest) throws Exception {
+        AuthorizationService.AuthResult authResult = authorizationService.findAuthInfo(allPackagesRequest.getUserId(), allPackagesRequest.getPassword());
+        if (!authResult.isSuccess()) {
+            throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringAllPackagesError);
+        }
+
+        List<UserPackage> activePackages = signalService.findAllPackages(allPackagesRequest.getUserId());
+        if (activePackages.isEmpty()) {
+            throw clientExcp(SignalError.ERROR_019, XmlErrorFormat.MonitoringAllPackagesError);
+        }
+
+        MonitoringAllPackagesResponse response = new MonitoringAllPackagesResponse();
+        response.setStatus("OK");
+        response.setMembercode(allPackagesRequest.getUserId().substring(0, 6));
+        PackagesAll packages = new PackagesAll();
+        activePackages.forEach(data -> {
+            PackageAll pack = new PackageAll();
+            pack.setPackageName(data.getPackageName());
+            pack.setCreatedAt(data.getCreatedAt() != null ? data.getCreatedAt().toString() : null);
+            pack.setStatus(data.getStatus());
+            if (data.getStatus().equals(Status.DELETED.name())) {
+                pack.setDeactivatedAt(data.getDeactivatedAt() != null ? data.getDeactivatedAt().toString() : null);
+            }
+            packages.getPackages().add(pack);
+        });
+        response.setPackages(packages);
+        return ResponseEntity.ok(response);
+    }
+}

src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java

@@ -12,7 +12,7 @@ import java.util.List;
 public interface PhoneFidMapMapper {
 
     void registerSubjects(@Param("subjects") List<Subject> subjects, @Param("packageId") Long packageId);
-    void register(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap, @Param("packageId") Long packageId);
+    void register(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
     void registerWithId(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
 
     List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("offset") int offset, @Param("limit") int limit);

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

@@ -1,8 +1,6 @@
 package ru.nbch.credit_tracker.model;
 
-import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+public class DummyPayload implements StagePayload {
-
-public class DummyPayload implements TaskPayloadV2 {
 
     private final boolean poison;
 

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

@@ -1,11 +1,10 @@
 package ru.nbch.credit_tracker.model;
 
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
 
 import java.util.List;
 
-public class FlagsPayload implements TaskPayloadV2 {
+public class FlagsPayload implements StagePayload {
     private final Long packageId;
     private final List<PhoneFidMap> phoneFidMaps;
     private final boolean poison;

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

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.entities.dto;
+package ru.nbch.credit_tracker.model;
 
 
 import java.time.LocalDate;
@@ -9,7 +9,8 @@ import lombok.Setter;
 @Getter
 @Setter
 @Builder
-public class IdProfile {
+//todo подумать над неймингом класс
+public class IdData {
     private Integer fid;
     private String serNum;
     private String idNum;

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

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.entities.dto;
+package ru.nbch.credit_tracker.model;
 
 
 import java.time.LocalDate;
@@ -9,7 +9,8 @@ import lombok.Setter;
 @Getter
 @Setter
 @Builder
-public class NameProfile {
+//todo подумать над неймингом класс
+public class NameData {
     private Integer fid;
     private String firstName;
     private String lastName;

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

@@ -1,8 +1,6 @@
 package ru.nbch.credit_tracker.model;
 
-import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+public class PackageToSavePayloadV2 implements StagePayload {
-
-public class PackageToSavePayloadV2 implements TaskPayloadV2 {
     private final Long packageId;
     private final String filePath;
     private final boolean poison;

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

@@ -1,11 +1,10 @@
 package ru.nbch.credit_tracker.model;
 
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
 
 import java.util.List;
 
-public class PhoneFidPayloadV2 implements TaskPayloadV2 {
+public class PhoneFidPayloadV2 implements StagePayload {
     private final Long packageId;
     private final List<PhoneFidMap> phoneFidsBatch;
     private final boolean poison;

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

@@ -1,8 +1,6 @@
 package ru.nbch.credit_tracker.model;
 
-import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
+public class SendSignalsPayload implements StagePayload {
-
-public class SendSignalsPayload implements TaskPayloadV2 {
     private String signalXmlPath;
     private boolean poison;
 

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

@@ -0,0 +1,5 @@
+package ru.nbch.credit_tracker.model;
+
+public interface StagePayload {
+    boolean isPoison();
+}

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

@@ -0,0 +1,17 @@
+package ru.nbch.credit_tracker.model;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+@Builder
+public class SubjectProfile {
+    private String identifier;
+    private String phone;
+    private Integer fid;
+    private NameData nameData;
+    private IdData idData;
+
+}

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

@@ -1,13 +1,12 @@
 package ru.nbch.credit_tracker.model;
 
-import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+import ru.nbch.credit_tracker.entities.dto.SubjectProfileForSignal;
-import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
 
 import java.util.List;
 
-public class SubjectsProfilePayloadV2 implements TaskPayloadV2 {
+public class SubjectsProfilePayloadV2 implements StagePayload {
     private final Long packageId;
-    private final List<SubjectProfile> subjectProfiles;
+    private final List<SubjectProfileForSignal> subjectProfiles;
     private final boolean poison;
 
     public SubjectsProfilePayloadV2(boolean poison) {
@@ -16,7 +15,7 @@ public class SubjectsProfilePayloadV2 implements TaskPayloadV2 {
         this.poison = poison;
     }
 
-    public SubjectsProfilePayloadV2(Long packageId, List<SubjectProfile> subjectProfiles) {
+    public SubjectsProfilePayloadV2(Long packageId, List<SubjectProfileForSignal> subjectProfiles) {
         this.packageId = packageId;
         this.subjectProfiles = subjectProfiles;
         this.poison = false;
@@ -26,7 +25,7 @@ public class SubjectsProfilePayloadV2 implements TaskPayloadV2 {
         return packageId;
     }
 
-    public List<SubjectProfile> getSubjectProfiles() {
+    public List<SubjectProfileForSignal> getSubjectProfiles() {
         return subjectProfiles;
     }
 

src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java

@@ -0,0 +1,38 @@
+package ru.nbch.credit_tracker.model.impl;
+
+import java.util.List;
+import ru.nbch.credit_tracker.model.StagePayload;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+
+
+public class SubjectPayload implements StagePayload {
+    private final Long packageId;
+    private final List<SubjectProfile> subjects;
+    public static final SubjectPayload POISON = new SubjectPayload(true);
+    private final boolean poison;
+
+    public SubjectPayload(boolean poison) {
+        this.packageId = null;
+        this.subjects = null;
+        this.poison = poison;
+    }
+
+    public SubjectPayload(Long packageId, List<SubjectProfile> subjects) {
+        this.packageId = packageId;
+        this.subjects = subjects;
+        this.poison = false;
+    }
+
+    public Long getPackageId() {
+        return packageId;
+    }
+
+    public List<SubjectProfile> getSubjects() {
+        return subjects;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return poison;
+    }
+}

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

@@ -12,6 +12,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Repository;
+import ru.nbch.credit_tracker.entities.indic.Name;
 
 @Repository
 public class PhoneRepository {
@@ -69,6 +70,7 @@ public class PhoneRepository {
         );
     }
 
+    @Deprecated
     public List<FidsDataWithName> findFidsWithName(List<String> phones) {
         MapSqlParameterSource params = new MapSqlParameterSource();
         params.addValue("phones", phones);
@@ -86,6 +88,22 @@ public class PhoneRepository {
             ));
     }
 
+    public List<Name> findNameByPhones(List<String> phones) {
+        MapSqlParameterSource params = new MapSqlParameterSource();
+        params.addValue("phones", phones);
+
+        return jdbcTemplate.query(SELCT_FIDS_WITH_NAME, params, (rs, rowNum) ->
+            Name.builder()
+                .fid(rs.getInt("FID"))
+                .number(rs.getString("NUMBER"))
+                .fileSinceDate(getLocalDate(rs, "FILE_SINCE_DT_NAME"))
+                .first(rs.getString("FIRST"))
+                .middle(rs.getString("MIDDLE"))
+                .last(rs.getString("NAME_1"))
+                .birthDate(getLocalDate(rs, "BIRTH_DT"))
+                .build());
+    }
+
     private static LocalDate getLocalDate(ResultSet rs, String col) throws SQLException {
         java.sql.Date d = rs.getDate(col);
         return d != null ? d.toLocalDate() : null;

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

@@ -6,8 +6,8 @@ import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.entities.dto.IdProfile;
+import ru.nbch.credit_tracker.model.IdData;
-import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+import ru.nbch.credit_tracker.entities.dto.SubjectProfileForSignal;
 import ru.nbch.credit_tracker.mapper.indic.PersonMapper;
 
 import java.time.LocalDate;
@@ -89,8 +89,8 @@ public class PersonService {
         return resultMap;
     }
 
-    public void filledPassportsData(List<SubjectProfile> subjectProfiles) {
+    public void filledPassportsData(List<SubjectProfileForSignal> subjectProfiles) {
-        List<Integer> fids = subjectProfiles.stream().map(SubjectProfile::getFid).collect(Collectors.toList());
+        List<Integer> fids = subjectProfiles.stream().map(SubjectProfileForSignal::getFid).collect(Collectors.toList());
         if (fids.isEmpty()) {
             return;
         }
@@ -102,7 +102,7 @@ public class PersonService {
         subjectProfiles.forEach(subjectProfile -> {
             PassportData passport = findPassport(passportsMap.get(subjectProfile.getFid()));
             if (passport != null) {
-                IdProfile idProfile = IdProfile.builder()
+                IdData idProfile = IdData.builder()
                     .serNum(passport.getSerNum())
                     .idNum(passport.getIdNum())
                     .issDate(passport.getIssDate())

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

@@ -1,14 +1,22 @@
 package ru.nbch.credit_tracker.service.indic;
 
+import java.time.LocalDate;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BinaryOperator;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
+import ru.nbch.credit_tracker.entities.indic.Name;
+import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.repository.PhoneRepository;
-
+import ru.nbch.credit_tracker.utils.CTUtil;
-import java.time.LocalDate;
-import java.util.*;
-import java.util.stream.Collectors;
 
 @Slf4j
 @Service
@@ -111,6 +119,30 @@ public class PhoneService2 {
         log.debug("End merging fids and names");
     }
 
+    Comparator<Name> byDateThenId =
+        Comparator.comparing(Name::getFileSinceDate)   // LocalDate natural order
+            .thenComparing(Name::getFid);
+
+    /**
+     * Поиск данных по таблицам PHONE_NORM и NAME по списку номеров телефонов
+     * @param subjects
+     * @return - отображение NAME к номеру телефону, отфильтрованному по логике
+     * FID с максимальным значением FILE_SINCE_DT. Если FILE_SINCE_DT оказался одинаковый, берется максимальный FID
+     */
+    public Map<String, Name> findNameByNumber(List<SubjectProfile> subjects) {
+        List<String> phones = subjects.stream()
+            .map(subject -> CTUtil.normalizePhoneNumber(subject.getPhone()))
+            .toList();
+
+        List<Name> names = phoneRepository.findNameByPhones(phones);
+
+        return names.stream()
+            .collect(Collectors.toMap(
+                Name::getNumber,
+                Function.identity(),
+                BinaryOperator.maxBy(byDateThenId)
+            ));
+    }
     /**
      * Ищет фиды по номерам телефона. Обогащает {@link PhoneFidMap} для найденных фидов данными из Name и фидом.
      * @return Map: key - FID, value - {@link PhoneFidMap}
@@ -171,6 +203,7 @@ public class PhoneService2 {
         return phoneFidMap;
     }
 
+    @Deprecated
     private PhoneRepository.FidsDataWithName defineNameData(List<PhoneRepository.FidsDataWithName> dataList) {
         PhoneRepository.FidsDataWithName currentData = null;
         for (PhoneRepository.FidsDataWithName fidsData : dataList) {
@@ -191,6 +224,26 @@ public class PhoneService2 {
         return currentData;
     }
 
+    private Name defineNameDataV2(List<Name> dataList) {
+        Name currentData = null;
+        for (Name fidsData : dataList) {
+            if (currentData == null) {
+                currentData = fidsData;
+                continue;
+            }
+            LocalDate currentDate = currentData.getFileSinceDate();
+            LocalDate newDate = fidsData.getFileSinceDate();
+            if (currentDate != null && newDate != null) {
+                if (currentDate.isBefore(newDate)) {
+                    currentData = fidsData;
+                }
+            } else if (currentDate == null) {
+                currentData = fidsData;
+            }
+        }
+        return currentData;
+    }
+
     private Map<Integer, PhoneFidMap> merger(List<PhoneFidMap> subjects, List<PhoneRepository.FidsData> phonesByFid) {
         Map<Integer, PhoneFidMap> phoneFidMap = new HashMap<>();
         Iterator<PhoneRepository.FidsData> iterator = phonesByFid.iterator();

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

@@ -2,32 +2,58 @@ package ru.nbch.credit_tracker.service.signal;
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import org.apache.ibatis.session.ExecutorType;
 import org.apache.ibatis.session.SqlSession;
 import org.apache.ibatis.session.SqlSessionFactory;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
+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;
 
 @Service
 public class PhoneFidMapService {
 
     private final PhoneFidMapMapper mapper;
     private final SqlSessionFactory sqlSessionFactory;
+
     public PhoneFidMapService(PhoneFidMapMapper mapper,
                               @Qualifier("sqlSessionSignal") SqlSessionFactory sqlSessionFactory) {
         this.mapper = mapper;
         this.sqlSessionFactory = sqlSessionFactory;
     }
 
+    @Deprecated
     public void registerSubjects(List<Subject> subjects, Long packageId) {
         mapper.registerSubjects(subjects, packageId);
     }
 
-    public void register(List<PhoneFidMap> phoneFidMap, Long packageId) {
+    /**
-        mapper.register(phoneFidMap, packageId);
+     * Регистрирует в базе список субъектов переданных в запросе
+     *
+     * @param subjects  - список субъектов
+     * @param packageId - уникальный идентификатор пакета
+     */
+    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()))
+                    .map(Name::getFid)
+                    .orElse(null);
+
+                return PhoneFidMap.builder()
+                    .phone(subjectProfile.getPhone())
+                    .subjectId(subjectProfile.getIdentifier())
+                    .packageId(packageId)
+                    .fid(fid)
+                    .build();
+            }
+        ).toList();
+        mapper.register(phoneFidMaps);
     }
 
     public List<PhoneFidMap> findFullPhoneFidMap(Long packageId, int offset, int limit) {

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

@@ -1,5 +0,0 @@
-package ru.nbch.credit_tracker.service.task;
-
-public interface Dispatcher<X> {
-    void handle(X taskPayload);
-}

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

@@ -1,5 +1,5 @@
 package ru.nbch.credit_tracker.service.task;
 
-public interface DispatcherV2<X, Z> {
+public interface Handler<X, Z> {
     Z handle(X taskPayload);
 }

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

@@ -0,0 +1,86 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+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.context.ApplicationContext;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.config.task.StageQueueFactory;
+import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+
+@Component
+@Slf4j
+public class PipelinePackageManager {
+    private final ApplicationContext applicationContext;
+    private final StaxXmlProcessor xmlProcessor;
+    private final StageQueueFactory<SubjectPayload> saveSubjectQueueFactory;
+    private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
+    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
+    private final static int BATCH_SIZE = 10_000;
+
+    public PipelinePackageManager(ApplicationContext applicationContext,
+                                  StaxXmlProcessor xmlProcessor,
+                                  StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
+                                  StageQueueFactory<SubjectPayload> idEnricherQueueFactory) {
+        this.applicationContext = applicationContext;
+        this.xmlProcessor = xmlProcessor;
+        this.saveSubjectQueueFactory = saveSubjectQueueFactory;
+        this.idEnricherQueueFactory = idEnricherQueueFactory;
+    }
+
+    public void buildAndRun(Long packageId, String pkgPath) throws Exception {
+        SaveSubjectHandler saveSubjectHandler = applicationContext.getBean(SaveSubjectHandler.class);
+        BlockingQueue<SubjectPayload> saveSubjectQueue = saveSubjectQueueFactory.getObject();
+        BlockingQueue<SubjectPayload> idEnricherQueue = idEnricherQueueFactory.getObject();
+
+        StageRunnable<SubjectPayload, SubjectPayload> saveSubjectStage = new StageRunnable<>(saveSubjectHandler,
+            saveSubjectQueue,
+            idEnricherQueue,
+            SubjectPayload.POISON
+        );
+        Future<?> future = executorService.submit(saveSubjectStage);
+
+        executorService.submit(() -> {
+            log.info("Start registering subjects. PackageId: {}", packageId);
+            try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(pkgPath)), 8 * 1024 * 1024)) {
+                xmlProcessor.read(in, MonitoringRequest.class,
+                    request -> {
+                        if (!request.getSubjects().getSubject().isEmpty()) {
+                            List<SubjectProfile> subjectProfileBatch = request.getSubjects().getSubject().stream().map(subject ->
+                                SubjectProfile.builder()
+                                    .identifier(subject.getId())
+                                    .phone(subject.getClientId())
+                                    .build()
+
+                            ).collect(Collectors.toList());
+                            log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
+                            saveSubjectQueue.add(new SubjectPayload(packageId, subjectProfileBatch));
+                            request.getSubjects().getSubject().clear();
+                        }
+                    },
+                    request -> request.getSubjects().getSubject().isEmpty()
+                    ,
+                    BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
+            } catch (IOException e) {
+                throw new RuntimeException("Error while parsing monitoring request", e);
+            } finally {
+                saveSubjectQueue.add(new SubjectPayload(true));
+            }
+        });
+        future.get(10_000, TimeUnit.SECONDS);
+        log.info("Subjects were registered. PackageId: {}", packageId);
+    }
+}

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

@@ -1,62 +0,0 @@
-package ru.nbch.credit_tracker.service.task;
-
-import java.io.BufferedInputStream;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.stream.Collectors;
-import lombok.extern.slf4j.Slf4j;
-import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
-import ru.nbch.credit_tracker.model.PackageToSavePayloadV2;
-import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
-import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
-
-@Slf4j
-public class PipelinePackageProcessor implements Runnable {
-    private final StaxXmlProcessor xmlProcessor;
-    private final static int BATCH_SIZE = 10_000;
-    private final PackageToSavePayloadV2 taskPayload;
-
-    public PipelinePackageProcessor(StaxXmlProcessor xmlProcessor,
-                                    PackageToSavePayloadV2 taskPayload) {
-        this.xmlProcessor = xmlProcessor;
-        this.taskPayload = taskPayload;
-    }
-    @Override
-    public void run() {
-        BlockingQueue<PhoneFidPayloadV2> savingRequest2 = new LinkedBlockingQueue<>();
-        Long packageId = taskPayload.getPackageId();
-        Path pkgPath = Paths.get(taskPayload.getFilePath());
-        log.info("Start registering subjects. PackageId: {}", taskPayload.getPackageId());
-        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) {
-            xmlProcessor.read(in, MonitoringRequest.class,
-                request -> {
-                    if (!request.getSubjects().getSubject().isEmpty()) {
-                        List<PhoneFidMap> phoneFidMap = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder()
-                            .subjectId(subject.getId())
-                            .recUuid(UUID.randomUUID())
-                            .phone(subject.getClientId())
-                            .packageId(packageId)
-                            .build()).collect(Collectors.toList());
-                        log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
-                        savingRequest2.add(new PhoneFidPayloadV2(packageId, phoneFidMap));
-                        request.getSubjects().getSubject().clear();
-                    }
-                },
-                request -> request.getSubjects().getSubject().isEmpty()
-                ,
-                BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
-        } catch (IOException e) {
-            throw new RuntimeException("Error while parsing monitoring request", e);
-        } finally {
-            savingRequest2.add(new PhoneFidPayloadV2(true));
-        }
-        log.info("Subjects were registered. PackageId: {}", packageId);
-    }
-}

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

@@ -12,31 +12,31 @@ import java.util.concurrent.TimeUnit;
 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;
 
 @Slf4j
-public class TaskConsumerV2<X extends TaskPayloadV2, Z extends TaskPayloadV2>
+public class Stage<X extends StagePayload, Z extends StagePayload>
 {
-    private final DispatcherV2<X, Z> dispatcher;
+    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 TaskConsumerV2(DispatcherV2<X, Z> dispatcher,
+    public Stage(Handler<X, Z> dispatcher,
                  BlockingQueue<X> in,
                  BlockingQueue<Z> out,
                  Z POISON) {
         this.dispatcher = dispatcher;
         this.in = in;
         this.out = out;
-        this.ioPool = newIoExecutor();
+        this.ioPool = Executors.newFixedThreadPool(10);
         this.POISON = POISON;
     }
 
     public void start() {
         List<Future<Z>> inflight = new ArrayList<>();
         AtomicInteger inFlightCount = new AtomicInteger();
-
         try {
             while (true) {
                 X b = takeQuiet(in, Duration.ofSeconds(5));

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

@@ -0,0 +1,135 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+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.concurrent.atomic.AtomicInteger;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import ru.nbch.credit_tracker.model.StagePayload;
+
+@Slf4j
+public class StageRunnable<X extends StagePayload, Z extends StagePayload> implements Runnable {
+    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,
+                         BlockingQueue<X> in,
+                         BlockingQueue<Z> out,
+                         Z POISON) {
+        this.dispatcher = dispatcher;
+        this.in = in;
+        this.out = out;
+        this.ioPool = Executors.newFixedThreadPool(10);
+        this.POISON = POISON;
+    }
+
+    @Override
+    public void run() {
+        List<Future<Z>> inflight = new ArrayList<>();
+        AtomicInteger inFlightCount = new AtomicInteger();
+        try {
+            while (true) {
+                X b = takeQuiet(in, Duration.ofSeconds(5));
+                if (b == null) continue;
+                if (b.isPoison()) {
+                    // дождаться всех in-flight задач
+                    for (Future<Z> f : inflight) {
+                        putQuiet(out, f.get());
+                    }
+                    putQuiet(out, POISON);
+                    return;
+                }
+
+                Future<Z> f = ioPool.submit(() -> {
+                        Z p = dispatcher.handle(b);
+                        putQuiet(out, p);
+                        return p;
+                    }
+                );
+                inflight.add(f);
+
+                // чтобы не накапливать слишком много future в памяти, периодически сливаем
+                if (inFlightCount.incrementAndGet() % 8 == 0) {
+                    drainDone(inflight, out);
+                }
+            }
+        } catch (Exception e) {
+            log.error("Error: {}", ExceptionUtils.getStackTrace(e));
+            putQuiet(out, POISON);
+        } finally {
+            drainAll(inflight, out);
+        }
+    }
+
+    void drainDone(List<Future<Z>> fs, BlockingQueue<Z> out) throws Exception {
+        Iterator<Future<Z>> it = fs.iterator();
+        while (it.hasNext()) {
+            Future<Z> f = it.next();
+            if (f.isDone()) {
+                putQuiet(out, f.get());
+                it.remove();
+            }
+        }
+    }
+
+    void drainAll(List<Future<Z>> fs, BlockingQueue<Z> out) {
+        for (Future<Z> f : fs) {
+            try {
+                putQuiet(out, f.get());
+            } catch (Exception e) {
+                log.error("Error: {}", ExceptionUtils.getStackTrace(e));
+            }
+        }
+        fs.clear();
+    }
+
+    private ExecutorService newIoExecutor() {
+        // один carrier на ядро, задачи — виртуальные
+        return Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("vt-io-", 0).factory());
+    }
+
+    void putQuiet(BlockingQueue<Z> q, Z b) {
+        try {
+            q.put(b);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    X takeQuiet(BlockingQueue<X> q, Duration timeout) {
+        try {
+            return q.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return null;
+        }
+    }
+
+    static void sleep(long ms) {
+        try {
+            Thread.sleep(ms);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+//    @Override
+    public void stop() {
+
+    }
+
+//    @Override
+    public boolean isRunning() {
+        return true;
+    }
+}

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

@@ -1,5 +0,0 @@
-package ru.nbch.credit_tracker.service.task;
-
-public interface TaskPayloadV2 {
-    boolean isPoison();
-}

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

@@ -0,0 +1,64 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.entities.indic.Name;
+import ru.nbch.credit_tracker.model.NameData;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+import ru.nbch.credit_tracker.service.indic.PhoneService2;
+import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
+import ru.nbch.credit_tracker.service.task.Handler;
+import ru.nbch.credit_tracker.utils.CTUtil;
+
+/**
+ * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных.
+ * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки.
+ */
+@Service
+@Scope("prototype")
+@Slf4j
+//todo опять нейминг класса, варианты: ResolveSubject; DefineIcrsSubject; IcrsMapping/Resolving; PackageSaver; Filler...
+public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayload> {
+    private final PhoneService2 phoneService2;
+    private final PhoneFidMapService phoneFidMapService;
+
+    public SaveSubjectHandler(PhoneService2 phoneService2,
+                              PhoneFidMapService phoneFidMapService) {
+        this.phoneService2 = phoneService2;
+        this.phoneFidMapService = phoneFidMapService;
+    }
+
+    @Override
+    public SubjectPayload handle(SubjectPayload taskPayload) {
+        List<SubjectProfile> subjects = taskPayload.getSubjects();
+        Long packageId = taskPayload.getPackageId();
+        log.debug("SaveSubjectStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size());
+
+        Map<String, Name> nameByPhone = phoneService2.findNameByNumber(subjects);
+        phoneFidMapService.register(subjects, nameByPhone, packageId);
+
+        //fill name info
+        subjects.stream()
+            .filter(subject -> nameByPhone.containsKey(CTUtil.normalizePhoneNumber(subject.getPhone())))
+            .forEach(subject -> {
+                Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone()));
+                NameData nameData = NameData.builder()
+                    .middleName(name.getMiddle())
+                    .firstName(name.getFirst())
+                    .lastName(name.getLast())
+                    .fileSinceDate(name.getFileSinceDate())
+                    .birthDate(name.getBirthDate())
+                    .build();
+
+                subject.setFid(name.getFid());
+                subject.setNameData(nameData);
+            });
+
+        //todo стоит ли делать имутабл для subject?
+        return new SubjectPayload(taskPayload.getPackageId(), subjects);
+    }
+}

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

@@ -1,16 +1,17 @@
-package ru.nbch.credit_tracker.service.task;
+package ru.nbch.credit_tracker.service.task.impl.dprctd;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.builder.PersonFactoryBuilderV2;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+import ru.nbch.credit_tracker.entities.dto.SubjectProfileForSignal;
 import ru.nbch.credit_tracker.entities.external_request.signals.Person;
 import ru.nbch.credit_tracker.model.PartFilePayload;
 import ru.nbch.credit_tracker.model.SendSignalsPayload;
 import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2;
 import ru.nbch.credit_tracker.service.SignalService;
 import ru.nbch.credit_tracker.service.indic.PersonService;
+import ru.nbch.credit_tracker.service.task.Handler;
 import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2;
 
 import java.io.IOException;
@@ -22,7 +23,7 @@ import java.util.concurrent.BlockingQueue;
 
 @Service
 @Slf4j
-public class FidIdEnricherV2 implements DispatcherV2<SubjectsProfilePayloadV2, SendSignalsPayload>{
+public class FidIdEnricherV2 implements Handler<SubjectsProfilePayloadV2, SendSignalsPayload> {
     private final PersonService personService;
     private final SignalService signalService;
     private final CreditTrackerSettings config;
@@ -39,11 +40,11 @@ public class FidIdEnricherV2 implements DispatcherV2<SubjectsProfilePayloadV2, S
 
     @Override
     public SendSignalsPayload handle(SubjectsProfilePayloadV2 taskPayload) {
-        List<SubjectProfile> subjectsProfile = taskPayload.getSubjectProfiles();
+        List<SubjectProfileForSignal> subjectsProfile = taskPayload.getSubjectProfiles();
         log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
         personService.filledPassportsData(subjectsProfile);
         List<Person> personOnProcessing = subjectsProfile.stream()
-            .filter(SubjectProfile::isFilled)
+            .filter(SubjectProfileForSignal::isFilled)
             .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build())
             .toList();
         log.debug("personOnProcessing size: {}", personOnProcessing.size());

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

@@ -1,11 +1,11 @@
-package ru.nbch.credit_tracker.service.task;
+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.entities.dto.NameProfile;
+import ru.nbch.credit_tracker.model.NameData;
 import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
+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;
@@ -14,10 +14,11 @@ 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 DispatcherV2<PhoneFidPayloadV2, SubjectsProfilePayloadV2> {
+public class FidNameEnricherV2 implements Handler<PhoneFidPayloadV2, SubjectsProfilePayloadV2> {
     private final PhoneService2 phoneService2;
     private final PhoneFidMapMapper mapper;
     private final BlockingQueue<FlagsPayload> flagsPayloadQueue;
@@ -40,14 +41,14 @@ public class FidNameEnricherV2 implements DispatcherV2<PhoneFidPayloadV2, Subjec
         flagsPayloadQueue.add(new FlagsPayload(taskPayload.getPackageId(), phoneFidMap)); // TODO find out how to use it in other way
         //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать
 
-        List<SubjectProfile> subjectProfiles = phoneFidMap.stream()
+        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 -> {
-                    SubjectProfile subjectProfile = new SubjectProfile();
+                    SubjectProfileForSignal subjectProfile = new SubjectProfileForSignal();
                     subjectProfile.setId(pf.getId());
                     subjectProfile.setFid(pf.getFid());
-                    subjectProfile.setNameProfile(NameProfile.builder()
+                    subjectProfile.setNameProfile(NameData.builder()
                             .firstName(pf.getFirstName())
                             .middleName(pf.getMiddleName())
                             .lastName(pf.getLastName())

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

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service.task;
+package ru.nbch.credit_tracker.service.task.impl.dprctd;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -9,10 +9,11 @@ import ru.nbch.credit_tracker.model.FlagsPayload;
 import ru.nbch.credit_tracker.service.indic.FlagsService2;
 
 import java.util.List;
+import ru.nbch.credit_tracker.service.task.Handler;
 
 @Service
 @Slf4j
-public class FlagsEnricher implements DispatcherV2<FlagsPayload, DummyPayload>{
+public class FlagsEnricher implements Handler<FlagsPayload, DummyPayload> {
 
     private final FlagsService2 flagsService;
     private final PhoneFidMapMapper mapper;

src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java

@@ -101,4 +101,8 @@ public class CTUtil {
             log.warn("File {} hasn't been deleted", filePath);
         }
     }
+
+    public static String normalizePhoneNumber(String phoneNumber) {
+        return "9" + phoneNumber;
+    }
 }

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

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

src/test/java/ru/nbch/credit_tracker/service/PhoneService2Test.java

@@ -124,7 +124,7 @@ public class PhoneService2Test {
     void defineFidsAndNameByPhoneTest() {
         PhoneService2 phoneService = new PhoneService2(phoneRepository);
 
-        fidsDataWithNameEqualSinceDt.sort(byDbOrderWithName);
+//        fidsDataWithNameEqualSinceDt.sort(byDbOrderWithName);
         when(phoneRepository.findFidsWithName(any())).thenReturn(fidsDataWithNameEqualSinceDt);
 
         List<PhoneFidMap> phoneFidMap = new ArrayList<>();
@@ -172,9 +172,10 @@ public class PhoneService2Test {
             )
     );
 
-    Comparator<PhoneRepository.FidsDataWithName> byDbOrderWithName =
+    //todo понять зачем в тесте использовался этот компоратор
-            Comparator.comparing(PhoneRepository.FidsDataWithName::getPhone, Comparator.nullsFirst(Comparator.naturalOrder())) // pn.number ASC
+//    Comparator<PhoneRepository.FidsDataWithName> byDbOrderWithName =
-                    .thenComparing(PhoneRepository.FidsDataWithName::getFileSinceDate, Comparator.nullsLast(Comparator.reverseOrder())) // file_since_dt DESC
+//            Comparator.comparing(PhoneRepository.FidsDataWithName::getPhone, Comparator.nullsFirst(Comparator.naturalOrder())) // pn.number ASC
-                    .thenComparing(PhoneRepository.FidsDataWithName::getFid, Comparator.reverseOrder());
+//                    .thenComparing(PhoneRepository.FidsDataWithName::getFileSinceDate, Comparator.nullsLast(Comparator.reverseOrder())) // file_since_dt DESC
+//                    .thenComparing(PhoneRepository.FidsDataWithName::getFid, Comparator.reverseOrder());
 
 }