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

remove poison step + delete tmp xml file request after validation

Browse source
This commit is contained in:
etreschenkov 2025-11-21 16:16:48 +03:00
parent 9794cc6244
commit 7e641162ab
20 changed files with 89 additions and 144 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java
View file

@ -62,10 +62,10 @@ public class MonitoringRequestValidator {
log.trace("Validated zip content ok!");
}
public ValidationResult readAndValidateMonitoringRequest(String filePath) {
public ValidationResult readAndValidateMonitoringRequest(Path filePath) {
log.debug("Validating monitoring request...");
ValidationResult validationResult = new ValidationResult();
try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(filePath)), 8 * 1024 * 1024)) {
try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(filePath), 8 * 1024 * 1024)) {
staxXmlProcessor.read(in, MonitoringRequest.class,
request -> {
try {

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

@ -1,5 +1,6 @@
package ru.nbch.credit_tracker.config.task;
import java.nio.file.Path;
import java.util.function.BiFunction;
import java.util.function.Supplier;
import org.springframework.context.annotation.Bean;
@ -17,8 +18,8 @@ import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@Configuration
public class PipelineStarterConfig {
@Bean
public BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) {
return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName, creditTrackerSettings);
public BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) {
return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, creditTrackerSettings);
}
@Bean

33
src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java
View file

@ -2,11 +2,13 @@ package ru.nbch.credit_tracker.facade;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.time.LocalDateTime;
import java.util.List;
import javax.xml.stream.XMLStreamException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.springframework.core.io.FileUrlResource;
import org.springframework.core.io.Resource;
import org.springframework.http.ResponseEntity;
@ -73,22 +75,22 @@ public class SignalFacadeV3 {
monitoringRequestValidator.validateZipContent(file);
log.trace("Saving monitoring request to tmp file...");
String filePath;
Path tmpReqFilePath;
try {
String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator;
filePath = ZipUtils.writeTmpFile(folderPath, file);
String folderPath = config.getLoadDir() + File.separator + "tmp" + File.separator;
tmpReqFilePath = ZipUtils.writeTmpFile(folderPath, file);
} catch (IOException e) {
throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringError, e);
}
log.debug("File saved to {}", filePath);
log.debug("File saved to {}", tmpReqFilePath);
ValidationResult validationResult = null;
try {
validationResult = monitoringRequestValidator.readAndValidateMonitoringRequest(filePath);
validationResult = monitoringRequestValidator.readAndValidateMonitoringRequest(tmpReqFilePath);
if (!validationResult.getValidationErrors().isEmpty()) {
MonitoringValidationError firstError = validationResult.getValidationErrors().get(0);
MonitoringValidationError firstError = validationResult.getValidationErrors().getFirst();
signalService.rejectPackage(validationResult.getUserId(), validationResult.getPackageName());
FileUtils.delete(new File(filePath));
Files.delete(tmpReqFilePath);
throw firstError.getSignalException();
}
} catch (RuntimeException e) {
@ -103,6 +105,13 @@ public class SignalFacadeV3 {
String packageName = validationResult.getPackageName();
String userId = validationResult.getUserId();
String password = validationResult.getPassword();
Path parsedFilePath = tmpReqFilePath.resolveSibling(Path.of(userId.substring(0, 6) + "_" + packageName + ".xml"));
try {
Files.move(tmpReqFilePath, parsedFilePath, StandardCopyOption.REPLACE_EXISTING);
} catch (IOException e) {
throw new RuntimeException(e);
}
try {
// проверка авторизации
AuthorizationService.AuthResult authResult = authorizationService.findAuthInfo(userId, password);
@ -120,9 +129,13 @@ public class SignalFacadeV3 {
Long packageId = signalService.registerPackage(userId, packageName);
LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId));
log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name());
pipelinePackageManager.buildAndRun(packageId, filePath);
pipelinePackageManager.buildAndRun(packageId, parsedFilePath);
} catch (Exception e) {
CTUtil.removeFile(filePath);
try {
Files.delete(parsedFilePath);
} catch (IOException ex) {
throw new RuntimeException(ex);
}
if (e instanceof SignalClientException) {
throw (SignalClientException) e;
}

16
src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java
View file

@ -1,16 +0,0 @@
package ru.nbch.credit_tracker.model;
public class DummyPayload implements StagePayload {
private final boolean poison;
public static DummyPayload POISON = new DummyPayload(true);
public DummyPayload(boolean poison) {
this.poison = poison;
}
@Override
public boolean isPoison() {
return poison;
}
}

37
src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java
View file

@ -1,37 +0,0 @@
package ru.nbch.credit_tracker.model;
import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
import java.util.List;
public class FlagsPayload implements StagePayload {
private final Long packageId;
private final List<PhoneFidMap> phoneFidMaps;
private final boolean poison;
public FlagsPayload(Long packageId, List<PhoneFidMap> fidsMapByBatch) {
this.packageId = packageId;
this.phoneFidMaps = fidsMapByBatch;
this.poison = false;
}
public FlagsPayload(boolean poison) {
this.packageId = null;
this.phoneFidMaps = null;
this.poison = poison;
}
public Long getPackageId() {
return packageId;
}
public List<PhoneFidMap> getPhoneFidMaps() {
return phoneFidMaps;
}
@Override
public boolean isPoison() {
return poison;
}
}

7
src/main/java/ru/nbch/credit_tracker/model/PoisonType.java Normal file
View file

@ -0,0 +1,7 @@
package ru.nbch.credit_tracker.model;
public enum PoisonType {
NONE,
NORMALLY,
EXCEPTIONALLY
}

16
src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java
View file

@ -1,16 +0,0 @@
package ru.nbch.credit_tracker.model;
public class SendSignalsPayload implements StagePayload {
private String signalXmlPath;
public static final SendSignalsPayload POISON = new SendSignalsPayload(true);
private boolean poison;
public SendSignalsPayload(boolean poison) {
this.poison = poison;
}
@Override
public boolean isPoison() {
return poison;
}
}

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

@ -2,16 +2,17 @@ package ru.nbch.credit_tracker.model.impl;
import java.nio.file.Path;
import lombok.Getter;
import ru.nbch.credit_tracker.model.PoisonType;
import ru.nbch.credit_tracker.model.StagePayload;
@Getter
public class PartFilePayload implements StagePayload {
private final Path partPath;
private final Long packageId;
private final boolean poison;
public static final PartFilePayload POISON = new PartFilePayload(true);
private final PoisonType poison;
public static final PartFilePayload POISON = new PartFilePayload(PoisonType.NORMALLY);
public PartFilePayload(boolean poison) {
public PartFilePayload(PoisonType poison) {
this.poison = poison;
this.partPath = null;
this.packageId = null;
@ -20,11 +21,11 @@ public class PartFilePayload implements StagePayload {
public PartFilePayload(Path partPath, Long packageId) {
this.partPath = partPath;
this.packageId = packageId;
this.poison = false;
this.poison = PoisonType.NONE;
}
@Override
public boolean isPoison() {
return poison;
return poison == PoisonType.NORMALLY;
}
}

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

@ -1,6 +1,7 @@
package ru.nbch.credit_tracker.model.impl;
import java.util.List;
import ru.nbch.credit_tracker.model.PoisonType;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.model.SubjectProfile;
@ -8,10 +9,10 @@ import ru.nbch.credit_tracker.model.SubjectProfile;
public class SubjectPayload implements StagePayload {
private final Long packageId;
private final List<SubjectProfile> subjects;
public static final SubjectPayload POISON = new SubjectPayload(true);
private final boolean poison;
public static final SubjectPayload POISON = new SubjectPayload(PoisonType.NORMALLY);
private final PoisonType poison;
public SubjectPayload(boolean poison) {
public SubjectPayload(PoisonType poison) {
this.packageId = null;
this.subjects = null;
this.poison = poison;
@ -20,7 +21,7 @@ public class SubjectPayload implements StagePayload {
public SubjectPayload(Long packageId, List<SubjectProfile> subjects) {
this.packageId = packageId;
this.subjects = subjects;
this.poison = false;
this.poison = PoisonType.NONE;
}
public Long getPackageId() {
@ -33,6 +34,6 @@ public class SubjectPayload implements StagePayload {
@Override
public boolean isPoison() {
return poison;
return poison == PoisonType.NORMALLY;
}
}

1
src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java
View file

@ -139,6 +139,7 @@ public class SignalRestService {
try {
SgnlReportMonitorigSetupError report = xmlProcessor.read(answer, SgnlReportMonitorigSetupError.class);
if (report.getOperationReport().getErrorsForNextReport() != null) {
//todo set reject_status = 0
String fileName = report.getOperationReport().getErrorsForNextReport().getFilename();
log.warn("Найдены ошибки по пакету. packCode={}, packageId={}, fileName = {}", packCode, packageId, fileName);
return fileName;

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

@ -3,4 +3,6 @@ package ru.nbch.credit_tracker.service.task;
public interface Handler<X, Z> {
Z handle(X taskPayload);
default int preferredConcurrency() {return 10;}
Z cookPoison();
}

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

@ -1,5 +1,6 @@
package ru.nbch.credit_tracker.service.task;
import java.nio.file.Path;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@ -35,7 +36,7 @@ public class PipelinePackageManager {
private final SaveSubjectHandler saveSubjectHandler;
private final FidIdEnricher fidIdEnricherHandler;
private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
private final BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
private final CreditTrackerSettings creditTrackerSettings;
private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> {
Thread t = new Thread(r);
@ -51,7 +52,7 @@ public class PipelinePackageManager {
@Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
FlagsEnricherV2 flagsEnricherV2,
CreditTrackerSettings creditTrackerSettings) {
this.signalRestService = signalRestService;
@ -67,7 +68,7 @@ public class PipelinePackageManager {
this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS);
}
public void buildAndRun(Long packageId, String pkgPath) throws Exception {
public void buildAndRun(Long packageId, Path pkgPath) throws Exception {
if (!semaphore.tryAcquire()) {
throw new RuntimeException("Too many concurrent requests");
}
@ -75,12 +76,10 @@ public class PipelinePackageManager {
try {
Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
.handler(saveSubjectHandler)
.poison(SubjectPayload.POISON)
.handlerResult(flagsEnricherV2)
.andThen()
.stage(sendSignalQueueFactory.getObject())
.handler(fidIdEnricherHandler)
.poison(PartFilePayload.POISON)
.andThen()
.fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath))
.pollFinishingQueue(pipelineTerminatorFactory.get())

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

@ -22,7 +22,6 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
private final Handler<X, Z> dispatcher;
private final BlockingQueue<X> in;
private final BlockingQueue<Z> out;
private final Z POISON;
private final ExecutorService ioPool;
private Consumer<Z> consumer = null;
private final String dispatcherName;
@ -33,8 +32,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
public Stage(Handler<X, Z> dispatcher,
BlockingQueue<X> in,
BlockingQueue<Z> out,
Z POISON) {
BlockingQueue<Z> out) {
this.dispatcher = dispatcher;
this.in = in;
this.out = out;
@ -54,7 +52,6 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
callerRunsPolicy.rejectedExecution(r, executor);
}
);
this.POISON = POISON;
}
public void setPipeline(Pipeline pipeline) {
@ -87,7 +84,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
}
f.get();
}
putQuiet(out, POISON);
putQuiet(out, dispatcher.cookPoison());
log.info("stage {} finished.", dispatcherName);
return;
}
@ -131,7 +128,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
}
} catch (Exception e) {
log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
putQuiet(out, POISON);
putQuiet(out, dispatcher.cookPoison());
}
}
@ -147,7 +144,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
private boolean wasInterrupted() {
if (Thread.interrupted()) {
log.warn("pipeline was interrupted...");
//putQuiet(out, POISON);
//putQuiet(out, dispatcher.cookPoison());
return true;
}
return false;
@ -184,10 +181,6 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
return in;
}
public Consumer<Z> getConsumer() {
return consumer;
}
public void setConsumer(Consumer<Z> consumer) {
this.consumer = consumer;
}

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

@ -74,4 +74,9 @@ public class FidIdEnricher implements Handler<SubjectPayload, PartFilePayload> {
}
return done;
}
@Override
public PartFilePayload cookPoison() {
return PartFilePayload.POISON;
}
}

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

@ -22,13 +22,13 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Long packageId;
private final String pkgPath;
private final Path pkgPath;
private final StaxXmlProcessor xmlProcessor;
private final int BATCH_SIZE;
private boolean isInterrupted = false;
public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, String pkgPath,
public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, Path pkgPath,
CreditTrackerSettings creditTrackerSettings) {
this.packageId = packageId;
this.pkgPath = pkgPath;
@ -46,7 +46,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
log.info("Start registering subjects. PackageId: {}", packageId);
AtomicInteger counter = new AtomicInteger(0);
try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(pkgPath)), 8 * 1024 * 1024)) {
try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) {
xmlProcessor.read(in, MonitoringRequest.class,
request -> {
if (!request.getSubjects().getSubject().isEmpty()) {
@ -62,23 +62,28 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
try {
firstQueue.put(new SubjectPayload(packageId, subjectProfileBatch));
} catch (InterruptedException e) {
isInterrupted = true;
log.error("Pipeline starter was interrupted!");
throw new PipelineStarterException();
}
request.getSubjects().getSubject().clear();
}
},
request -> request.getSubjects().getSubject().isEmpty()
,
BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
request -> request.getSubjects().getSubject().isEmpty(),
BATCH_SIZE, "/MonitoringRequest/Subjects/Subject"
);
} catch (IOException e) {
log.error("{}", ExceptionUtils.getStackTrace(e));
} catch (PipelineStarterException e) {
log.info("Pipeline starter shitting down...");
} finally {
if (isInterrupted) {
//noinspection ReturnInsideFinallyBlock
return;
}
log.debug("Send poison pill");
try {
firstQueue.put(new SubjectPayload(true));
firstQueue.put(SubjectPayload.POISON);
} catch (InterruptedException e) {
log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e));
}

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

@ -71,4 +71,9 @@ public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayloa
public int preferredConcurrency() {
return 15;
}
@Override
public SubjectPayload cookPoison() {
return SubjectPayload.POISON;
}
}

20
src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java
View file

@ -8,21 +8,19 @@ import ru.nbch.credit_tracker.service.task.Handler;
import ru.nbch.credit_tracker.service.task.Stage;
import ru.nbch.credit_tracker.service.task.setup.step.BuildStep;
import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep;
import ru.nbch.credit_tracker.service.task.setup.step.PoisonStep;
/**
* Step Builder для Stage:
* Обязательная последовательность: handler(...) -> poison(...) -> buildStage()
*/
public class StageBuilderV2<FIRST extends StagePayload, I extends StagePayload, O extends StagePayload>
implements HandlerStep<FIRST, I, O>, PoisonStep<FIRST, I, O>, BuildStep<FIRST, O> {
implements HandlerStep<FIRST, I, O>, BuildStep<FIRST, O> {
private PipeLineBuilderV2<FIRST, O> parent;
private final BlockingQueue<I> inputQueue;
private final BlockingQueue<O> outputQueue;
private Handler<I, O> handler;
private O poison;
Consumer<O> consumer = null;
private Consumer<O> consumer = null;
private StageBuilderV2(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
this.inputQueue = Objects.requireNonNull(inputQueue, "inputQueue is required");
@ -43,18 +41,11 @@ public class StageBuilderV2<FIRST extends StagePayload, I extends StagePayload,
/** Шаг 1: задать обработчик */
@Override
public PoisonStep<FIRST, I, O> handler(Handler<I, O> handler) {
public BuildStep<FIRST, O> handler(Handler<I, O> handler) {
this.handler = Objects.requireNonNull(handler, "handler is required");
return this;
}
/** Шаг 2: задать poison pill */
@Override
public BuildStep<FIRST, O> poison(O poison) {
this.poison = Objects.requireNonNull(poison, "poison is required");
return this;
}
public BuildStep<FIRST, O> handlerResult(Consumer<O> consumer) {
this.consumer = Objects.requireNonNull(consumer, "consumer is required");
return this;
@ -69,11 +60,8 @@ public class StageBuilderV2<FIRST extends StagePayload, I extends StagePayload,
if (handler == null) {
throw new IllegalStateException("handler must be set before buildStage()");
}
if (poison == null) {
throw new IllegalStateException("poison must be set before buildStage()");
}
Stage<I, O> stage = new Stage<>(handler, inputQueue, outputQueue, poison);
Stage<I, O> stage = new Stage<>(handler, inputQueue, outputQueue);
stage.setConsumer(consumer);
parent.addStage(stage);
return parent;

2
src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java
View file

@ -4,5 +4,5 @@ import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.service.task.Handler;
public interface HandlerStep<FIRST extends StagePayload, I extends StagePayload, O extends StagePayload> {
PoisonStep<FIRST, I, O> handler(Handler<I, O> handler);
BuildStep<FIRST, O> handler(Handler<I, O> handler);
}

7
src/main/java/ru/nbch/credit_tracker/service/task/setup/step/PoisonStep.java
View file

@ -1,7 +0,0 @@
package ru.nbch.credit_tracker.service.task.setup.step;
import ru.nbch.credit_tracker.model.StagePayload;
public interface PoisonStep<FIRST extends StagePayload, I extends StagePayload, O extends StagePayload> {
BuildStep<FIRST, O> poison(O poison);
}

6
src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java
View file

@ -37,11 +37,11 @@ public class ZipUtils {
file.getOriginalFilename() != null && file.getOriginalFilename().toLowerCase().endsWith(".zip");
}
public static String writeTmpFile(String path, MultipartFile file) throws IOException {
public static Path writeTmpFile(String path, MultipartFile file) throws IOException {
CTUtil.createFolder(path);
String filePath = path + UUID.randomUUID() + "monitoring_request.xml";
Path filePath = Path.of(path, UUID.randomUUID() + "_monitoring_request.xml");
try (OutputStream out = Files.newOutputStream(Path.of(filePath))) {
try (OutputStream out = Files.newOutputStream(filePath)) {
ZipInputStream zipInputStream = new ZipInputStream(file.getInputStream());
zipInputStream.getNextEntry();
byte[] buffer = new byte[8 * 1024 * 1024];