From 7e641162ab74a26e2e0f732c8b80f586b5617048 Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Fri, 21 Nov 2025 16:16:48 +0300 Subject: [PATCH] remove poison step + delete tmp xml file request after validation --- .../component/MonitoringRequestValidator.java | 4 +- .../config/task/PipelineStarterConfig.java | 5 ++- .../credit_tracker/facade/SignalFacadeV3.java | 33 ++++++++++++----- .../credit_tracker/model/DummyPayload.java | 16 -------- .../credit_tracker/model/FlagsPayload.java | 37 ------------------- .../nbch/credit_tracker/model/PoisonType.java | 7 ++++ .../model/SendSignalsPayload.java | 16 -------- .../model/impl/PartFilePayload.java | 11 +++--- .../model/impl/SubjectPayload.java | 11 +++--- .../service/signal/SignalRestService.java | 1 + .../credit_tracker/service/task/Handler.java | 2 + .../service/task/PipelinePackageManager.java | 9 ++--- .../credit_tracker/service/task/Stage.java | 15 ++------ .../service/task/impl/FidIdEnricher.java | 5 +++ .../service/task/impl/KickPipeLineImpl.java | 21 +++++++---- .../service/task/impl/SaveSubjectHandler.java | 5 +++ .../service/task/setup/StageBuilderV2.java | 20 ++-------- .../service/task/setup/step/HandlerStep.java | 2 +- .../service/task/setup/step/PoisonStep.java | 7 ---- .../nbch/credit_tracker/utils/ZipUtils.java | 6 +-- 20 files changed, 89 insertions(+), 144 deletions(-) delete mode 100644 src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/PoisonType.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/setup/step/PoisonStep.java diff --git a/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java b/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java index d04fee8..b28b163 100644 --- a/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java +++ b/src/main/java/ru/nbch/credit_tracker/component/MonitoringRequestValidator.java @@ -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 { 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 9bd49ca..221c86a 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 @@ -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> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) { - return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName, creditTrackerSettings); + public BiFunction> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) { + return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, creditTrackerSettings); } @Bean diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java index 288b96a..0664d10 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -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; } diff --git a/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java b/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java deleted file mode 100644 index 137c6c8..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java +++ /dev/null @@ -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; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java b/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java deleted file mode 100644 index 89bb7c1..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/FlagsPayload.java +++ /dev/null @@ -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 phoneFidMaps; - private final boolean poison; - - public FlagsPayload(Long packageId, List 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 getPhoneFidMaps() { - return phoneFidMaps; - } - - @Override - public boolean isPoison() { - return poison; - } - -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/PoisonType.java b/src/main/java/ru/nbch/credit_tracker/model/PoisonType.java new file mode 100644 index 0000000..2917ddf --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/PoisonType.java @@ -0,0 +1,7 @@ +package ru.nbch.credit_tracker.model; + +public enum PoisonType { + NONE, + NORMALLY, + EXCEPTIONALLY +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java b/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java deleted file mode 100644 index 1c15021..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/SendSignalsPayload.java +++ /dev/null @@ -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; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java index 9998061..049dfea 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java @@ -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; } } 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 8f9cd24..f9e4b1e 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 @@ -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 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 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; } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java index 03c1adf..fc418cc 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java @@ -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; 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 8d46872..942e5ad 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 @@ -3,4 +3,6 @@ package ru.nbch.credit_tracker.service.task; public interface Handler { Z handle(X taskPayload); 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 2eae8fa..3e6e01d 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 @@ -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> pipelineTerminatorFactory; - private final BiFunction> pipelineStarterFactory; + private final BiFunction> 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 sendSignalQueueFactory, SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, Supplier> pipelineTerminatorFactory, - BiFunction> pipelineStarterFactory, + BiFunction> 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()) 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 4422e4a..19f6c97 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 @@ -22,7 +22,6 @@ public class Stage implements Ru private final Handler dispatcher; private final BlockingQueue in; private final BlockingQueue out; - private final Z POISON; private final ExecutorService ioPool; private Consumer consumer = null; private final String dispatcherName; @@ -33,8 +32,7 @@ public class Stage implements Ru public Stage(Handler dispatcher, BlockingQueue in, - BlockingQueue out, - Z POISON) { + BlockingQueue out) { this.dispatcher = dispatcher; this.in = in; this.out = out; @@ -54,7 +52,6 @@ public class Stage implements Ru callerRunsPolicy.rejectedExecution(r, executor); } ); - this.POISON = POISON; } public void setPipeline(Pipeline pipeline) { @@ -87,7 +84,7 @@ public class Stage implements Ru } f.get(); } - putQuiet(out, POISON); + putQuiet(out, dispatcher.cookPoison()); log.info("stage {} finished.", dispatcherName); return; } @@ -131,7 +128,7 @@ public class Stage 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 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 implements Ru return in; } - public Consumer getConsumer() { - return consumer; - } - public void setConsumer(Consumer consumer) { this.consumer = consumer; } 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 251fed5..dca3bdf 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 @@ -74,4 +74,9 @@ public class FidIdEnricher implements Handler { } return done; } + + @Override + public PartFilePayload cookPoison() { + return PartFilePayload.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 24e42b2..ee1c8e0 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 @@ -22,13 +22,13 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils; public class KickPipeLineImpl implements IPipeLineStarter { 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 { public void startFillingQueue(BlockingQueue 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 { 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)); } 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 47dd3e8..034a11b 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 @@ -71,4 +71,9 @@ public class SaveSubjectHandler implements Handler poison(...) -> buildStage() */ public class StageBuilderV2 - implements HandlerStep, PoisonStep, BuildStep { + implements HandlerStep, BuildStep { private PipeLineBuilderV2 parent; private final BlockingQueue inputQueue; private final BlockingQueue outputQueue; private Handler handler; - private O poison; - Consumer consumer = null; + private Consumer consumer = null; private StageBuilderV2(BlockingQueue inputQueue, BlockingQueue outputQueue) { this.inputQueue = Objects.requireNonNull(inputQueue, "inputQueue is required"); @@ -43,18 +41,11 @@ public class StageBuilderV2 handler(Handler handler) { + public BuildStep handler(Handler handler) { this.handler = Objects.requireNonNull(handler, "handler is required"); return this; } - /** Шаг 2: задать poison pill */ - @Override - public BuildStep poison(O poison) { - this.poison = Objects.requireNonNull(poison, "poison is required"); - return this; - } - public BuildStep handlerResult(Consumer consumer) { this.consumer = Objects.requireNonNull(consumer, "consumer is required"); return this; @@ -69,11 +60,8 @@ public class StageBuilderV2 stage = new Stage<>(handler, inputQueue, outputQueue, poison); + Stage stage = new Stage<>(handler, inputQueue, outputQueue); stage.setConsumer(consumer); parent.addStage(stage); return parent; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java index 95de602..0f8ece4 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java @@ -4,5 +4,5 @@ import ru.nbch.credit_tracker.model.StagePayload; import ru.nbch.credit_tracker.service.task.Handler; public interface HandlerStep { - PoisonStep handler(Handler handler); + BuildStep handler(Handler handler); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/PoisonStep.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/PoisonStep.java deleted file mode 100644 index d2600c7..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/PoisonStep.java +++ /dev/null @@ -1,7 +0,0 @@ -package ru.nbch.credit_tracker.service.task.setup.step; - -import ru.nbch.credit_tracker.model.StagePayload; - -public interface PoisonStep { - BuildStep poison(O poison); -} diff --git a/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java b/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java index 8acc2fd..296cfde 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/ZipUtils.java @@ -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];