From 395613ae01aa653495ea5e6c1f11a23791eb303a Mon Sep 17 00:00:00 2001
From: etreschenkov <ETreschenkov@mfd.ru>
Date: Tue, 11 Nov 2025 16:47:58 +0300
Subject: [PATCH] flag calc

---
 .../config/task/StageConfig.java              |  5 ----
 .../service/task/PipelinePackageManager.java  | 27 +++++--------------
 2 files changed, 6 insertions(+), 26 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 d36faac..0c5e988 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
@@ -23,9 +23,4 @@ public class StageConfig {
         return new StageQueueFactory<>(10000);
     }
 
-    @Bean("flagQueueFactory")
-    public StageQueueFactory<SubjectPayload> flagQueueFactory() {
-        return new StageQueueFactory<>(100);
-    }
-
 }
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 c3016a7..625fced 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
@@ -8,7 +8,6 @@ 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;
@@ -16,62 +15,48 @@ 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.FlagsEnricher;
 import ru.nbch.credit_tracker.service.task.impl.FlagsEnricherV2;
 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;
 
 @Component
 @Slf4j
 public class PipelinePackageManager {
-    private final ApplicationContext applicationContext;
-    private final StaxXmlProcessor xmlProcessor;
     private final SignalRestService signalRestService;
+    private final FlagsEnricherV2 flagsEnricherV2;
     private final StageQueueFactory<SubjectPayload> saveSubjectQueueFactory;
     private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
     private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
-    private final StageQueueFactory<SubjectPayload> flagQueueFactory;
     private final SaveSubjectHandler saveSubjectHandler;
     private final FidIdEnricher fidIdEnricherHandler;
     private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
-    private final Supplier<FlagsEnricher> flagEnricherFactory;
     private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
-    private final FlagsEnricherV2 flagsEnricherV2;
     private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> {
         Thread t = new Thread(r);
         t.setName("pipe-starter-%d".formatted(t.threadId()));
         return t;
     });
     private final Semaphore semaphore;
-    private final static int BATCH_SIZE = 10_000;
     private final static int MAX_PARALLEL_REQUESTS = 3;
 
-    public PipelinePackageManager(ApplicationContext applicationContext,
-                                  StaxXmlProcessor xmlProcessor,
-                                  SignalRestService signalRestService,
+    public PipelinePackageManager(SignalRestService signalRestService,
                                   @Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
                                   @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
                                   @Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
-                                  @Qualifier("flagQueueFactory") StageQueueFactory<SubjectPayload> flagQueueFactory,
                                   SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
                                   Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
-                                  Supplier<FlagsEnricher> flagEnricherFactory,
-                                  BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory, FlagsEnricherV2 flagsEnricherV2) {
-        this.applicationContext = applicationContext;
-        this.xmlProcessor = xmlProcessor;
+                                  BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
+                                  FlagsEnricherV2 flagsEnricherV2) {
         this.signalRestService = signalRestService;
         this.saveSubjectQueueFactory = saveSubjectQueueFactory;
         this.idEnricherQueueFactory = idEnricherQueueFactory;
         this.sendSignalQueueFactory = sendSignalQueueFactory;
-        this.flagQueueFactory = flagQueueFactory;
         this.saveSubjectHandler = saveSubjectHandler;
         this.fidIdEnricherHandler = fidIdEnricherHandler;
         this.pipelineTerminatorFactory = pipelineTerminatorFactory;
-        this.flagEnricherFactory = flagEnricherFactory;
         this.pipelineStarterFactory = pipelineStarterFactory;
         this.flagsEnricherV2 = flagsEnricherV2;
         this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS);
@@ -86,7 +71,7 @@ public class PipelinePackageManager {
             Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                     .handler(saveSubjectHandler)
                     .poison(SubjectPayload.POISON)
-//                    .handlerResult(flagsEnricherV2)
+                    .handlerResult(flagsEnricherV2)
                     .andThen()
                 .stage(sendSignalQueueFactory.getObject())
                     .handler(fidIdEnricherHandler)
@@ -118,7 +103,7 @@ public class PipelinePackageManager {
                         pipeline.getStages().forEach(Stage::close);
                         semaphore.release();
                     }
-            });
+                });
         } catch (Exception e) {
             semaphore.release();
             throw e;