implement terminator

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<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml) {
         return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName);
     }
+
+    @Bean
+    public Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) {
+        return () -> new XmlMergeCoordinator(creditTrackerSettings);
+    }
+
 }

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;

src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java

@@ -0,0 +1,4 @@
+package ru.nbch.credit_tracker.model;
+
+public interface TerminatePayload {
+}

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

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;
+    }
+}

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<String, Resource> body = new LinkedMultiValueMap<>();

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<Stage<?, ?>> 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<Stage<?, ?>> getStages() {
         return stages;
     }

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<SubjectPayload> saveSubjectQueueFactory;
     private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
     private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
     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 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<SubjectPayload> saveSubjectQueueFactory,
                                   @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
                                   @Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
                                   SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
+                                  Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
                                   BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> 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()
+            executorService.submit(pipeline.getTerminator());
-                .doneFuture()
+
-                .orTimeout(5, TimeUnit.SECONDS)
+            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;

src/main/java/ru/nbch/credit_tracker/service/task/Stage.java

@@ -63,6 +63,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
                 }
                 if (b == null) continue;
                 if (b.isPoison()) {
+                    log.debug("Got poison");
                     // дождаться всех in-flight задач
                     for (Future<Z> f : inflight) {
                         if (wasInterrupted()) {

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.CompletableFuture;
-import java.util.concurrent.Executors;
 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
+//@Service
-public final class XmlMergeCoordinator implements InitializingBean
+public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload> {
-//    implements IPipeLineTerminator<PartFilePayload>
-{
-    private final BlockingQueue<PartFilePayload> mergerQueue;
     private final Path mergedDir;
     private final Charset encoding = Charset.forName("windows-1251");
     private final int batchSize = 10;
-    private final CreditTrackerSettings config;
+    private final CompletableFuture<SignalXmlFilePayload> done = new CompletableFuture<>();
+    private Path out;
+    private BlockingQueue<PartFilePayload> inQueue;
 
-    private final ExecutorService mergerExec = Executors.newSingleThreadExecutor();
+    public XmlMergeCoordinator(CreditTrackerSettings config)  {
-
-    public XmlMergeCoordinator(BlockingQueue<PartFilePayload> mergerQueue,
-                               CreditTrackerSettings config) throws IOException {
-        this.mergerQueue = mergerQueue;
         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<PartFilePayload> mergerQueue) {
+        this.inQueue = mergerQueue;
     }
 
-    public void shutdown() {
+    @Override
-        mergerExec.shutdownNow();
+    public CompletableFuture<SignalXmlFilePayload> getDone() {
-    }
+        return done;
-
-    private void runLoop() {
-        List<PartFilePayload> 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();
-            }
-        }
     }
 
     private void mergeBatch(List<PartFilePayload> 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 {
+    public void run() {
-        mergerExec.submit(this::runLoop);
+        List<PartFilePayload> 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<PartFilePayload> queue) {
-//
-//    }
 }

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<Q extends StagePayload> {
+public interface IPipeLineTerminator<Q extends StagePayload, X extends TerminatePayload> extends Runnable {
-    void startPollingQueue(BlockingQueue<Q> queue);
+    void setPollingQueue(BlockingQueue<Q> queue);
+    CompletableFuture<X> getDone();
 }

src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java

@@ -13,6 +13,8 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
     private final List<Stage<?, ?>> stages;
     private Stage<FIRST, ?> firstStage;
     private BlockingQueue<LAST> lastQueue;
+    private IPipeLineStarter<FIRST> pipeLineStarter;
+    private IPipeLineTerminator<LAST, ?> pipeLineTerminator;
 
     private PipeLineBuilderV2() {
         this.stages = new ArrayList<>();
@@ -31,20 +33,34 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
     public <NEXT extends StagePayload>
     HandlerStep<FIRST, LAST, NEXT> stage(BlockingQueue<NEXT> outputQueue) {
         HandlerStep<FIRST, LAST, NEXT> step = StageBuilderV2.create(lastQueue, outputQueue);
+        lastQueue = (BlockingQueue<LAST>) outputQueue;
         ((StageBuilderV2<FIRST, LAST, NEXT>) step).setParent((PipeLineBuilderV2<FIRST, NEXT>) this);
         return step;
     }
 
-    public Pipeline fillInitialQueue(IPipeLineStarter<FIRST> pipeLineStarter) {
+    public PipeLineBuilderV2<FIRST, LAST> fillInitialQueue(IPipeLineStarter<FIRST> pipeLineStarter) {
         Objects.requireNonNull(pipeLineStarter, "pipeLineStarter is required");
         if (firstStage == null) {
             throw new IllegalStateException("Pipeline has no stages");
         }
+        this.pipeLineStarter = pipeLineStarter;
+        return this;
+    }
+
+    public PipeLineBuilderV2<FIRST, LAST> pollFinishingQueue(IPipeLineTerminator<LAST, ?> 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;
     }