diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java index 221c86a..d0a9b05 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java @@ -6,9 +6,11 @@ import java.util.function.Supplier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineImpl; import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; @@ -18,8 +20,11 @@ import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @Configuration public class PipelineStarterConfig { @Bean - public BiFunction> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) { - return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, creditTrackerSettings); + public BiFunction> pipelineStarterFactory(StaxXmlProcessor xml, + CreditTrackerSettings creditTrackerSettings, + UserPackageMapper userPackageMapper, + PhoneService2 phoneService2) { + return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, userPackageMapper, phoneService2, creditTrackerSettings); } @Bean 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 index 0c5e988..c38ce62 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java @@ -10,17 +10,22 @@ public class StageConfig { @Bean("saveSubjectQueueFactory") public StageQueueFactory saveSubjectQueueFactory() { - return new StageQueueFactory<>(5); + return new StageQueueFactory<>(50); + } + + @Bean("nameEnricherQueueFactory") + public StageQueueFactory nameEnricherQueueFactory() { + return new StageQueueFactory<>(20); } @Bean("idEnricherQueueFactory") public StageQueueFactory idEnricherQueueFactory() { - return new StageQueueFactory<>(); + return new StageQueueFactory<>(20); } @Bean("partFileQueueFactory") public StageQueueFactory partFileQueueFactory() { - return new StageQueueFactory<>(10000); + return new StageQueueFactory<>(100); } } 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 f23ceb5..3ca07f4 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 @@ -14,6 +14,7 @@ public interface PhoneFidMapMapper { void registerSubjects(@Param("subjects") List subjects, @Param("packageId") Long packageId); void register(@Param("phoneFidMap") List phoneFidMap); List registerJson(@Param("json") String json); + void setFidJson(@Param("json") String json); 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/mapper/signal/UserPackageMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java index ad23c67..8134408 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java @@ -26,6 +26,7 @@ public interface UserPackageMapper { void updatePackageLastDownloadedReport(@Param("id") Long id, @Param("fileName") String fileName); void updatePackageOnMonitoringSuccess(@Param("id") Long id, @Param("monitoringAt") LocalDateTime monitoringAt); + void setLastSearchedDt(@Param("id") Long id); UserPackage findPackage(@Param("id") Long id); 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 index f9e4b1e..60938a3 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java @@ -8,18 +8,24 @@ import ru.nbch.credit_tracker.model.SubjectProfile; public class SubjectPayload implements StagePayload { private final Long packageId; + private final String memberCode; + private final String packageName; private final List subjects; - public static final SubjectPayload POISON = new SubjectPayload(PoisonType.NORMALLY); + public static final SubjectPayload POISON = new SubjectPayload(null, PoisonType.NORMALLY); private final PoisonType poison; - public SubjectPayload(PoisonType poison) { - this.packageId = null; + public SubjectPayload(Long packageId, PoisonType poison) { + this.memberCode = null; + this.packageName = null; + this.packageId = packageId; this.subjects = null; this.poison = poison; } - public SubjectPayload(Long packageId, List subjects) { + public SubjectPayload(Long packageId, String memberCode, String packageName, List subjects) { this.packageId = packageId; + this.memberCode = memberCode; + this.packageName = packageName; this.subjects = subjects; this.poison = PoisonType.NONE; } @@ -32,6 +38,14 @@ public class SubjectPayload implements StagePayload { return subjects; } + public String getMemberCode() { + return memberCode; + } + + public String getPackageName() { + return packageName; + } + @Override public boolean isPoison() { return poison == PoisonType.NORMALLY; diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java b/src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java index 9ac947a..62d31ac 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java @@ -32,7 +32,7 @@ public class SignalServiceV2 { .userId(userId) .packageName(packageName) .createdAt(LocalDateTime.now()) - .status(Status.PROCESSING.name()) + .status(Status.ACCEPTED.name()) .build(); return userPackageService.insertPackage(userPackage); } 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 530bffc..8edd9ed 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 @@ -65,14 +65,29 @@ public class PhoneService2 { String json = entityToJsonMapper.writeValueAsString(subjects.stream() .map(phoneFidMap -> new HashMap() {{ put("subject_id", phoneFidMap.getSubjectId()); - put("fid", phoneFidMap.getFid()); +// put("fid", phoneFidMap.getFid()); put("phone", phoneFidMap.getPhone()); put("package_id", phoneFidMap.getPackageId()); - put("calculated_at", phoneFidMap.getCalculatedAt()); +// put("calculated_at", phoneFidMap.getCalculatedAt()); }}).toList()); return phoneFidMapMapper.registerJson(json); } catch (JsonProcessingException e) { throw new RuntimeException(e); } } + + public void updateFid(List subjects) { + try { + //fixme, временное решение + String json = entityToJsonMapper.writeValueAsString(subjects.stream() + .map(phoneFidMap -> new HashMap() {{ + put("id", phoneFidMap.getId()); + put("fid", phoneFidMap.getFid()); + put("calculated_at", phoneFidMap.getCalculatedAt()); + }}).toList()); + phoneFidMapMapper.setFidJson(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java index 942e5ad..4b7d615 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java @@ -2,6 +2,9 @@ package ru.nbch.credit_tracker.service.task; public interface Handler { Z handle(X taskPayload); + default void handleOnPoison(X poison){ + //do nothing + } default int preferredConcurrency() {return 10;} Z cookPoison(); 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 1032b1d..880111e 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 @@ -12,11 +12,13 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.task.StageQueueFactory; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; +import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler; import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator; @@ -32,9 +34,12 @@ public class PipelinePackageManager { private final FlagsEnricher flagsEnricherV2; private final StageQueueFactory saveSubjectQueueFactory; private final StageQueueFactory idEnricherQueueFactory; + private final StageQueueFactory nameEnricherQueueFactory; private final StageQueueFactory sendSignalQueueFactory; private final SaveSubjectHandler saveSubjectHandler; + private final UserPackageMapper userPackageMapper; private final FidIdEnricher fidIdEnricherHandler; + private final FidNameEnricher fidNameEnricher; private final Supplier> pipelineTerminatorFactory; private final BiFunction> pipelineStarterFactory; private final CreditTrackerSettings creditTrackerSettings; @@ -50,22 +55,28 @@ public class PipelinePackageManager { public PipelinePackageManager(SignalRestService signalRestService, @Qualifier("saveSubjectQueueFactory") StageQueueFactory saveSubjectQueueFactory, @Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, + @Qualifier("nameEnricherQueueFactory") StageQueueFactory nameEnricherQueueFactory, @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, - SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, + SaveSubjectHandler saveSubjectHandler, + UserPackageMapper userPackageMapper, + FidIdEnricher fidIdEnricherHandler, Supplier> pipelineTerminatorFactory, BiFunction> pipelineStarterFactory, - FlagsEnricher flagsEnricherV2, + FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher, CreditTrackerSettings creditTrackerSettings, PipelineBackoff pipelineBackoff) { this.signalRestService = signalRestService; this.saveSubjectQueueFactory = saveSubjectQueueFactory; this.idEnricherQueueFactory = idEnricherQueueFactory; + this.nameEnricherQueueFactory = nameEnricherQueueFactory; this.sendSignalQueueFactory = sendSignalQueueFactory; this.saveSubjectHandler = saveSubjectHandler; + this.userPackageMapper = userPackageMapper; this.fidIdEnricherHandler = fidIdEnricherHandler; this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.pipelineStarterFactory = pipelineStarterFactory; this.flagsEnricherV2 = flagsEnricherV2; + this.fidNameEnricher = fidNameEnricher; this.creditTrackerSettings = creditTrackerSettings; this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); this.pipelineBackoff = pipelineBackoff; @@ -79,6 +90,9 @@ public class PipelinePackageManager { try { Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) .handler(saveSubjectHandler) + .andThen() + .stage(nameEnricherQueueFactory.getObject()) + .handler(fidNameEnricher) .handlerResult(flagsEnricherV2) .andThen() .stage(sendSignalQueueFactory.getObject()) @@ -103,6 +117,7 @@ public class PipelinePackageManager { log.error(ExceptionUtils.getStackTrace(e)); return; } + userPackageMapper.setLastSearchedDt(packageId); SignalXmlFilePayload res = (SignalXmlFilePayload) v; log.trace("Signal file by path: {} is completed", res.getSignalXmlFilePath()); // signalRestService.setupMonitoring(res.getSignalXmlFilePath().toString()); 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 264cc48..825e378 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 @@ -80,6 +80,7 @@ public class Stage implements Ru f.get(); } log.trace("Stage time stats: {}", stopWatch.info()); + dispatcher.handleOnPoison(b); putQuiet(out, dispatcher.cookPoison()); log.info("stage {} finished.", dispatcherName); return; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java index 450a87a..c7b9d1d 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java @@ -44,12 +44,13 @@ public class FidIdEnricher implements Handler { .toList(); log.debug("personOnProcessing size: {}", personOnProcessing.size()); Long packageId = taskPayload.getPackageId(); - Path partPath = writeSignals(personOnProcessing, packageId); + Path partPath = writeSignals(personOnProcessing, taskPayload); return new PartFilePayload(partPath, packageId); } - private Path writeSignals(List persons, Long packageId) { - Path dir = Paths.get(creditTrackerSettings.getLoadDir(), "tmp_folder"); + private Path writeSignals(List persons, SubjectPayload taskPayload) { + Path dir = Paths.get(creditTrackerSettings.getLoadDir(), + "tmp", taskPayload.getMemberCode(), taskPayload.getPackageName()); try { Files.createDirectories(dir); } catch (IOException e) { @@ -57,7 +58,7 @@ public class FidIdEnricher implements Handler { } String base = UUID.randomUUID() + "_output.xml"; - log.debug("Write tmp xml by packageId: {}", packageId); + log.debug("Write tmp xml by packageId: {}", taskPayload.getPackageId()); Path tmp = dir.resolve(base + ".part"); Path done = dir.resolve(base); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java new file mode 100644 index 0000000..3f099a2 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java @@ -0,0 +1,76 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.PhoneFidMap; +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.task.Handler; +import ru.nbch.credit_tracker.utils.CTUtil; + +/** + * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных. + * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки. + */ +@Service +@Slf4j +public class FidNameEnricher implements Handler { + private final PhoneService2 phoneService2; + private final CreditTrackerSettings creditTrackerSettings; + + public FidNameEnricher(PhoneService2 phoneService2, + CreditTrackerSettings creditTrackerSettings) { + this.phoneService2 = phoneService2; + this.creditTrackerSettings = creditTrackerSettings; + } + + @Override + public SubjectPayload handle(SubjectPayload taskPayload) { + List subjects = taskPayload.getSubjects(); + log.debug("NameEnricherStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size()); + + LocalDateTime now = LocalDateTime.now(); + Map nameByPhone = phoneService2.findNameByNumber(subjects); + List subjectForSave = subjects.stream().map(subject -> { + Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone())); + if (name != null) { + 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); + } + + return PhoneFidMap.builder() + .id(subject.getMapId()) + .fid(subject.getFid()) + .calculatedAt(now) + .build(); + }).toList(); + + phoneService2.updateFid(subjectForSave); + return taskPayload; + } + + @Override + public int preferredConcurrency() { + return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); + } + + @Override + public SubjectPayload cookPoison() { + return SubjectPayload.POISON; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java index 282bf47..83a04a0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java @@ -11,10 +11,14 @@ import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.exceptions.SignalClientException; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.PoisonType; 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.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @@ -24,15 +28,23 @@ public class KickPipeLineImpl implements IPipeLineStarter { private final Long packageId; private final Path pkgPath; private final StaxXmlProcessor xmlProcessor; + private final UserPackageMapper userPackageMapper; + private final PhoneService2 phoneService2; private final int BATCH_SIZE; private boolean isInterrupted = false; - public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, Path pkgPath, + public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, + Long packageId, + Path pkgPath, + UserPackageMapper userPackageMapper, + PhoneService2 phoneService2, CreditTrackerSettings creditTrackerSettings) { this.packageId = packageId; this.pkgPath = pkgPath; this.xmlProcessor = xmlProcessor; + this.userPackageMapper = userPackageMapper; + this.phoneService2 = phoneService2; this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize(); } @@ -45,6 +57,7 @@ public class KickPipeLineImpl implements IPipeLineStarter { @Override public void startFillingQueue(BlockingQueue firstQueue) { log.info("Start registering subjects. PackageId: {}", packageId); + UserPackage userPackage = userPackageMapper.findPackage(packageId); AtomicInteger counter = new AtomicInteger(0); try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) { xmlProcessor.read(in, MonitoringRequest.class, @@ -60,7 +73,7 @@ public class KickPipeLineImpl implements IPipeLineStarter { int count = counter.incrementAndGet(); log.trace("Reading subject's {} batch. Total count: {}. PackageId: {}", count, request.getSubjects().getSubject().size(), packageId); try { - firstQueue.put(new SubjectPayload(packageId, subjectProfileBatch)); + firstQueue.put(new SubjectPayload(packageId,userPackage.getUserId(), userPackage.getPackageName(), subjectProfileBatch)); } catch (InterruptedException e) { isInterrupted = true; log.error("Pipeline starter was interrupted!"); @@ -83,7 +96,8 @@ public class KickPipeLineImpl implements IPipeLineStarter { } log.debug("Send poison pill"); try { - firstQueue.put(SubjectPayload.POISON); + SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); + firstQueue.put(poison); } catch (InterruptedException e) { log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); } 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 index a8fc92e..476c3d0 100644 --- 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 @@ -1,19 +1,16 @@ package ru.nbch.credit_tracker.service.task.impl; -import java.time.LocalDateTime; import java.util.List; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.PhoneFidMap; -import ru.nbch.credit_tracker.entities.indic.Name; -import ru.nbch.credit_tracker.model.NameData; +import ru.nbch.credit_tracker.enums.Status; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; 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.task.Handler; -import ru.nbch.credit_tracker.utils.CTUtil; /** * Стадия для сопоставления FID к номеру телефона и регистрации пакета в базе данных. @@ -21,14 +18,16 @@ import ru.nbch.credit_tracker.utils.CTUtil; */ @Service @Slf4j -//todo опять нейминг класса, варианты: ResolveSubject; DefineIcrsSubject; IcrsMapping/Resolving; PackageSaver; Filler... public class SaveSubjectHandler implements Handler { private final PhoneService2 phoneService2; + private final UserPackageMapper userPackageMapper; private final CreditTrackerSettings creditTrackerSettings; public SaveSubjectHandler(PhoneService2 phoneService2, + UserPackageMapper userPackageMapper, CreditTrackerSettings creditTrackerSettings) { this.phoneService2 = phoneService2; + this.userPackageMapper = userPackageMapper; this.creditTrackerSettings = creditTrackerSettings; } @@ -37,32 +36,11 @@ public class SaveSubjectHandler implements Handler subjects = taskPayload.getSubjects(); Long packageId = taskPayload.getPackageId(); log.debug("SaveSubjectStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size()); - - LocalDateTime now = LocalDateTime.now(); - Map nameByPhone = phoneService2.findNameByNumber(subjects); - List subjectForSave = subjects.stream().map(subject -> { - Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone())); - if (name != null) { - 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); - } - - return PhoneFidMap.builder() - .subjectId(subject.getIdentifier()) - .packageId(packageId) - .phone(subject.getPhone()) - .fid(subject.getFid()) - .calculatedAt(now) - .build(); - }).toList(); + List subjectForSave = subjects.stream().map(subject -> PhoneFidMap.builder() + .subjectId(subject.getIdentifier()) + .packageId(packageId) + .phone(subject.getPhone()) + .build()).toList(); List ids = phoneService2.registerSubjects(subjectForSave); for (int i = 0; i < subjects.size(); i++) { @@ -71,9 +49,15 @@ public class SaveSubjectHandler implements Handler { - private final Path mergedDir; private final Charset encoding = Charset.forName("windows-1251"); - private final int batchSize = 10; private final CompletableFuture done = new CompletableFuture<>(); private Path out; + private final CreditTrackerSettings config; private BlockingQueue inQueue; public XmlMergeCoordinator(CreditTrackerSettings config) { - this.mergedDir = Path.of(config.getLoadDir() + File.separator + "tmp_folder" + File.separator + "merged" + File.separator); + this.config = config; } @Override public void run() { - try { - Files.createDirectories(mergedDir); - } catch (IOException e) { - throw new RuntimeException(e); - } while (!Thread.currentThread().isInterrupted()) { try { PartFilePayload part = inQueue.poll(5, TimeUnit.SECONDS); @@ -57,7 +47,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator inputs) throws Exception { - log.debug("Starting merge batch; size: {}", inputs.size()); - Long packageId = inputs.stream().findFirst().get().getPackageId(); - String outName = "merged_" + packageId + ".xml"; + private void mergeBatch(PartFilePayload input) throws Exception { + log.debug("Starting merge"); + Long packageId = input.getPackageId(); + Path partFilePath = input.getPartPath(); if (out == null) { - out = mergedDir.resolve(outName); + out = partFilePath.resolveSibling("merged_" + packageId + ".xml"); } XMLOutputFactory xof = XMLOutputFactory.newFactory(); try (OutputStream os = Files.newOutputStream(out, StandardOpenOption.CREATE, StandardOpenOption.APPEND)) { XMLStreamWriter w = xof.createXMLStreamWriter(os, encoding.name()); - - for (PartFilePayload in : inputs) { - log.debug("Copy person from file {}", in.getPartPath()); - copyPersonsChildren(in.getPartPath(), w); - } - + log.debug("Copy person from file {}", input.getPartPath()); + copyPersonsChildren(input.getPartPath(), w); w.flush(); w.close(); } - for (PartFilePayload p : inputs) Files.deleteIfExists(p.getPartPath()); -// inputs.clear(); + Files.deleteIfExists(input.getPartPath()); } /** diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 1e55556..3b3c523 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -52,37 +52,38 @@ SELECT row_number() OVER () AS rn, subject_id, - fid, phone, - package_id, - flag_debt5000, - flag_90_12, - calculated_at + package_id FROM json_to_recordset(#{json}::json) AS t( subject_id varchar(255), - fid int, phone text, - package_id bigint, - flag_debt5000 smallint, - flag_90_12 smallint, - calculated_at timestamp + package_id bigint ) ) INSERT INTO signals_phone_fid_map - (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at) + (subject_id, phone, package_id) SELECT subject_id, - fid, phone, - package_id, - flag_debt5000, - flag_90_12, - calculated_at + package_id FROM src ORDER BY rn RETURNING id; + + update signals_phone_fid_map dst + set + fid = src.fid, + calculated_at = src.calculated_at + FROM json_to_recordset(#{json}::json) AS src( + id bigint, + fid int, + calculated_at timestamp + ) + where dst.id = src.id; + + INSERT INTO signals_phone_fid_map (subject_id, phone, package_id) values diff --git a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml index 91b7b7b..fe58d65 100644 --- a/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml @@ -82,6 +82,12 @@ WHERE id = #{id} + + UPDATE signals_user_packages + SET last_searched_at = now() + WHERE id = #{id} + +