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 deleted file mode 100644 index 78889af..0000000 --- a/src/main/java/ru/nbch/credit_tracker/config/task/ConsumerConfigV2.java +++ /dev/null @@ -1,77 +0,0 @@ -package ru.nbch.credit_tracker.config.task; - -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import ru.nbch.credit_tracker.model.*; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.task.*; - -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; -import ru.nbch.credit_tracker.service.task.impl.dprctd.FidIdEnricherV2; -import ru.nbch.credit_tracker.service.task.impl.dprctd.FidNameEnricherV2; -import ru.nbch.credit_tracker.service.task.impl.dprctd.FlagsEnricher; - -@Configuration -public class ConsumerConfigV2 { - - @Bean - public BlockingQueue savingRequest2() { - return new LinkedBlockingQueue<>(); - } - - @Bean - public BlockingQueue flagsPayload() { - return new LinkedBlockingQueue<>(); - } - - @Bean - public BlockingQueue enricherQueue2() { - return new LinkedBlockingQueue<>(); - } - - @Bean - public BlockingQueue enricherIdQueue2() { - return new LinkedBlockingQueue<>(); - } - - @Bean - public StagePayload POISON() { - return () -> true; - } - - @SuppressWarnings("unchecked") - @Bean - public Integer consumerConfigurer(BlockingQueue savingRequest2, - BlockingQueue enricherQueue2, - BlockingQueue enricherIdQueue2, - BlockingQueue flagsPayloadQueue, - FidNameEnricherV2 fidNameEnricherV2, - FidIdEnricherV2 fidIdEnricherV2, - FlagsEnricher flagsEnricher) { - - SubjectsProfilePayloadV2 POISON = new SubjectsProfilePayloadV2(true); - SendSignalsPayload FID_ID_ENRICHER_POISON = new SendSignalsPayload(true); - DummyPayload FLAGS_POISON = new DummyPayload(true); - Stage fidNameEnricher = new Stage(fidNameEnricherV2, savingRequest2, enricherQueue2, POISON); - Stage fidIdEnricher = new Stage(fidIdEnricherV2, enricherQueue2, enricherIdQueue2, FID_ID_ENRICHER_POISON); - Stage flagsEnricherConsumer = new Stage(flagsEnricher, flagsPayloadQueue, new LinkedBlockingQueue(), FLAGS_POISON); - Thread thread1 = new Thread(fidNameEnricher::start); - Thread thread2 = new Thread(fidIdEnricher::start); - Thread thread3 = new Thread(flagsEnricherConsumer::start); - thread1.start(); - thread2.start(); - thread3.start(); - return 0; - } - - @Bean - public StageQueueFactory saveSubjectQueueFactory() { - return new StageQueueFactory<>(); - } - - @Bean - public StageQueueFactory idEnricherQueueFactory() { - return new StageQueueFactory<>(); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java new file mode 100644 index 0000000..f2102d7 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java @@ -0,0 +1,19 @@ +package ru.nbch.credit_tracker.config.task; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; + +@Configuration +public class StageConfig { + + @Bean("saveSubjectQueueFactory") + public StageQueueFactory saveSubjectQueueFactory() { + return new StageQueueFactory<>(); + } + + @Bean("idEnricherQueueFactory") + public StageQueueFactory idEnricherQueueFactory() { + return new StageQueueFactory<>(); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java b/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java index cddf566..7a86f42 100644 --- a/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java +++ b/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java @@ -16,29 +16,29 @@ import ru.nbch.credit_tracker.entities.user_response.success.MonitoringSetupResp import ru.nbch.credit_tracker.enums.SignalError; import ru.nbch.credit_tracker.enums.XmlErrorFormat; import ru.nbch.credit_tracker.exceptions.SignalException; -import ru.nbch.credit_tracker.facade.SignalFacadeV2; +import ru.nbch.credit_tracker.facade.SignalFacadeV3; import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp; @RestController @RequestMapping("/api/v3/signal") public class SignalControllerV3 { - private final SignalFacadeV2 signalFacadeV2; + private final SignalFacadeV3 signalFacadeV3; - public SignalControllerV3(SignalFacadeV2 signalFacadeV2) { - this.signalFacadeV2 = signalFacadeV2; + public SignalControllerV3(SignalFacadeV3 signalFacadeV3) { + this.signalFacadeV3 = signalFacadeV3; } @PostMapping(value = "/set", produces = MediaType.APPLICATION_XML_VALUE) public ResponseEntity setupMonitoring(@RequestParam("file") MultipartFile file) { - signalFacadeV2.setupMonitoringRequest(file); + signalFacadeV3.setupMonitoringRequest(file); return ResponseEntity.ok().body(new MonitoringSetupResponse()); } @PostMapping(value = "/active_packages", consumes = MediaType.APPLICATION_XML_VALUE, produces = MediaType.APPLICATION_XML_VALUE) public ResponseEntity findActivePackages(@RequestBody MonitoringActivePackagesRequest activePackagesRequest) { try { - return signalFacadeV2.findActivePackages(activePackagesRequest); + return signalFacadeV3.findActivePackages(activePackagesRequest); } catch (Exception e) { throwOrWrap(SignalError.ERROR_013, XmlErrorFormat.MonitoringActivePackagesError, e); return null; @@ -48,7 +48,7 @@ public class SignalControllerV3 { @PostMapping(value = "/all_packages", consumes = MediaType.APPLICATION_XML_VALUE, produces = MediaType.APPLICATION_XML_VALUE) public ResponseEntity findAllPackages(@RequestBody MonitoringAllPackagesRequest allPackagesRequest) { try { - return signalFacadeV2.findAllPackages(allPackagesRequest); + return signalFacadeV3.findAllPackages(allPackagesRequest); } catch (Exception e) { throwOrWrap(SignalError.ERROR_013, XmlErrorFormat.MonitoringAllPackagesError, e); return null; diff --git a/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java deleted file mode 100644 index c17d10e..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/PackageToSavePayloadV2.java +++ /dev/null @@ -1,31 +0,0 @@ -package ru.nbch.credit_tracker.model; - -public class PackageToSavePayloadV2 implements StagePayload { - private final Long packageId; - private final String filePath; - private final boolean poison; - - public PackageToSavePayloadV2(boolean poison) { - this.packageId = null; - this.filePath = null; - this.poison = true; - } - public PackageToSavePayloadV2(Long packageId, String filePath) { - this.packageId = packageId; - this.filePath = filePath; - this.poison = false; - } - - public Long getPackageId() { - return packageId; - } - - public String getFilePath() { - return filePath; - } - - @Override - public boolean isPoison() { - return poison; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java b/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java deleted file mode 100644 index 3900a38..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/PhoneFidPayloadV2.java +++ /dev/null @@ -1,36 +0,0 @@ -package ru.nbch.credit_tracker.model; - -import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; - -import java.util.List; - -public class PhoneFidPayloadV2 implements StagePayload { - private final Long packageId; - private final List phoneFidsBatch; - private final boolean poison; - - public PhoneFidPayloadV2(boolean poison) { - this.packageId = null; - this.phoneFidsBatch = null; - this.poison = poison; - } - - public PhoneFidPayloadV2(Long packageId, List phoneFidsBatch) { - this.packageId = packageId; - this.phoneFidsBatch = phoneFidsBatch; - this.poison = false; - } - - public Long getPackageId() { - return packageId; - } - - public List getPhoneFidsBatch() { - return phoneFidsBatch; - } - - @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 b30dab4..d1b00b1 100644 --- a/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java +++ b/src/main/java/ru/nbch/credit_tracker/repository/PhoneRepository.java @@ -46,12 +46,7 @@ public class PhoneRepository { FROM PHONE_NORM pn JOIN NAME nf ON nf.fid = pn.fid - WHERE number in (:phones) - ORDER BY - pn.number ASC, - pn.file_since_dt DESC, - pn.fid DESC - WITH UR + WHERE number in (:phones) WITH UR """; private final NamedParameterJdbcTemplate jdbcTemplate; 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 40d9326..efb33ec 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 @@ -119,8 +119,8 @@ public class PhoneService2 { log.debug("End merging fids and names"); } - Comparator byDateThenId = - Comparator.comparing(Name::getFileSinceDate) // LocalDate natural order + Comparator byDateThenIdNullSafe = + Comparator.comparing(Name::getFileSinceDate, Comparator.nullsFirst(Comparator.naturalOrder())) .thenComparing(Name::getFid); /** @@ -140,7 +140,7 @@ public class PhoneService2 { .collect(Collectors.toMap( Name::getNumber, Function.identity(), - BinaryOperator.maxBy(byDateThenId) + BinaryOperator.maxBy(byDateThenIdNullSafe) )); } /** 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 1e8e022..e3b8fe7 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 @@ -14,6 +14,7 @@ import ru.nbch.credit_tracker.entities.indic.Name; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject; import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.utils.CTUtil; @Service public class PhoneFidMapService { @@ -41,7 +42,7 @@ public class PhoneFidMapService { public void register(List subjects, Map nameByPhone, Long packageId) { List phoneFidMaps = subjects.stream().map(subjectProfile -> { Integer fid = Optional - .of(nameByPhone.get(subjectProfile.getPhone())) + .ofNullable(nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone()))) .map(Name::getFid) .orElse(null); 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 index d9fc602..2a1601c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java @@ -9,9 +9,9 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.task.StageQueueFactory; @@ -33,8 +33,8 @@ public class PipelinePackageManager { public PipelinePackageManager(ApplicationContext applicationContext, StaxXmlProcessor xmlProcessor, - StageQueueFactory saveSubjectQueueFactory, - StageQueueFactory idEnricherQueueFactory) { + @Qualifier("saveSubjectQueueFactory") StageQueueFactory saveSubjectQueueFactory, + @Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory) { this.applicationContext = applicationContext; this.xmlProcessor = xmlProcessor; this.saveSubjectQueueFactory = saveSubjectQueueFactory; @@ -46,7 +46,7 @@ public class PipelinePackageManager { BlockingQueue saveSubjectQueue = saveSubjectQueueFactory.getObject(); BlockingQueue idEnricherQueue = idEnricherQueueFactory.getObject(); - StageRunnable saveSubjectStage = new StageRunnable<>(saveSubjectHandler, + Stage saveSubjectStage = new Stage<>(saveSubjectHandler, saveSubjectQueue, idEnricherQueue, SubjectPayload.POISON @@ -80,7 +80,7 @@ public class PipelinePackageManager { saveSubjectQueue.add(new SubjectPayload(true)); } }); - future.get(10_000, TimeUnit.SECONDS); +// future.get(10_000, TimeUnit.SECONDS); log.info("Subjects were registered. PackageId: {}", packageId); } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java index 83166ce..f6ba888 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java @@ -15,8 +15,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils; import ru.nbch.credit_tracker.model.StagePayload; @Slf4j -public class Stage -{ +public class Stage implements Runnable { private final Handler dispatcher; private final BlockingQueue in; private final BlockingQueue out; @@ -34,7 +33,8 @@ public class Stage this.POISON = POISON; } - public void start() { + @Override + public void run() { List> inflight = new ArrayList<>(); AtomicInteger inFlightCount = new AtomicInteger(); try { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidNameEnricherV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidNameEnricherV2.java deleted file mode 100644 index d80f5d5..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidNameEnricherV2.java +++ /dev/null @@ -1,67 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl.dprctd; - -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.model.NameData; -import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; -import ru.nbch.credit_tracker.entities.dto.SubjectProfileForSignal; -import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; -import ru.nbch.credit_tracker.model.FlagsPayload; -import ru.nbch.credit_tracker.model.PhoneFidPayloadV2; -import ru.nbch.credit_tracker.model.SubjectsProfilePayloadV2; -import ru.nbch.credit_tracker.service.indic.PhoneService2; - -import java.util.List; -import java.util.concurrent.BlockingQueue; -import ru.nbch.credit_tracker.service.task.Handler; - -@Service -@Slf4j -public class FidNameEnricherV2 implements Handler { - private final PhoneService2 phoneService2; - private final PhoneFidMapMapper mapper; - private final BlockingQueue flagsPayloadQueue; - - public FidNameEnricherV2(PhoneService2 phoneService2, - PhoneFidMapMapper mapper, - BlockingQueue flagsPayloadQueue) { - this.phoneService2 = phoneService2; - this.mapper = mapper; - this.flagsPayloadQueue = flagsPayloadQueue; - } - - @Override - public SubjectsProfilePayloadV2 handle(PhoneFidPayloadV2 taskPayload) { - List phoneFidMap = taskPayload.getPhoneFidsBatch(); - log.debug("FidNameEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMap.size()); - mapper.registerWithId(phoneFidMap); // надо зарегистрировать phoneFidMap'ы т.к далее id участвует в формированиии запроса в оф. сервис uid = id - phoneService2.defineFidsAndNameByPhone2(phoneFidMap); - log.debug("Found {} fids by batch", phoneFidMap.stream().filter(s -> s.getFid() != null).count()); - flagsPayloadQueue.add(new FlagsPayload(taskPayload.getPackageId(), phoneFidMap)); // TODO find out how to use it in other way - //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать - - List subjectProfiles = phoneFidMap.stream() - .filter(subject -> subject.getFid() != null) - .filter(this::requiredFieldPresent) // TODO log.warn("Person does not have required fields. Fid {}", fid); - .map(pf -> { - SubjectProfileForSignal subjectProfile = new SubjectProfileForSignal(); - subjectProfile.setId(pf.getId()); - subjectProfile.setFid(pf.getFid()); - subjectProfile.setNameProfile(NameData.builder() - .firstName(pf.getFirstName()) - .middleName(pf.getMiddleName()) - .lastName(pf.getLastName()) - .birthDate(pf.getBirthDate()) - .fileSinceDate(pf.getFileSinceDate()) - .build()); - return subjectProfile; - }).toList(); - - return new SubjectsProfilePayloadV2(taskPayload.getPackageId(), subjectProfiles); - } - - private boolean requiredFieldPresent(PhoneFidMap phoneFidMap) { - return !StringUtils.isAllBlank(phoneFidMap.getFirstName(), phoneFidMap.getLastName()) && phoneFidMap.getBirthDate() != null; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java index 2a06769..f5e2185 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java @@ -28,7 +28,7 @@ public class FlagsEnricher implements Handler { public DummyPayload handle(FlagsPayload taskPayload) { List phoneFidMaps = taskPayload.getPhoneFidMaps(); log.debug("FlagsEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMaps.size()); - flagsService.fillFlagsData(phoneFidMaps); +// flagsService.fillFlagsData(phoneFidMaps); log.debug("Start updating subject's batch"); if (!phoneFidMaps.isEmpty()) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/StageRunnable.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/StageForDeletion.java similarity index 90% rename from src/main/java/ru/nbch/credit_tracker/service/task/StageRunnable.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/StageForDeletion.java index 3be4c35..d17bcf7 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/StageRunnable.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/StageForDeletion.java @@ -1,4 +1,4 @@ -package ru.nbch.credit_tracker.service.task; +package ru.nbch.credit_tracker.service.task.impl.dprctd; import java.time.Duration; import java.util.ArrayList; @@ -13,19 +13,21 @@ import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import ru.nbch.credit_tracker.model.StagePayload; +import ru.nbch.credit_tracker.service.task.Handler; @Slf4j -public class StageRunnable implements Runnable { +public class StageForDeletion +{ 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) { + public StageForDeletion(Handler dispatcher, + BlockingQueue in, + BlockingQueue out, + Z POISON) { this.dispatcher = dispatcher; this.in = in; this.out = out; @@ -33,8 +35,7 @@ public class StageRunnable imple this.POISON = POISON; } - @Override - public void run() { + public void start() { List> inflight = new ArrayList<>(); AtomicInteger inFlightCount = new AtomicInteger(); try { diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 7f8dfc7..20b7462 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -25,7 +25,7 @@ - + INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id) values