From f10540e80a0947d8b080a4f1a462ad2f6eac0afd Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Mon, 10 Nov 2025 16:51:03 +0300 Subject: [PATCH] implement terminator --- .../config/task/PipelineStarterConfig.java | 12 ++ .../credit_tracker/model/DummyPayload.java | 1 + .../model/TerminatePayload.java | 4 + .../model/impl/PartFilePayload.java | 2 +- .../model/impl/SignalXmlFilePayload.java | 16 +++ .../service/signal/SignalRestService.java | 4 + .../credit_tracker/service/task/Pipeline.java | 10 ++ .../service/task/PipelinePackageManager.java | 34 ++++- .../credit_tracker/service/task/Stage.java | 1 + .../service/task/XmlMergeCoordinator.java | 127 +++++++++--------- .../task/setup/IPipeLineTerminator.java | 7 +- .../service/task/setup/PipeLineBuilderV2.java | 18 ++- 12 files changed, 164 insertions(+), 72 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java 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 443bebd..1077171 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,11 +1,17 @@ package ru.nbch.credit_tracker.config.task; import java.util.function.BiFunction; +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.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.task.XmlMergeCoordinator; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineImpl; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; +import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @Configuration @@ -14,4 +20,10 @@ public class PipelineStarterConfig { public BiFunction> pipelineStarterFactory(StaxXmlProcessor xml) { return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName); } + + @Bean + public Supplier> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) { + return () -> new XmlMergeCoordinator(creditTrackerSettings); + } + } diff --git a/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java b/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java index ed6e5da..137c6c8 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/DummyPayload.java @@ -3,6 +3,7 @@ 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; diff --git a/src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java b/src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java new file mode 100644 index 0000000..f115291 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java @@ -0,0 +1,4 @@ +package ru.nbch.credit_tracker.model; + +public interface TerminatePayload { +} 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 4897a42..9998061 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 @@ -25,6 +25,6 @@ public class PartFilePayload implements StagePayload { @Override public boolean isPoison() { - return false; + return poison; } } diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java new file mode 100644 index 0000000..4f1694a --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java @@ -0,0 +1,16 @@ +package ru.nbch.credit_tracker.model.impl; + +import java.nio.file.Path; +import ru.nbch.credit_tracker.model.TerminatePayload; + +public class SignalXmlFilePayload implements TerminatePayload { + private final Path signalXmlFilePath; + + public SignalXmlFilePayload(Path signalXmlFilePath) { + this.signalXmlFilePath = signalXmlFilePath; + } + + public Path getSignalXmlFilePath() { + return signalXmlFilePath; + } +} 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 ad08a0f..2b6143f 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 @@ -70,6 +70,10 @@ public class SignalRestService { */ public boolean setupMonitoring(Long packageId) throws Exception { String path = signalService.writeSignals(packageId); + return setupMonitoring(path); + } + + public boolean setupMonitoring(String path) throws Exception { Resource request = compressFile(path); MultiValueMap body = new LinkedMultiValueMap<>(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java b/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java index f90cd13..9b1c135 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java @@ -1,9 +1,11 @@ package ru.nbch.credit_tracker.service.task; import java.util.List; +import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; public class Pipeline { private Runnable starter; + private IPipeLineTerminator terminator; private List> stages; public Pipeline() { @@ -21,6 +23,14 @@ public class Pipeline { this.starter = starter; } + public IPipeLineTerminator getTerminator() { + return terminator; + } + + public void setTerminator(IPipeLineTerminator terminator) { + this.terminator = terminator; + } + public List> getStages() { return stages; } 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 32f3178..4bef0fc 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 @@ -5,16 +5,20 @@ import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; +import java.util.function.Supplier; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.task.StageQueueFactory; 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.SaveSubjectHandler; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; +import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @@ -24,11 +28,13 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils; public class PipelinePackageManager { private final ApplicationContext applicationContext; private final StaxXmlProcessor xmlProcessor; + private final SignalRestService signalRestService; private final StageQueueFactory saveSubjectQueueFactory; private final StageQueueFactory idEnricherQueueFactory; private final StageQueueFactory sendSignalQueueFactory; private final SaveSubjectHandler saveSubjectHandler; private final FidIdEnricher fidIdEnricherHandler; + private final Supplier> pipelineTerminatorFactory; private final BiFunction> pipelineStarterFactory; private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> { Thread t = new Thread(r); @@ -41,18 +47,22 @@ public class PipelinePackageManager { public PipelinePackageManager(ApplicationContext applicationContext, StaxXmlProcessor xmlProcessor, + SignalRestService signalRestService, @Qualifier("saveSubjectQueueFactory") StageQueueFactory saveSubjectQueueFactory, @Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, + Supplier> pipelineTerminatorFactory, BiFunction> pipelineStarterFactory) { this.applicationContext = applicationContext; this.xmlProcessor = xmlProcessor; + this.signalRestService = signalRestService; this.saveSubjectQueueFactory = saveSubjectQueueFactory; this.idEnricherQueueFactory = idEnricherQueueFactory; this.sendSignalQueueFactory = sendSignalQueueFactory; this.saveSubjectHandler = saveSubjectHandler; this.fidIdEnricherHandler = fidIdEnricherHandler; + this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.pipelineStarterFactory = pipelineStarterFactory; this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS); } @@ -70,20 +80,32 @@ public class PipelinePackageManager { .handler(fidIdEnricherHandler) .poison(PartFilePayload.POISON) .andThen() - .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)); + .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)) + .pollFinishingQueue(pipelineTerminatorFactory.get()) + .build(); + executorService.submit(pipeline.getStarter()); - pipeline.getLast() - .doneFuture() - .orTimeout(5, TimeUnit.SECONDS) + executorService.submit(pipeline.getTerminator()); + + pipeline.getTerminator() + .getDone() + .orTimeout(5, TimeUnit.MINUTES) .whenComplete((v, e) -> { if (e != null) { log.error(ExceptionUtils.getStackTrace(e)); - } else { - log.info("packageId={}: pipeline finished", packageId); } + SignalXmlFilePayload res = (SignalXmlFilePayload) v; + log.trace("Signal file by path: {} is completed", res.getSignalXmlFilePath()); + try { + signalRestService.setupMonitoring(res.getSignalXmlFilePath().toString()); + } catch (Exception ex) { + throw new RuntimeException(ex); + } + log.info("packageId={}: pipeline finished", packageId); pipeline.getStages().forEach(Stage::close); semaphore.release(); }); + } catch (Exception e) { semaphore.release(); throw e; 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 32e1ce8..4c2bbf3 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 @@ -63,6 +63,7 @@ public class Stage implements Ru } if (b == null) continue; if (b.isPoison()) { + log.debug("Got poison"); // дождаться всех in-flight задач for (Future f : inflight) { if (wasInterrupted()) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java index 2ec31a1..f08508c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java @@ -11,8 +11,7 @@ import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import javax.xml.stream.XMLInputFactory; import javax.xml.stream.XMLOutputFactory; @@ -22,77 +21,42 @@ import javax.xml.stream.XMLStreamReader; import javax.xml.stream.XMLStreamWriter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.model.impl.PartFilePayload; +import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; +import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; @Slf4j -@Service -public final class XmlMergeCoordinator implements InitializingBean -// implements IPipeLineTerminator -{ - private final BlockingQueue mergerQueue; +//@Service +public final class XmlMergeCoordinator implements IPipeLineTerminator { private final Path mergedDir; private final Charset encoding = Charset.forName("windows-1251"); private final int batchSize = 10; - private final CreditTrackerSettings config; + private final CompletableFuture done = new CompletableFuture<>(); + private Path out; + private BlockingQueue inQueue; - private final ExecutorService mergerExec = Executors.newSingleThreadExecutor(); - - public XmlMergeCoordinator(BlockingQueue mergerQueue, - CreditTrackerSettings config) throws IOException { - this.mergerQueue = mergerQueue; + public XmlMergeCoordinator(CreditTrackerSettings config) { this.mergedDir = Path.of(config.getLoadDir() + File.separator + "tmp_folder" + File.separator + "merged" + File.separator); - this.config = config; - Files.createDirectories(mergedDir); } - public void start() { - mergerExec.submit(this::runLoop); + + @Override + public void setPollingQueue(BlockingQueue mergerQueue) { + this.inQueue = mergerQueue; } - public void shutdown() { - mergerExec.shutdownNow(); - } - - private void runLoop() { - List batch = new ArrayList<>(batchSize); - while (!Thread.currentThread().isInterrupted()) { - try { - PartFilePayload first = mergerQueue.poll(5, TimeUnit.SECONDS); - if (first != null) batch.add(first); - // добираем до batchSize, но не ждём бесконечно - while (batch.size() < batchSize) { - PartFilePayload p = mergerQueue.poll(1000, TimeUnit.MILLISECONDS); - if (p == null) { - if (first != null) { - log.debug("Timeout poll mergerQueue, break"); - } - break; - } - batch.add(p); - } - if (!batch.isEmpty()) { - log.debug("Starting merge batch; size: {}", batch.size()); - mergeBatch(batch); -// удаляем исходники после успешного мерджа - for (PartFilePayload p : batch) Files.deleteIfExists(p.getPartPath()); - batch.clear(); - } - } catch (InterruptedException ie) { - Thread.currentThread().interrupt(); - } catch (Exception e) { - log.error("Error: {}", ExceptionUtils.getStackTrace(e)); - batch.clear(); - } - } + @Override + public CompletableFuture getDone() { + return done; } private void mergeBatch(List inputs) throws Exception { Long packageId = inputs.stream().findFirst().get().getPackageId(); String outName = "merged_" + packageId + ".xml"; - Path out = mergedDir.resolve(outName); + if (out == null) { + out = mergedDir.resolve(outName); + } XMLOutputFactory xof = XMLOutputFactory.newFactory(); try (OutputStream os = Files.newOutputStream(out, StandardOpenOption.CREATE, StandardOpenOption.APPEND)) { @@ -155,12 +119,51 @@ public final class XmlMergeCoordinator implements InitializingBean } @Override - public void afterPropertiesSet() throws Exception { - mergerExec.submit(this::runLoop); + public void run() { + List batch = new ArrayList<>(batchSize); + try { + Files.createDirectories(mergedDir); + } catch (IOException e) { + throw new RuntimeException(e); + } + while (!Thread.currentThread().isInterrupted()) { + try { + PartFilePayload part = inQueue.poll(5, TimeUnit.SECONDS); + + if (part == null) { + continue; + } + if (part.isPoison()) { + done.complete(new SignalXmlFilePayload(out)); + return; + } + batch.add(part); + // добираем до batchSize, но не ждём бесконечно + //todo подозрительное действие, прочекать не будет ли мешать poison + while (batch.size() < batchSize) { + PartFilePayload p = inQueue.poll(1000, TimeUnit.MILLISECONDS); + if (p == null) { + break; + } + if (p.isPoison()) { + done.complete(new SignalXmlFilePayload(out)); + return; + } + batch.add(p); + } + if (!batch.isEmpty()) { + log.debug("Starting merge batch; size: {}", batch.size()); + mergeBatch(batch); +// удаляем исходники после успешного мерджа + for (PartFilePayload p : batch) Files.deleteIfExists(p.getPartPath()); + batch.clear(); + } + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } catch (Exception e) { + log.error("Error: {}", ExceptionUtils.getStackTrace(e)); + batch.clear(); + } + } } -// -// @Override -// public void startPollingQueue(BlockingQueue queue) { -// -// } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java index f548630..803f7d0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java @@ -1,8 +1,11 @@ package ru.nbch.credit_tracker.service.task.setup; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; import ru.nbch.credit_tracker.model.StagePayload; +import ru.nbch.credit_tracker.model.TerminatePayload; -public interface IPipeLineTerminator { - void startPollingQueue(BlockingQueue queue); +public interface IPipeLineTerminator extends Runnable { + void setPollingQueue(BlockingQueue queue); + CompletableFuture getDone(); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java index dfb6d33..46725a3 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java @@ -13,6 +13,8 @@ public class PipeLineBuilderV2> stages; private Stage firstStage; private BlockingQueue lastQueue; + private IPipeLineStarter pipeLineStarter; + private IPipeLineTerminator pipeLineTerminator; private PipeLineBuilderV2() { this.stages = new ArrayList<>(); @@ -31,20 +33,34 @@ public class PipeLineBuilderV2 HandlerStep stage(BlockingQueue outputQueue) { HandlerStep step = StageBuilderV2.create(lastQueue, outputQueue); + lastQueue = (BlockingQueue) outputQueue; ((StageBuilderV2) step).setParent((PipeLineBuilderV2) this); return step; } - public Pipeline fillInitialQueue(IPipeLineStarter pipeLineStarter) { + public PipeLineBuilderV2 fillInitialQueue(IPipeLineStarter pipeLineStarter) { Objects.requireNonNull(pipeLineStarter, "pipeLineStarter is required"); if (firstStage == null) { throw new IllegalStateException("Pipeline has no stages"); } + this.pipeLineStarter = pipeLineStarter; + return this; + } + + public PipeLineBuilderV2 pollFinishingQueue(IPipeLineTerminator pipeLineTerminator) { + Objects.requireNonNull(pipeLineTerminator, "pipeLineTerminator is required"); + this.pipeLineTerminator = pipeLineTerminator; + return this; + } + + public Pipeline build() { Pipeline p = new Pipeline(); p.setStarter(() -> { stages.forEach(Stage::startRunning); pipeLineStarter.startFillingQueue(firstStage.getIn()); }); + pipeLineTerminator.setPollingQueue(lastQueue); + p.setTerminator(pipeLineTerminator); p.setStages(stages); return p; }