treshenkov/credit_tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add new stage; status model

Browse source
This commit is contained in:
etreschenkov 2025-11-24 15:32:07 +03:00
parent d06ea8334b
commit c65d88f645
18 changed files with 226 additions and 99 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java
View file

@ -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<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) {
return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, creditTrackerSettings);
public BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml,
CreditTrackerSettings creditTrackerSettings,
UserPackageMapper userPackageMapper,
PhoneService2 phoneService2) {
return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, userPackageMapper, phoneService2, creditTrackerSettings);
}
@Bean

11
src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java
View file

@ -10,17 +10,22 @@ public class StageConfig {
@Bean("saveSubjectQueueFactory")
public StageQueueFactory<SubjectPayload> saveSubjectQueueFactory() {
return new StageQueueFactory<>(5);
return new StageQueueFactory<>(50);
}
@Bean("nameEnricherQueueFactory")
public StageQueueFactory<SubjectPayload> nameEnricherQueueFactory() {
return new StageQueueFactory<>(20);
}
@Bean("idEnricherQueueFactory")
public StageQueueFactory<SubjectPayload> idEnricherQueueFactory() {
return new StageQueueFactory<>();
return new StageQueueFactory<>(20);
}
@Bean("partFileQueueFactory")
public StageQueueFactory<PartFilePayload> partFileQueueFactory() {
return new StageQueueFactory<>(10000);
return new StageQueueFactory<>(100);
}
}

1
src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java
View file

@ -14,6 +14,7 @@ public interface PhoneFidMapMapper {
void registerSubjects(@Param("subjects") List<Subject> subjects, @Param("packageId") Long packageId);
void register(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
List<Long> registerJson(@Param("json") String json);
void setFidJson(@Param("json") String json);
void registerWithId(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("offset") int offset, @Param("limit") int limit);

1
src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java
View file

@ -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);

22
src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java
View file

@ -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<SubjectProfile> 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<SubjectProfile> subjects) {
public SubjectPayload(Long packageId, String memberCode, String packageName, List<SubjectProfile> 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;

2
src/main/java/ru/nbch/credit_tracker/service/SignalServiceV2.java
View file

@ -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);
}

19
src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java
View file

