From 5d3507e12bfd888ccb3e0be74f6c14e641f035b9 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Thu, 30 Oct 2025 15:04:58 +0300 Subject: [PATCH] some refactoring :/ --- .../builder/PersonFactoryBuilderV2.java | 14 +- .../config/task/ConsumerConfigV2.java | 22 ++- .../config/task/StageQueueFactory.java | 24 +++ .../controller/SignalControllerV3.java | 65 +++++++ .../entities/dto/SubjectProfile.java | 14 -- .../entities/dto/SubjectProfileForSignal.java | 17 ++ .../credit_tracker/entities/indic/Name.java | 20 ++ .../credit_tracker/facade/SignalFacadeV3.java | 172 ++++++++++++++++++ .../mapper/signal/PhoneFidMapMapper.java | 2 +- .../credit_tracker/model/DummyPayload.java | 4 +- .../credit_tracker/model/FlagsPayload.java | 3 +- .../dto/IdProfile.java => model/IdData.java} | 5 +- .../NameProfile.java => model/NameData.java} | 5 +- .../model/PackageToSavePayloadV2.java | 4 +- .../model/PhoneFidPayloadV2.java | 3 +- .../model/SendSignalsPayload.java | 4 +- .../credit_tracker/model/StagePayload.java | 5 + .../credit_tracker/model/SubjectProfile.java | 17 ++ .../model/SubjectsProfilePayloadV2.java | 11 +- .../model/impl/SubjectPayload.java | 38 ++++ .../repository/PhoneRepository.java | 18 ++ .../service/indic/PersonService.java | 10 +- .../service/indic/PhoneService2.java | 61 ++++++- .../service/signal/PhoneFidMapService.java | 32 +++- .../service/task/Dispatcher.java | 5 - .../task/{DispatcherV2.java => Handler.java} | 2 +- .../service/task/PipelinePackageManager.java | 86 +++++++++ .../task/PipelinePackageProcessor.java | 62 ------- .../task/{TaskConsumerV2.java => Stage.java} | 16 +- .../service/task/StageRunnable.java | 135 ++++++++++++++ .../service/task/TaskPayloadV2.java | 5 - .../service/task/impl/SaveSubjectHandler.java | 64 +++++++ .../{ => impl/dprctd}/FidIdEnricherV2.java | 11 +- .../{ => impl/dprctd}/FidNameEnricherV2.java | 15 +- .../task/{ => impl/dprctd}/FlagsEnricher.java | 5 +- .../ru/nbch/credit_tracker/utils/CTUtil.java | 4 + .../mapper/signal/PhoneFidMapMapper.xml | 6 +- .../service/PhoneService2Test.java | 11 +- 38 files changed, 833 insertions(+), 164 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java create mode 100644 src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java create mode 100644 src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfileForSignal.java create mode 100644 src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java create mode 100644 src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java rename src/main/java/ru/nbch/credit_tracker/{entities/dto/IdProfile.java => model/IdData.java} (72%) rename src/main/java/ru/nbch/credit_tracker/{entities/dto/NameProfile.java => model/NameData.java} (71%) create mode 100644 src/main/java/ru/nbch/credit_tracker/model/StagePayload.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/Dispatcher.java rename src/main/java/ru/nbch/credit_tracker/service/task/{DispatcherV2.java => Handler.java} (66%) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageProcessor.java rename src/main/java/ru/nbch/credit_tracker/service/task/{TaskConsumerV2.java => Stage.java} (91%) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/StageRunnable.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/TaskPayloadV2.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java rename src/main/java/ru/nbch/credit_tracker/service/task/{ => impl/dprctd}/FidIdEnricherV2.java (89%) rename src/main/java/ru/nbch/credit_tracker/service/task/{ => impl/dprctd}/FidNameEnricherV2.java (84%) rename src/main/java/ru/nbch/credit_tracker/service/task/{ => impl/dprctd}/FlagsEnricher.java (90%) diff --git a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java index 5dddf84..9e1940e 100644 --- a/src/main/java/ru/nbch/credit_tracker/builder/PersonFactoryBuilderV2.java +++ b/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.entities.dto.NameProfile; -import ru.nbch.credit_tracker.entities.dto.SubjectProfile; +import ru.nbch.credit_tracker.model.IdData; +import ru.nbch.credit_tracker.model.NameData; +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(); - IdProfile passportData = personData.getIdProfile(); + NameData nameData = personData.getNameProfile(); + IdData passportData = personData.getIdProfile(); Person person = new Person(); person.setS(nameData.getLastName()); person.setN(nameData.getFirstName()); diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java b/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java index 618de65..78889af 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java +++ b/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); - TaskConsumerV2 fidIdEnricher = new TaskConsumerV2(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON); - TaskConsumerV2 flagsEnricherConsumer = new TaskConsumerV2(flagsEnricher, flagsPayloadQueue, new LinkedBlockingQueue(), FLAGS_POISON); + 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); @@ -60,4 +64,14 @@ public class ConsumerConfigV2 { thread3.start(); return 0; } + + @Bean + public StageQueueFactory saveSubjectQueueFactory() { + return new StageQueueFactory<>(); + } + + @Bean + public StageQueueFactory idEnricherQueueFactory() { + return new StageQueueFactory<>(); + } } diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java new file mode 100644 index 0000000..423708a --- /dev/null +++ b/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 implements FactoryBean> { + + @Override + public BlockingQueue getObject() throws Exception { + return new LinkedBlockingQueue<>(); + } + + @Override + public Class getObjectType() { + return BlockingQueue.class; + } + + @Override + public boolean isSingleton() { + return false; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java b/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java new file mode 100644 index 0000000..cddf566 --- /dev/null +++ b/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 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 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 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); + } +} + diff --git a/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java b/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java deleted file mode 100644 index f2cfc37..0000000 --- a/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfile.java +++ /dev/null @@ -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; -} diff --git a/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfileForSignal.java b/src/main/java/ru/nbch/credit_tracker/entities/dto/SubjectProfileForSignal.java new file mode 100644 index 0000000..a51f088 --- /dev/null +++ b/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; +} diff --git a/src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java b/src/main/java/ru/nbch/credit_tracker/entities/indic/Name.java new file mode 100644 index 0000000..93b1723 --- /dev/null +++ b/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; + +} diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java new file mode 100644 index 0000000..db761a1 --- /dev/null +++ b/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 findActivePackages(MonitoringActivePackagesRequest activePackagesRequest) throws Exception { + AuthorizationService.AuthResult authResult = authorizationService.findAuthInfo(activePackagesRequest.getUserId(), activePackagesRequest.getPassword()); + if (!authResult.isSuccess()) { + throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringActivePackagesError); + } + + List 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 findAllPackages(MonitoringAllPackagesRequest allPackagesRequest) throws Exception { + AuthorizationService.AuthResult authResult = authorizationService.findAuthInfo(allPackagesRequest.getUserId(), allPackagesRequest.getPassword()); + if (!authResult.isSuccess()) { + throw clientExcp(authResult.getError(), XmlErrorFormat.MonitoringAllPackagesError); + } + + List 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); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java index f78a0bc..6daadbd 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java +++ b/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 subjects, @Param("packageId") Long packageId); - void register(@Param("phoneFidMap") List phoneFidMap, @Param("packageId") Long packageId); + void register(@Param("phoneFidMap") List phoneFidMap); void registerWithId(@Param("phoneFidMap") List phoneFidMap); List findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("offset") int offset, @Param("limit") int limit); diff --git a/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java b/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java index b8166d7..ed6e5da 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java +++ b/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 TaskPayloadV2 { +public class DummyPayload implements StagePayload { private final boolean poison; diff --git a/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java b/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java index f58c386..ed4f160 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java +++ b/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 phoneFidMaps; private final boolean poison; diff --git a/src/main/java/ru/nbch/credit_tracker/entities/dto/IdProfile.java b/src/main/java/ru/nbch/credit_tracker/model/IdData.java similarity index 72% rename from src/main/java/ru/nbch/credit_tracker/entities/dto/IdProfile.java rename to src/main/java/ru/nbch/credit_tracker/model/IdData.java index 3be4e8b..7f98ac8 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/dto/IdProfile.java +++ b/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; diff --git a/src/main/java/ru/nbch/credit_tracker/entities/dto/NameProfile.java b/src/main/java/ru/nbch/credit_tracker/model/NameData.java similarity index 71% rename from src/main/java/ru/nbch/credit_tracker/entities/dto/NameProfile.java rename to src/main/java/ru/nbch/credit_tracker/model/NameData.java index ea27e2f..076b887 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/dto/NameProfile.java +++ b/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; diff --git a/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java index 640dd7a..c17d10e 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java +++ b/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 TaskPayloadV2 { +public class PackageToSavePayloadV2 implements StagePayload { private final Long packageId; private final String filePath; private final boolean poison; diff --git a/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java index 52aaeb2..3900a38 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java +++ b/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 phoneFidsBatch; private final boolean poison; diff --git a/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java b/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java index 3f72c73..d3c64e3 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java +++ b/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 TaskPayloadV2 { +public class SendSignalsPayload implements StagePayload { private String signalXmlPath; private boolean poison; diff --git a/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java b/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java new file mode 100644 index 0000000..5fbd1e1 --- /dev/null +++ b/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(); +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java b/src/main/java/ru/nbch/credit_tracker/model/SubjectProfile.java new file mode 100644 index 0000000..d284669 --- /dev/null +++ b/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; + +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/SubjectsProfilePayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/SubjectsProfilePayloadV2.java index 6cf8eb7..bede3c9 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/SubjectsProfilePayloadV2.java +++ b/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.service.task.TaskPayloadV2; +import ru.nbch.credit_tracker.entities.dto.SubjectProfileForSignal; import java.util.List; -public class SubjectsProfilePayloadV2 implements TaskPayloadV2 { +public class SubjectsProfilePayloadV2 implements StagePayload { private final Long packageId; - private final List subjectProfiles; + private final List 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 subjectProfiles) { + public SubjectsProfilePayloadV2(Long packageId, List subjectProfiles) { this.packageId = packageId; this.subjectProfiles = subjectProfiles; this.poison = false; @@ -26,7 +25,7 @@ public class SubjectsProfilePayloadV2 implements TaskPayloadV2 { return packageId; } - public List getSubjectProfiles() { + public List getSubjectProfiles() { return subjectProfiles; } diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java new file mode 100644 index 0000000..8f9cd24 --- /dev/null +++ b/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 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 subjects) { + this.packageId = packageId; + this.subjects = subjects; + this.poison = false; + } + + public Long getPackageId() { + return packageId; + } + + public List getSubjects() { + return subjects; + } + + @Override + public boolean isPoison() { + return poison; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java index c63c754..b30dab4 100644 --- a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java +++ b/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 findFidsWithName(List phones) { MapSqlParameterSource params = new MapSqlParameterSource(); params.addValue("phones", phones); @@ -86,6 +88,22 @@ public class PhoneRepository { )); } + public List findNameByPhones(List 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; diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java index c5d35cd..4a70b83 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonService.java +++ b/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.entities.dto.SubjectProfile; +import ru.nbch.credit_tracker.model.IdData; +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 subjectProfiles) { - List fids = subjectProfiles.stream().map(SubjectProfile::getFid).collect(Collectors.toList()); + public void filledPassportsData(List subjectProfiles) { + List 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()) diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java index cf691c3..40d9326 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java +++ b/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 java.time.LocalDate; -import java.util.*; -import java.util.stream.Collectors; +import ru.nbch.credit_tracker.utils.CTUtil; @Slf4j @Service @@ -111,6 +119,30 @@ public class PhoneService2 { log.debug("End merging fids and names"); } + Comparator 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 findNameByNumber(List subjects) { + List phones = subjects.stream() + .map(subject -> CTUtil.normalizePhoneNumber(subject.getPhone())) + .toList(); + + List 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 dataList) { PhoneRepository.FidsDataWithName currentData = null; for (PhoneRepository.FidsDataWithName fidsData : dataList) { @@ -191,6 +224,26 @@ public class PhoneService2 { return currentData; } + private Name defineNameDataV2(List 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 merger(List subjects, List phonesByFid) { Map phoneFidMap = new HashMap<>(); Iterator iterator = phonesByFid.iterator(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java index 3962e5c..1e8e022 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PhoneFidMapService.java @@ -2,35 +2,61 @@ 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 subjects, Long packageId) { mapper.registerSubjects(subjects, packageId); } - public void register(List phoneFidMap, Long packageId) { - mapper.register(phoneFidMap, packageId); + /** + * Регистрирует в базе список субъектов переданных в запросе + * + * @param subjects - список субъектов + * @param packageId - уникальный идентификатор пакета + */ + public void register(List subjects, Map nameByPhone, Long packageId) { + List 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 findFullPhoneFidMap(Long packageId, int offset, int limit) { + public List findFullPhoneFidMap(Long packageId, int offset, int limit) { return mapper.findFullPhoneFidMap(packageId, offset, limit); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Dispatcher.java b/src/main/java/ru/nbch/credit_tracker/service/task/Dispatcher.java deleted file mode 100644 index 1e46e57..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Dispatcher.java +++ /dev/null @@ -1,5 +0,0 @@ -package ru.nbch.credit_tracker.service.task; - -public interface Dispatcher { - void handle(X taskPayload); -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/DispatcherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java similarity index 66% rename from src/main/java/ru/nbch/credit_tracker/service/task/DispatcherV2.java rename to src/main/java/ru/nbch/credit_tracker/service/task/Handler.java index 063c515..e4819b9 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/DispatcherV2.java +++ b/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 { +public interface Handler { Z handle(X taskPayload); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java new file mode 100644 index 0000000..d9fc602 --- /dev/null +++ b/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 saveSubjectQueueFactory; + private final StageQueueFactory idEnricherQueueFactory; + private final ExecutorService executorService = Executors.newFixedThreadPool(10); + private final static int BATCH_SIZE = 10_000; + + public PipelinePackageManager(ApplicationContext applicationContext, + StaxXmlProcessor xmlProcessor, + StageQueueFactory saveSubjectQueueFactory, + StageQueueFactory 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 saveSubjectQueue = saveSubjectQueueFactory.getObject(); + BlockingQueue idEnricherQueue = idEnricherQueueFactory.getObject(); + + StageRunnable 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 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); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageProcessor.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageProcessor.java deleted file mode 100644 index ee86b06..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageProcessor.java +++ /dev/null @@ -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 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 = 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); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumerV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java similarity index 91% rename from src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumerV2.java rename to src/main/java/ru/nbch/credit_tracker/service/task/Stage.java index 36d840c..83166ce 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/TaskConsumerV2.java +++ b/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 +public class Stage { - private final DispatcherV2 dispatcher; + private final Handler dispatcher; private final BlockingQueue in; private final BlockingQueue out; private final Z POISON; private final ExecutorService ioPool; - public TaskConsumerV2(DispatcherV2 dispatcher, - BlockingQueue in, - BlockingQueue out, - Z POISON) { + public Stage(Handler dispatcher, + BlockingQueue in, + BlockingQueue 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> inflight = new ArrayList<>(); AtomicInteger inFlightCount = new AtomicInteger(); - try { while (true) { X b = takeQuiet(in, Duration.ofSeconds(5)); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/StageRunnable.java b/src/main/java/ru/nbch/credit_tracker/service/task/StageRunnable.java new file mode 100644 index 0000000..3be4c35 --- /dev/null +++ b/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 implements Runnable { + private final Handler dispatcher; + private final BlockingQueue in; + private final BlockingQueue out; + private final Z POISON; + private final ExecutorService ioPool; + + public StageRunnable(Handler dispatcher, + BlockingQueue in, + BlockingQueue 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> 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 f : inflight) { + putQuiet(out, f.get()); + } + putQuiet(out, POISON); + return; + } + + Future 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> fs, BlockingQueue out) throws Exception { + Iterator> it = fs.iterator(); + while (it.hasNext()) { + Future f = it.next(); + if (f.isDone()) { + putQuiet(out, f.get()); + it.remove(); + } + } + } + + void drainAll(List> fs, BlockingQueue out) { + for (Future 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 q, Z b) { + try { + q.put(b); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + X takeQuiet(BlockingQueue 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; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/TaskPayloadV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/TaskPayloadV2.java deleted file mode 100644 index ed6e5c8..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/TaskPayloadV2.java +++ /dev/null @@ -1,5 +0,0 @@ -package ru.nbch.credit_tracker.service.task; - -public interface TaskPayloadV2 { - boolean isPoison(); -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java new file mode 100644 index 0000000..e89fca3 --- /dev/null +++ b/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 { + 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 subjects = taskPayload.getSubjects(); + Long packageId = taskPayload.getPackageId(); + log.debug("SaveSubjectStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size()); + + Map 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); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidIdEnricherV2.java similarity index 89% rename from src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricherV2.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidIdEnricherV2.java index 90dcd96..43b92f8 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/FidIdEnricherV2.java +++ b/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{ +public class FidIdEnricherV2 implements Handler { private final PersonService personService; private final SignalService signalService; private final CreditTrackerSettings config; @@ -39,11 +40,11 @@ public class FidIdEnricherV2 implements DispatcherV2 subjectsProfile = taskPayload.getSubjectProfiles(); + List subjectsProfile = taskPayload.getSubjectProfiles(); log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); personService.filledPassportsData(subjectsProfile); List personOnProcessing = subjectsProfile.stream() - .filter(SubjectProfile::isFilled) + .filter(SubjectProfileForSignal::isFilled) .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build()) .toList(); log.debug("personOnProcessing size: {}", personOnProcessing.size()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidNameEnricherV2.java similarity index 84% rename from src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidNameEnricherV2.java index ed9be88..d80f5d5 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/FidNameEnricherV2.java +++ b/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 { +public class FidNameEnricherV2 implements Handler { private final PhoneService2 phoneService2; private final PhoneFidMapMapper mapper; private final BlockingQueue flagsPayloadQueue; @@ -40,14 +41,14 @@ public class FidNameEnricherV2 implements DispatcherV2 subjectProfiles = phoneFidMap.stream() + List 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()) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java similarity index 90% rename from src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java index b65f1f7..2a06769 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/FlagsEnricher.java +++ b/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{ +public class FlagsEnricher implements Handler { private final FlagsService2 flagsService; private final PhoneFidMapMapper mapper; diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java index 0383d4d..46c8bb9 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java +++ b/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; + } } diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index aacf6a6..7f8dfc7 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -25,11 +25,11 @@ - - INSERT INTO signals_phone_fid_map (subject_id, phone, package_id) + + INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id) values - (#{phoneFid.subjectId}, #{phoneFid.phone}, #{packageId}) + (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}) diff --git a/src/test/java/ru/nbch/credit_tracker/service/PhoneService2Test.java b/src/test/java/ru/nbch/credit_tracker/service/PhoneService2Test.java index 87389a4..95e9ff9 100644 --- a/src/test/java/ru/nbch/credit_tracker/service/PhoneService2Test.java +++ b/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 = new ArrayList<>(); @@ -172,9 +172,10 @@ public class PhoneService2Test { ) ); - Comparator byDbOrderWithName = - Comparator.comparing(PhoneRepository.FidsDataWithName::getPhone, Comparator.nullsFirst(Comparator.naturalOrder())) // pn.number ASC - .thenComparing(PhoneRepository.FidsDataWithName::getFileSinceDate, Comparator.nullsLast(Comparator.reverseOrder())) // file_since_dt DESC - .thenComparing(PhoneRepository.FidsDataWithName::getFid, Comparator.reverseOrder()); + //todo понять зачем в тесте использовался этот компоратор +// Comparator byDbOrderWithName = +// Comparator.comparing(PhoneRepository.FidsDataWithName::getPhone, Comparator.nullsFirst(Comparator.naturalOrder())) // pn.number ASC +// .thenComparing(PhoneRepository.FidsDataWithName::getFileSinceDate, Comparator.nullsLast(Comparator.reverseOrder())) // file_since_dt DESC +// .thenComparing(PhoneRepository.FidsDataWithName::getFid, Comparator.reverseOrder()); }