From 930a563172509e83a9f753ae730a51cd2953d7b1 Mon Sep 17 00:00:00 2001
From: etreschenkov <ETreschenkov@mfd.ru>
Date: Wed, 26 Nov 2025 14:59:32 +0300
Subject: [PATCH] ./

---
 .../config/task/StageConfig.java              |  6 +--
 .../model/impl/PartFilePayload.java           |  8 +++-
 .../service/task/PipelinePackageManager.java  | 10 +----
 .../credit_tracker/service/task/Stage.java    |  4 +-
 .../service/task/impl/FidIdEnricher.java      |  2 +-
 .../service/task/impl/KickPipeLineImpl.java   |  2 +-
 .../task/impl/XmlMergeCoordinator.java        | 43 ++++++++++++++++++-
 7 files changed, 58 insertions(+), 17 deletions(-)

diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java
index 517c872..79ad85f 100644
--- a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java
+++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java
@@ -15,17 +15,17 @@ public class StageConfig {
 
     @Bean("nameEnricherQueueFactory")
     public StageQueueFactory<SubjectPayload> nameEnricherQueueFactory() {
-        return new StageQueueFactory<>(20);
+        return new StageQueueFactory<>(10);
     }
 
     @Bean("idEnricherQueueFactory")
     public StageQueueFactory<SubjectPayload> idEnricherQueueFactory() {
-        return new StageQueueFactory<>(20);
+        return new StageQueueFactory<>(10);
     }
 
     @Bean("partFileQueueFactory")
     public StageQueueFactory<PartFilePayload> partFileQueueFactory() {
-        return new StageQueueFactory<>(500);
+        return new StageQueueFactory<>(100);
     }
 
 }
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 049dfea..52015ab 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
@@ -9,18 +9,24 @@ import ru.nbch.credit_tracker.model.StagePayload;
 public class PartFilePayload implements StagePayload {
     private final Path partPath;
     private final Long packageId;
+    private final String membercode;
+    private final String packageName;
     private final PoisonType poison;
     public static final PartFilePayload POISON = new PartFilePayload(PoisonType.NORMALLY);
 
     public PartFilePayload(PoisonType poison) {
         this.poison = poison;
+        this.membercode = null;
+        this.packageName = null;
         this.partPath = null;
         this.packageId = null;
     }
 
-    public PartFilePayload(Path partPath, Long packageId) {
+    public PartFilePayload(Path partPath, Long packageId, String membercode, String packageName) {
         this.partPath = partPath;
         this.packageId = packageId;
+        this.membercode = membercode;
+        this.packageName = packageName;
         this.poison = PoisonType.NONE;
     }
 
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 354ca1d..13678c1 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
@@ -21,7 +21,6 @@ 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.FidNameEnricher;
 import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
-import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler;
 import ru.nbch.credit_tracker.service.task.impl.Saver;
 import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
@@ -34,11 +33,9 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 public class PipelinePackageManager {
     private final SignalRestService signalRestService;
     private final FlagsEnricher flagsEnricherV2;
-    private final StageQueueFactory<SubjectPayload> saveSubjectQueueFactory;
     private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
     private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory;
     private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
-    private final SaveSubjectHandler saveSubjectHandler;
     private final UserPackageMapper userPackageMapper;
     private final FidIdEnricher fidIdEnricherHandler;
     private final FidNameEnricher fidNameEnricher;
@@ -59,14 +56,11 @@ public class PipelinePackageManager {
     private final Semaphore semaphore;
     private final PipelineBackoff pipelineBackoff;
     private final Saver saver;
-    private final static int MAX_PARALLEL_REQUESTS = 3;
 
     public PipelinePackageManager(SignalRestService signalRestService,
-                                  @Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
                                   @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
                                   @Qualifier("nameEnricherQueueFactory") StageQueueFactory<SubjectPayload> nameEnricherQueueFactory,
                                   @Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
-                                  SaveSubjectHandler saveSubjectHandler,
                                   UserPackageMapper userPackageMapper,
                                   FidIdEnricher fidIdEnricherHandler,
                                   Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
@@ -75,11 +69,9 @@ public class PipelinePackageManager {
                                   CreditTrackerSettings creditTrackerSettings,
                                   PipelineBackoff pipelineBackoff, Saver saver) {
         this.signalRestService = signalRestService;
-        this.saveSubjectQueueFactory = saveSubjectQueueFactory;
         this.idEnricherQueueFactory = idEnricherQueueFactory;
         this.nameEnricherQueueFactory = nameEnricherQueueFactory;
         this.sendSignalQueueFactory = sendSignalQueueFactory;
-        this.saveSubjectHandler = saveSubjectHandler;
         this.userPackageMapper = userPackageMapper;
         this.fidIdEnricherHandler = fidIdEnricherHandler;
         this.pipelineTerminatorFactory = pipelineTerminatorFactory;
@@ -102,7 +94,7 @@ public class PipelinePackageManager {
         try {
             Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                     .handler(fidNameEnricher)
-                    .handlerResult(flagsEnricherV2)
+//                    .handlerResult(flagsEnricherV2)
                     .andThen()
                 .stage(sendSignalQueueFactory.getObject())
                     .handler(fidIdEnricherHandler)
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 8fa190c..44d1f40 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
@@ -73,6 +73,8 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
             int counter = 0;
             while (true) {
                 X b = takeQuiet();
+                counter++;
+                log.trace("Take item from queue counter: {}", counter);
                 if (wasInterrupted()) {
                     break;
                 }
@@ -91,9 +93,9 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
                     log.info("stage {} finished.", dispatcherName);
                     return;
                 }
-                counter++;
                 int finalCounter = counter;
                 AtomicInteger retryForSingleTask = new AtomicInteger(0);
+                log.trace("submit task");
                 Future<Void> f = ioPool.submit(() -> {
                     while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) {
                         try {
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 c7b9d1d..79785f2 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
@@ -45,7 +45,7 @@ public class FidIdEnricher implements Handler<SubjectPayload, PartFilePayload> {
         log.debug("personOnProcessing size: {}", personOnProcessing.size());
         Long packageId = taskPayload.getPackageId();
         Path partPath = writeSignals(personOnProcessing, taskPayload);
-        return new PartFilePayload(partPath, packageId);
+        return new PartFilePayload(partPath, packageId, taskPayload.getMemberCode(), taskPayload.getPackageName());
     }
 
     private Path writeSignals(List<Person> persons, SubjectPayload taskPayload) {
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 a76b344..f890a9e 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
@@ -97,7 +97,7 @@ public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
         }
     }
 //    @Override
-    public void startFillingQueueXml(BlockingQueue<SubjectPayload> firstQueue) {
+    public void startFillingQueueXmnl(BlockingQueue<SubjectPayload> firstQueue) {
         log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath);
         UserPackage userPackage = userPackageMapper.findPackage(packageId);
         AtomicInteger counter = new AtomicInteger(0);
diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java
index c6c5e79..57725ae 100644
--- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java
@@ -1,7 +1,10 @@
 package ru.nbch.credit_tracker.service.task.impl;
 
+import java.io.BufferedWriter;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
 import java.nio.charset.Charset;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -46,6 +49,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
                 }
                 if (part.isPoison()) {
                     log.debug("Got poison");
+                    finishMerge();
                     done.complete(new SignalXmlFilePayload(out));
                     log.info("Stage time stats: {}", stopWatch.info());
                     return;
@@ -70,13 +74,37 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
         log.debug("Starting merge");
         Long packageId = input.getPackageId();
         Path partFilePath = input.getPartPath();
+        boolean first = false;
         if (out == null) {
+            first = true;
             out = partFilePath.resolveSibling("merged_" + packageId + ".xml");
         }
 
         XMLOutputFactory xof = XMLOutputFactory.newFactory();
         try (OutputStream os = Files.newOutputStream(out, StandardOpenOption.CREATE, StandardOpenOption.APPEND)) {
             XMLStreamWriter w = xof.createXMLStreamWriter(os, encoding.name());
+
+            if (first) {
+                w.writeStartDocument(encoding.name(), "1.0");
+
+                w.writeStartElement("Signals");
+                w.writeAttribute("online", "2");
+                w.writeAttribute("processWithErrors", "1");
+
+                w.writeStartElement("Id");
+                w.writeCharacters("%s_%s".formatted(input.getMembercode(), input.getPackageName()));
+                w.writeEndElement(); // </Id>
+
+                w.writeStartElement("Auth");
+                w.writeStartElement("User");
+                w.writeCharacters(config.getSignalsuser());
+                w.writeEndElement(); // </User>
+                w.writeEndElement(); // </Auth>
+
+                // <Conditions><Persons>
+                w.writeStartElement("Conditions");
+                w.writeStartElement("Persons");
+            }
             log.debug("Copy person from file {}", input.getPartPath());
             copyPersonsChildren(input.getPartPath(), w);
             w.flush();
@@ -85,6 +113,20 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
         Files.deleteIfExists(input.getPartPath());
     }
 
+    private void finishMerge() throws Exception {
+        if (out == null) {
+            return;
+        }
+
+        try (Writer w = new BufferedWriter(new OutputStreamWriter(
+            Files.newOutputStream(out, StandardOpenOption.APPEND),
+            encoding))) {
+
+            w.write("</Persons>");
+            w.write("</Conditions>");
+            w.write("</Signals>");
+        }
+    }
     /**
      * Копирует всё содержимое внутри <Persons>...</Persons> входного файла
      * в текущий writer, но без повторного открытия/закрытия корневого тега.
@@ -123,7 +165,6 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
             r.close();
         }
     }
-
     private void writeStart(XMLStreamWriter w, XMLStreamReader r) throws XMLStreamException {
         w.writeStartElement(r.getLocalName());
         for (int i = 0; i < r.getAttributeCount(); i++) {