added pipeline builder

added queue size
commented out merging
stage bugfix

src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java

@@ -0,0 +1,17 @@
+package ru.nbch.credit_tracker.config.task;
+
+import java.util.function.BiFunction;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+import ru.nbch.credit_tracker.service.task.impl.KickPipeLineImpl;
+import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+
+@Configuration
+public class PipelineStarterConfig {
+    @Bean
+    public BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml) {
+        return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName);
+    }
+}

src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java

@@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.config.task;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import ru.nbch.credit_tracker.model.SendSignalsPayload;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 
 @Configuration
@@ -9,11 +10,17 @@ public class StageConfig {
 
     @Bean("saveSubjectQueueFactory")
     public StageQueueFactory<SubjectPayload> saveSubjectQueueFactory() {
-        return new StageQueueFactory<>();
+        return new StageQueueFactory<>(5);
     }
 
     @Bean("idEnricherQueueFactory")
     public StageQueueFactory<SubjectPayload> idEnricherQueueFactory() {
         return new StageQueueFactory<>();
     }
+
+    @Bean("signalsQueueFactory")
+    public StageQueueFactory<SendSignalsPayload> sendSignalsQueueFactory() {
+        return new StageQueueFactory<>(10);
+    }
+
 }

src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java

@@ -7,9 +7,18 @@ import ru.nbch.credit_tracker.model.StagePayload;
 
 public class StageQueueFactory<T extends StagePayload> implements FactoryBean<BlockingQueue<T>> {
 
+    private final int capacity;
+    public StageQueueFactory(int capacity) {
+        this.capacity = capacity;
+    }
+
+    public StageQueueFactory() {
+        this(10);
+    }
+
     @Override
     public BlockingQueue<T> getObject() throws Exception {
-        return new LinkedBlockingQueue<>();
+        return new LinkedBlockingQueue<>(capacity);
     }
 
     @Override

src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java

@@ -106,7 +106,7 @@ public class SignalFacadeV3 {
 
             Long packageId = signalService.registerPackage(userId, packageName);
             log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name());
-            pipelinePackageManager.buildAndRun(packageId, filePath);
+            pipelinePackageManager.buildAndRun2(packageId, filePath);
         } catch (Exception e) {
             CTUtil.removeFile(filePath);
             if (e instanceof SignalClientException) {

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

@@ -2,6 +2,7 @@ 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) {

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

@@ -9,6 +9,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -16,9 +17,13 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Component;
 import ru.nbch.credit_tracker.config.task.StageQueueFactory;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
+import ru.nbch.credit_tracker.model.SendSignalsPayload;
 import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler;
+import ru.nbch.credit_tracker.service.task.impl.dprctd.FidIdEnricher;
+import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
+import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 
 @Component
@@ -28,7 +33,10 @@ public class PipelinePackageManager {
     private final StaxXmlProcessor xmlProcessor;
     private final StageQueueFactory<SubjectPayload> saveSubjectQueueFactory;
     private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
+    private final StageQueueFactory<SendSignalsPayload> sendSignalQueueFactory;
     private final SaveSubjectHandler saveSubjectHandler;
+    private final FidIdEnricher fidIdEnricherHandler;
+    private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
     private final ExecutorService executorService = Executors.newFixedThreadPool(10);
     private final static int BATCH_SIZE = 10_000;
 
@@ -36,12 +44,17 @@ public class PipelinePackageManager {
                                   StaxXmlProcessor xmlProcessor,
                                   @Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
                                   @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
-                                  SaveSubjectHandler saveSubjectHandler) {
+                                  @Qualifier("signalsQueueFactory") StageQueueFactory<SendSignalsPayload> sendSignalQueueFactory,
+                                  SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
+                                  BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory) {
         this.applicationContext = applicationContext;
         this.xmlProcessor = xmlProcessor;
         this.saveSubjectQueueFactory = saveSubjectQueueFactory;
         this.idEnricherQueueFactory = idEnricherQueueFactory;
+        this.sendSignalQueueFactory = sendSignalQueueFactory;
         this.saveSubjectHandler = saveSubjectHandler;
+        this.fidIdEnricherHandler = fidIdEnricherHandler;
+        this.pipelineStarterFactory = pipelineStarterFactory;
     }
 
     public void buildAndRun(Long packageId, String pkgPath) throws Exception {
@@ -85,4 +98,20 @@ public class PipelinePackageManager {
 //        future.get(10_000, TimeUnit.SECONDS);
         log.info("Subjects were registered. PackageId: {}", packageId);
     }
+
+    public void buildAndRun2(Long packageId, String pkgPath) throws Exception {
+        BlockingQueue<SubjectPayload> idEnricherQueue = idEnricherQueueFactory.getObject();
+        BlockingQueue<SendSignalsPayload> signalsQueue = sendSignalQueueFactory.getObject();
+        var startPipeLine = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueue)
+                .handler(saveSubjectHandler)
+                .poison(SubjectPayload.POISON)
+                .andThen()
+            .stage(idEnricherQueue, signalsQueue)
+                .handler(fidIdEnricherHandler)
+                .poison(SendSignalsPayload.POISON)
+                .andThen()
+            .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath));
+        executorService.submit(startPipeLine);
+        log.info("Subjects were registered. PackageId: {}", packageId);
+    }
 }

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

