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

./

Browse source
This commit is contained in:
etreschenkov 2025-11-26 14:59:32 +03:00
parent c0cd36cd0a
commit 930a563172
7 changed files with 58 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

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

8
src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java
View file

@ -9,18 +9,24 @@ import ru.nbch.credit_tracker.model.StagePayload;
public class PartFilePayload implements StagePayload {
private final Path partPath;
private final Long packageId;
private final String membercode;
private final String packageName;
private final PoisonType poison;
public static final PartFilePayload POISON = new PartFilePayload(PoisonType.NORMALLY);
public PartFilePayload(PoisonType poison) {
this.poison = poison;
this.membercode = null;
this.packageName = null;
this.partPath = null;
this.packageId = null;
}
public PartFilePayload(Path partPath, Long packageId) {
public PartFilePayload(Path partPath, Long packageId, String membercode, String packageName) {
this.partPath = partPath;
this.packageId = packageId;
this.membercode = membercode;
this.packageName = packageName;
this.poison = PoisonType.NONE;
}

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

@ -21,7 +21,6 @@ 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.Saver;
import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
@ -34,11 +33,9 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
public class PipelinePackageManager {
private final SignalRestService signalRestService;
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;
@ -59,14 +56,11 @@ public class PipelinePackageManager {
private final Semaphore semaphore;
private final PipelineBackoff pipelineBackoff;
private final Saver saver;
private final static int MAX_PARALLEL_REQUESTS = 3;
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,
UserPackageMapper userPackageMapper,
FidIdEnricher fidIdEnricherHandler,
Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
@ -75,11 +69,9 @@ public class PipelinePackageManager {
CreditTrackerSettings creditTrackerSettings,
PipelineBackoff pipelineBackoff, Saver saver) {
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;
@ -102,7 +94,7 @@ public class PipelinePackageManager {
try {
Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
.handler(fidNameEnricher)
.handlerResult(flagsEnricherV2)
// .handlerResult(flagsEnricherV2)
.andThen()
.stage(sendSignalQueueFactory.getObject())
.handler(fidIdEnricherHandler)

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

@ -73,6 +73,8 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
int counter = 0;
while (true) {
X b = takeQuiet();
counter++;
log.trace("Take item from queue counter: {}", counter);
if (wasInterrupted()) {
break;
}
@ -91,9 +93,9 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
log.info("stage {} finished.", dispatcherName);
return;
}
counter++;
int finalCounter = counter;
AtomicInteger retryForSingleTask = new AtomicInteger(0);
log.trace("submit task");
Future<Void> f = ioPool.submit(() -> {
while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) {
try {

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

@ -45,7 +45,7 @@ public class FidIdEnricher implements Handler<SubjectPayload, PartFilePayload> {
log.debug("personOnProcessing size: {}", personOnProcessing.size());
Long packageId = taskPayload.getPackageId();
Path partPath = writeSignals(personOnProcessing, taskPayload);
return new PartFilePayload(partPath, packageId);
return new PartFilePayload(partPath, packageId, taskPayload.getMemberCode(), taskPayload.getPackageName());
}
private Path writeSignals(List<Person> persons, SubjectPayload taskPayload) {

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

@ -97,7 +97,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
}
}
// @Override
public void startFillingQueueXml(BlockingQueue<SubjectPayload> firstQueue) {
public void startFillingQueueXmnl(BlockingQueue<SubjectPayload> firstQueue) {
log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath);
UserPackage userPackage = userPackageMapper.findPackage(packageId);
AtomicInteger counter = new AtomicInteger(0);

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

@ -1,7 +1,10 @@
package ru.nbch.credit_tracker.service.task.impl;
import java.io.BufferedWriter;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
@ -46,6 +49,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
}
if (part.isPoison()) {
log.debug("Got poison");
finishMerge();
done.complete(new SignalXmlFilePayload(out));
log.info("Stage time stats: {}", stopWatch.info());
return;
@ -70,13 +74,37 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
log.debug("Starting merge");
Long packageId = input.getPackageId();
Path partFilePath = input.getPartPath();
boolean first = false;
if (out == null) {
first = true;
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());
if (first) {
w.writeStartDocument(encoding.name(), "1.0");
w.writeStartElement("Signals");
w.writeAttribute("online", "2");
w.writeAttribute("processWithErrors", "1");
w.writeStartElement("Id");
w.writeCharacters("%s_%s".formatted(input.getMembercode(), input.getPackageName()));
w.writeEndElement(); // </Id>
w.writeStartElement("Auth");
w.writeStartElement("User");
w.writeCharacters(config.getSignalsuser());
w.writeEndElement(); // </User>
w.writeEndElement(); // </Auth>
// <Conditions><Persons>
w.writeStartElement("Conditions");
w.writeStartElement("Persons");
}
log.debug("Copy person from file {}", input.getPartPath());
copyPersonsChildren(input.getPartPath(), w);
w.flush();
@ -85,6 +113,20 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
Files.deleteIfExists(input.getPartPath());
}
private void finishMerge() throws Exception {
if (out == null) {
return;
}
try (Writer w = new BufferedWriter(new OutputStreamWriter(
Files.newOutputStream(out, StandardOpenOption.APPEND),
encoding))) {
w.write("</Persons>");
w.write("</Conditions>");
w.write("</Signals>");
}
}
/**
* Копирует всё содержимое внутри <Persons>...</Persons> входного файла
* в текущий writer, но без повторного открытия/закрытия корневого тега.
@ -123,7 +165,6 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
r.close();
}
}
private void writeStart(XMLStreamWriter w, XMLStreamReader r) throws XMLStreamException {
w.writeStartElement(r.getLocalName());
for (int i = 0; i < r.getAttributeCount(); i++) {