@ -65,14 +65,29 @@ public class PhoneService2 {
String json = entityToJsonMapper.writeValueAsString(subjects.stream()
.map(phoneFidMap -> new HashMap<String, Object>() {{
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<PhoneFidMap> subjects) {
try {
//fixme, временное решение
String json = entityToJsonMapper.writeValueAsString(subjects.stream()
.map(phoneFidMap -> new HashMap<String, Object>() {{
put("id", phoneFidMap.getId());
put("fid", phoneFidMap.getFid());
put("calculated_at", phoneFidMap.getCalculatedAt());
}}).toList());
phoneFidMapMapper.setFidJson(json);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
}

3
src/main/java/ru/nbch/credit_tracker/service/task/Handler.java
View file

@ -2,6 +2,9 @@ package ru.nbch.credit_tracker.service.task;
public interface Handler<X, Z> {
Z handle(X taskPayload);
default void handleOnPoison(X poison){
//do nothing
}
default int preferredConcurrency() {return 10;}
Z cookPoison();

19
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java
View file

@ -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<SubjectPayload> saveSubjectQueueFactory;
private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory;
private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
private final SaveSubjectHandler saveSubjectHandler;
private final UserPackageMapper userPackageMapper;
private final FidIdEnricher fidIdEnricherHandler;
private final FidNameEnricher fidNameEnricher;
private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
private final BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
private final CreditTrackerSettings creditTrackerSettings;
@ -50,22 +55,28 @@ public class PipelinePackageManager {
public PipelinePackageManager(SignalRestService signalRestService,
@Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
@Qualifier("nameEnricherQueueFactory") StageQueueFactory<SubjectPayload> nameEnricherQueueFactory,
@Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
SaveSubjectHandler saveSubjectHandler,
UserPackageMapper userPackageMapper,
FidIdEnricher fidIdEnricherHandler,
Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> 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());

1
src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
View file

@ -80,6 +80,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
f.get();
}
log.trace("Stage time stats: {}", stopWatch.info());
dispatcher.handleOnPoison(b);
putQuiet(out, dispatcher.cookPoison());
log.info("stage {} finished.", dispatcherName);
return;

9
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java
View file

@ -44,12 +44,13 @@ public class FidIdEnricher implements Handler<SubjectPayload, PartFilePayload> {
.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<Person> persons, Long packageId) {
Path dir = Paths.get(creditTrackerSettings.getLoadDir(), "tmp_folder");
private Path writeSignals(List<Person> 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<SubjectPayload, PartFilePayload> {
}
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);

76
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java Normal file
View file

@ -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<SubjectPayload, SubjectPayload> {
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<SubjectProfile> subjects = taskPayload.getSubjects();
log.debug("NameEnricherStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size());
LocalDateTime now = LocalDateTime.now();
Map<String, Name> nameByPhone = phoneService2.findNameByNumber(subjects);
List<PhoneFidMap> 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;
}
}

20
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java
View file

@ -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<SubjectPayload> {
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<SubjectPayload> {
@Override
public void startFillingQueue(BlockingQueue<SubjectPayload> 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<SubjectPayload> {
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<SubjectPayload> {
}
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));
}

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

@ -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<SubjectPayload, SubjectPayload> {
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<SubjectPayload, SubjectPayloa
List<SubjectProfile> 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<String, Name> nameByPhone = phoneService2.findNameByNumber(subjects);
List<PhoneFidMap> 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<PhoneFidMap> subjectForSave = subjects.stream().map(subject -> PhoneFidMap.builder()
.subjectId(subject.getIdentifier())
.packageId(packageId)
.phone(subject.getPhone())
.build()).toList();
List<Long> ids = phoneService2.registerSubjects(subjectForSave);
for (int i = 0; i < subjects.size(); i++) {
@ -71,9 +49,15 @@ public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayloa
return taskPayload;
}
@Override
public void handleOnPoison(SubjectPayload poison) {
log.trace("Got poison; set package on PROCESSING by packageId: {}", poison.getPackageId());
userPackageMapper.updatePackageStatus(poison.getPackageId(), Status.PROCESSING.name());
}
@Override
public int preferredConcurrency() {
return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency();
return 20;
}
@Override

37
src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java
View file

@ -1,15 +1,11 @@
package ru.nbch.credit_tracker.service.task.impl;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@ -28,24 +24,18 @@ import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
@Slf4j
public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload> {
private final Path mergedDir;
private final Charset encoding = Charset.forName("windows-1251");
private final int batchSize = 10;
private final CompletableFuture<SignalXmlFilePayload> done = new CompletableFuture<>();
private Path out;
private final CreditTrackerSettings config;
private BlockingQueue<PartFilePayload> 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<PartFilePa
done.complete(new SignalXmlFilePayload(out));
return;
}
mergeBatch(Collections.singletonList(part));
mergeBatch(part);
} catch (InterruptedException ie) {
log.info("pipeline terminator was interrupted!");
done.completeExceptionally(new PipelineTerminatorException());
@ -71,28 +61,23 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
public static class PipelineTerminatorException extends RuntimeException {
}
private void mergeBatch(List<PartFilePayload> 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());
}
/**

33
src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml
View file

@ -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;
</select>
<update id="setFidJson" parameterType="string">
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;
</update>
<insert id="registerWithId" parameterType="java.util.List" useGeneratedKeys="true" keyColumn="id" keyProperty="id">
INSERT INTO signals_phone_fid_map (subject_id, phone, package_id)
values

6
src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml
View file

@ -82,6 +82,12 @@
WHERE id = #{id}
</update>
<update id="setLastSearchedDt">
UPDATE signals_user_packages
SET last_searched_at = now()
WHERE id = #{id}
</update>
<select id="findPackage" resultMap="packageDataResultMap"
resultType="ru.nbch.credit_tracker.entities.app.UserPackage">
SELECT id, membercode, package_name, created_at, status, deactivated_at, last_downloaded_report, monitoring_at

6
src/main/resources/sql/v2.21.ddl
View file

@ -4,12 +4,12 @@ CREATE TABLE app_version (
);
alter table signals_user_packages
alter column LAST_SEARCHED_AT type timestamp;
add LAST_SEARCHED_AT timestamp;
alter table signals_user_packages
alter column REJECT_STATUS type varchar(255);
add REJECT_STATUS varchar(255);
alter table signals_phone_fid_map
alter column SEARCHED_AT type timestamp;
add SEARCHED_AT timestamp;
CREATE TABLE signals_flags (