@@ -33,6 +33,10 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         this.POISON = POISON;
     }
 
+    public void startRunning() {
+        ioPool.submit(this);
+    }
+
     @Override
     public void run() {
         List<Future<Z>> inflight = new ArrayList<>();
@@ -44,7 +48,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
                 if (b.isPoison()) {
                     // дождаться всех in-flight задач
                     for (Future<Z> f : inflight) {
-                        putQuiet(out, f.get());
+                        f.get();
                     }
                     putQuiet(out, POISON);
                     return;
@@ -115,6 +119,10 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         }
     }
 
+    public BlockingQueue<X> getIn() {
+        return in;
+    }
+
     static void sleep(long ms) {
         try {
             Thread.sleep(ms);

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

@@ -1,13 +1,5 @@
 package ru.nbch.credit_tracker.service.task;
 
-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.PartFilePayload;
-
-import javax.xml.stream.*;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
@@ -22,9 +14,20 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLOutputFactory;
+import javax.xml.stream.XMLStreamConstants;
+import javax.xml.stream.XMLStreamException;
+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 ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.model.PartFilePayload;
 
 @Slf4j
-@Service
+//@Service
 public final class XmlMergeCoordinator implements InitializingBean {
     private final BlockingQueue<PartFilePayload> mergerQueue;
     private final Path mergedDir;

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

@@ -0,0 +1,61 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+
+public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final Long packageId;
+    private final String pkgPath;
+
+    private final StaxXmlProcessor xmlProcessor;
+    private final static int BATCH_SIZE = 10_000;
+
+
+    public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, String pkgPath) {
+        this.packageId = packageId;
+        this.pkgPath = pkgPath;
+        this.xmlProcessor = xmlProcessor;
+    }
+
+    @Override
+    public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
+        log.info("Start registering subjects. PackageId: {}", packageId);
+        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(Path.of(pkgPath)), 8 * 1024 * 1024)) {
+            xmlProcessor.read(in, MonitoringRequest.class,
+                request -> {
+                    if (!request.getSubjects().getSubject().isEmpty()) {
+                        List<SubjectProfile> subjectProfileBatch = request.getSubjects().getSubject().stream().map(subject ->
+                            SubjectProfile.builder()
+                                .identifier(subject.getId())
+                                .phone(subject.getClientId())
+                                .build()
+
+                        ).collect(Collectors.toList());
+                        log.trace("Reading subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
+                        firstQueue.add(new SubjectPayload(packageId, subjectProfileBatch));
+                        request.getSubjects().getSubject().clear();
+                    }
+                },
+                request -> request.getSubjects().getSubject().isEmpty()
+                ,
+                BATCH_SIZE, "/MonitoringRequest/Subjects/Subject");
+        } catch (IOException e) {
+            throw new RuntimeException("Error while parsing monitoring request", e);
+        } finally {
+            firstQueue.add(new SubjectPayload(true));
+        }
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FidIdEnricher.java

@@ -9,13 +9,11 @@ import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.util.List;
 import java.util.UUID;
-import java.util.concurrent.BlockingQueue;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.builder.PersonFactoryBuilderV2;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.external_request.signals.Person;
-import ru.nbch.credit_tracker.model.PartFilePayload;
 import ru.nbch.credit_tracker.model.SendSignalsPayload;
 import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
@@ -30,15 +28,15 @@ public class FidIdEnricher implements Handler<SubjectPayload, SendSignalsPayload
     private final PersonService personService;
     private final SignalService signalService;
     private final CreditTrackerSettings config;
-    private final BlockingQueue<PartFilePayload> mergerQueue;
+//    private final BlockingQueue<PartFilePayload> mergerQueue;
     public FidIdEnricher(PersonService personService,
                          SignalService signalService,
-                         CreditTrackerSettings config,
-                         BlockingQueue<PartFilePayload> mergerQueue) {
+                         CreditTrackerSettings config//,
+                         /*BlockingQueue<PartFilePayload> mergerQueue*/) {
         this.personService = personService;
         this.signalService = signalService;
         this.config = config;
-        this.mergerQueue = mergerQueue;
+//        this.mergerQueue = mergerQueue;
     }
 
     @Override
@@ -81,10 +79,10 @@ public class FidIdEnricher implements Handler<SubjectPayload, SendSignalsPayload
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
-        try {
-            mergerQueue.put(new PartFilePayload(done, packageId, null));
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }
+//        try {
+//            mergerQueue.put(new PartFilePayload(done, packageId, null));
+//        } catch (InterruptedException e) {
+//            throw new RuntimeException(e);
+//        }
     }
 }

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

@@ -0,0 +1,8 @@
+package ru.nbch.credit_tracker.service.task.setup;
+
+import java.util.concurrent.BlockingQueue;
+import ru.nbch.credit_tracker.model.StagePayload;
+
+public interface IPipeLineStarter<Q extends StagePayload> {
+    void startFillingQueue(BlockingQueue<Q> queue);
+}

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

@@ -0,0 +1,55 @@
+package ru.nbch.credit_tracker.service.task.setup;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import ru.nbch.credit_tracker.model.StagePayload;
+import ru.nbch.credit_tracker.service.task.Stage;
+import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep;
+
+public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePayload> {
+    private final List<Stage<?, ?>> stages;
+    private Stage<FIRST, ?> firstStage;
+
+    private PipeLineBuilderV2() {
+        this.stages = new ArrayList<>();
+    }
+
+    public static <I extends StagePayload, O extends StagePayload>
+    HandlerStep<I, I, O> first(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
+        PipeLineBuilderV2<I, O> builder = new PipeLineBuilderV2<>();
+        HandlerStep<I, I, O> step = StageBuilderV2.create(inputQueue, outputQueue);
+        ((StageBuilderV2<I, I, O>) step).setParent(builder);
+        return step;
+    }
+
+    @SuppressWarnings("unchecked")
+    public <NEXT extends StagePayload>
+    HandlerStep<FIRST, LAST, NEXT> stage(BlockingQueue<LAST> inputQueue, BlockingQueue<NEXT> outputQueue) {
+        HandlerStep<FIRST, LAST, NEXT> step = StageBuilderV2.create(inputQueue, outputQueue);
+        ((StageBuilderV2<FIRST, LAST, NEXT>) step).setParent((PipeLineBuilderV2<FIRST, NEXT>) this);
+        return step;
+    }
+
+    public Runnable fillInitialQueue(IPipeLineStarter<FIRST> pipeLineStarter) {
+        Objects.requireNonNull(pipeLineStarter, "pipeLineStarter is required");
+        if (firstStage == null) {
+            throw new IllegalStateException("Pipeline has no stages");
+        }
+        return () -> {
+            stages.forEach(Stage::startRunning);
+            pipeLineStarter.startFillingQueue(firstStage.getIn());
+        };
+    }
+
+    @SuppressWarnings("unchecked")
+    <I extends StagePayload, O extends StagePayload>
+    void addStage(Stage<I, O> stage) {
+        Objects.requireNonNull(stage, "stage is required");
+        if (stages.isEmpty()) {
+            this.firstStage = (Stage<FIRST, ?>) stage;
+        }
+        stages.add(stage);
+    }
+}

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

@@ -0,0 +1,76 @@
+package ru.nbch.credit_tracker.service.task.setup;
+
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import ru.nbch.credit_tracker.model.StagePayload;
+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> {
+
+    private PipeLineBuilderV2<FIRST, O> parent;
+    private final BlockingQueue<I> inputQueue;
+    private final BlockingQueue<O> outputQueue;
+    private Handler<I, O> handler;
+    private O poison;
+
+    private StageBuilderV2(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
+        this.inputQueue = Objects.requireNonNull(inputQueue, "inputQueue is required");
+        this.outputQueue = Objects.requireNonNull(outputQueue, "outputQueue is required");
+    }
+
+    /** Стартовая фабрика возвращает первый шаг — HandlerStep */
+    public static <FIRST extends StagePayload, I extends StagePayload, O extends StagePayload>
+    HandlerStep<FIRST, I, O> create(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
+        return new StageBuilderV2<>(inputQueue, outputQueue);
+    }
+
+    void setParent(PipeLineBuilderV2<FIRST, O> parent) {
+        this.parent = Objects.requireNonNull(parent, "parent is required");
+    }
+
+    // ===================== ШАГИ =====================
+
+    /** Шаг 1: задать обработчик */
+    @Override
+    public PoisonStep<FIRST, I, 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;
+    }
+
+    /** Шаг 3: собрать Stage и вернуться к PipeLineBuilder */
+    @Override
+    public PipeLineBuilderV2<FIRST, O> andThen() {
+        if (parent == null) {
+            throw new IllegalStateException("parent PipeLineBuilder is not set");
+        }
+        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);
+        parent.addStage(stage);
+        return parent;
+    }
+
+    // ===================== Step-interfaces =====================
+
+}

src/main/java/ru/nbch/credit_tracker/service/task/setup/step/BuildStep.java

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

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

@@ -0,0 +1,8 @@
+package ru.nbch.credit_tracker.service.task.setup.step;
+
+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);
+}

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

@@ -0,0 +1,7 @@
+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);
+}