attempt to remove unused classes

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

@@ -2,7 +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.PartFilePayload;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 
 @Configuration
@@ -18,8 +18,8 @@ public class StageConfig {
         return new StageQueueFactory<>();
     }
 
-    @Bean("signalsQueueFactory")
+    @Bean("partFileQueueFactory")
-    public StageQueueFactory<SendSignalsPayload> sendSignalsQueueFactory() {
+    public StageQueueFactory<PartFilePayload> partFileQueueFactory() {
         return new StageQueueFactory<>(10);
     }
 

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

@@ -1,25 +0,0 @@
-package ru.nbch.credit_tracker.model;
-
-import ru.nbch.credit_tracker.entities.external_request.signals.Signals;
-
-import java.nio.file.Path;
-
-public class PartFilePayload {
-    private final Path partPath;
-    private final Long packageId;
-    private final Signals signals;
-
-    public PartFilePayload(Path partPath, Long packageId, Signals signals) {
-        this.partPath = partPath;
-        this.packageId = packageId;
-        this.signals = signals;
-    }
-
-    public Path getPartPath() {
-        return partPath;
-    }
-
-    public Long getPackageId() {
-        return packageId;
-    }
-}

src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java

@@ -0,0 +1,30 @@
+package ru.nbch.credit_tracker.model.impl;
+
+import java.nio.file.Path;
+import lombok.Getter;
+import ru.nbch.credit_tracker.model.StagePayload;
+
+@Getter
+public class PartFilePayload implements StagePayload {
+    private final Path partPath;
+    private final Long packageId;
+    private final boolean poison;
+    public static final PartFilePayload POISON = new PartFilePayload(true);
+
+    public PartFilePayload(boolean poison) {
+        this.poison = poison;
+        this.partPath = null;
+        this.packageId = null;
+    }
+
+    public PartFilePayload(Path partPath, Long packageId) {
+        this.partPath = partPath;
+        this.packageId = packageId;
+        this.poison = false;
+    }
+
+    @Override
+    public boolean isPoison() {
+        return false;
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java

@@ -14,6 +14,7 @@ import java.sql.SQLException;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.*;
+import ru.nbch.credit_tracker.model.SubjectProfile;
 
 @Slf4j
 @Service
@@ -40,20 +41,20 @@ public class FlagsService2 {
     }
 
     // TODO optimize
-    public void fillFlagsData(List<PhoneFidMap> subjects) {
+    public void fillFlagsData(List<SubjectProfile> subjects) {
-        List<Integer> fids = subjects.stream().map(PhoneFidMap::getFid).filter(Objects::nonNull).toList();
+        List<Integer> fids = subjects.stream().map(SubjectProfile::getFid).filter(Objects::nonNull).toList();
         Map<Integer, FlagsService.FlagsData> fidsflagsDataMap = findFlags(fids);
 
         fidsflagsDataMap.forEach((fid, data) -> {
             subjects.stream().filter(s-> s.getFid() != null && s.getFid().equals(fid)).forEach(phoneFidMap -> {
-                phoneFidMap.setFlagDebt5000(data.getFlagDebt5000());
+//                phoneFidMap.setFlagDebt5000(data.getFlagDebt5000());
-                phoneFidMap.setFlag90days(data.getFlag90days());
+//                phoneFidMap.setFlag90days(data.getFlag90days());
-                phoneFidMap.setCalculatedAt(LocalDateTime.now());
+//                phoneFidMap.setCalculatedAt(LocalDateTime.now());
             });
         });
 
         subjects.stream().filter(s-> s.getFid() == null).forEach(phoneFidMap -> {
-            phoneFidMap.setCalculatedAt(LocalDateTime.now());
+//            phoneFidMap.setCalculatedAt(LocalDateTime.now());
         });
     }
 

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

@@ -10,10 +10,10 @@ 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.SendSignalsPayload;
+import ru.nbch.credit_tracker.model.impl.PartFilePayload;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+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.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;
@@ -26,7 +26,7 @@ public class PipelinePackageManager {
     private final StaxXmlProcessor xmlProcessor;
     private final StageQueueFactory<SubjectPayload> saveSubjectQueueFactory;
     private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
-    private final StageQueueFactory<SendSignalsPayload> sendSignalQueueFactory;
+    private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
     private final SaveSubjectHandler saveSubjectHandler;
     private final FidIdEnricher fidIdEnricherHandler;
     private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
@@ -39,7 +39,7 @@ public class PipelinePackageManager {
                                   StaxXmlProcessor xmlProcessor,
                                   @Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
                                   @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
-                                  @Qualifier("signalsQueueFactory") StageQueueFactory<SendSignalsPayload> sendSignalQueueFactory,
+                                  @Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
                                   SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
                                   BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory) {
         this.applicationContext = applicationContext;
@@ -65,7 +65,7 @@ public class PipelinePackageManager {
                     .andThen()
                 .stage(sendSignalQueueFactory.getObject())
                     .handler(fidIdEnricherHandler)
-                    .poison(SendSignalsPayload.POISON)
+                    .poison(PartFilePayload.POISON)
                     .andThen()
                 .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath));
             executorService.submit(pipeline.getStarter());

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

@@ -23,12 +23,15 @@ 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.PartFilePayload;
+import ru.nbch.credit_tracker.model.impl.PartFilePayload;
 
 @Slf4j
-//@Service
+@Service
-public final class XmlMergeCoordinator implements InitializingBean {
+public final class XmlMergeCoordinator implements InitializingBean
+//    implements IPipeLineTerminator<PartFilePayload>
+{
     private final BlockingQueue<PartFilePayload> mergerQueue;
     private final Path mergedDir;
     private final Charset encoding = Charset.forName("windows-1251");
@@ -155,4 +158,9 @@ public final class XmlMergeCoordinator implements InitializingBean {
     public void afterPropertiesSet() throws Exception {
         mergerExec.submit(this::runLoop);
     }
+//
+//    @Override
+//    public void startPollingQueue(BlockingQueue<PartFilePayload> queue) {
+//
+//    }
 }

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

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service.task.impl.dprctd;
+package ru.nbch.credit_tracker.service.task.impl;
 
 import java.io.IOException;
 import java.io.OutputStream;
@@ -14,7 +14,7 @@ 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.SendSignalsPayload;
+import ru.nbch.credit_tracker.model.impl.PartFilePayload;
 import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.indic.PersonServiceV2;
@@ -23,9 +23,10 @@ import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2;
 
 @Service
 @Slf4j
-public class FidIdEnricher implements Handler<SubjectPayload, SendSignalsPayload> {
+public class FidIdEnricher implements Handler<SubjectPayload, PartFilePayload> {
     private final PersonServiceV2 personService;
     private final CreditTrackerSettings config;
+
     public FidIdEnricher(PersonServiceV2 personService,
                          CreditTrackerSettings config) {
         this.personService = personService;
@@ -33,7 +34,7 @@ public class FidIdEnricher implements Handler<SubjectPayload, SendSignalsPayload
     }
 
     @Override
-    public SendSignalsPayload handle(SubjectPayload taskPayload) {
+    public PartFilePayload handle(SubjectPayload taskPayload) {
         List<SubjectProfile> subjectsProfile = taskPayload.getSubjects();
         log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
         personService.filledPassportsData(subjectsProfile);
@@ -42,11 +43,12 @@ public class FidIdEnricher implements Handler<SubjectPayload, SendSignalsPayload
             .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build())
             .toList();
         log.debug("personOnProcessing size: {}", personOnProcessing.size());
-        writeSignals(personOnProcessing, taskPayload.getPackageId());
+        Long packageId = taskPayload.getPackageId();
-        return new SendSignalsPayload(false);
+        Path partPath = writeSignals(personOnProcessing, packageId);
+        return new PartFilePayload(partPath, packageId);
     }
 
-    private void writeSignals(List<Person> persons, Long packageId){
+    private Path writeSignals(List<Person> persons, Long packageId) {
         Path dir = Paths.get(config.getLoadDir(), "tmp_folder");
         try {
             Files.createDirectories(dir);
@@ -70,6 +72,7 @@ public class FidIdEnricher implements Handler<SubjectPayload, SendSignalsPayload
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
+        return done;
 //        try {
 //            mergerQueue.put(new PartFilePayload(done, packageId, null));
 //        } catch (InterruptedException e) {

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

@@ -1,19 +1,18 @@
 package ru.nbch.credit_tracker.service.task.impl.dprctd;
 
+import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
 import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.model.DummyPayload;
-import ru.nbch.credit_tracker.model.FlagsPayload;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.indic.FlagsService2;
-
-import java.util.List;
 import ru.nbch.credit_tracker.service.task.Handler;
 
 @Service
 @Slf4j
-public class FlagsEnricher implements Handler<FlagsPayload, DummyPayload> {
+public class FlagsEnricher implements Handler<SubjectPayload, DummyPayload> {
 
     private final FlagsService2 flagsService;
     private final PhoneFidMapMapper mapper;
@@ -25,21 +24,22 @@ public class FlagsEnricher implements Handler<FlagsPayload, DummyPayload> {
     }
 
     @Override
-    public DummyPayload handle(FlagsPayload taskPayload) {
+    public DummyPayload handle(SubjectPayload taskPayload) {
-        List<PhoneFidMap> phoneFidMaps = taskPayload.getPhoneFidMaps();
+        List<SubjectProfile> phoneFidMaps = taskPayload.getSubjects();
         log.debug("FlagsEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMaps.size());
-//        flagsService.fillFlagsData(phoneFidMaps);
+        flagsService.fillFlagsData(phoneFidMaps);
 
-        log.debug("Start updating subject's batch");
+        //todo по какому ключу апдейтить записи, потому что id у нас тут нет, по subject_id? или все же сохранять id при вставке в SaveSubjectHandler
-        if (!phoneFidMaps.isEmpty()) {
+//        log.debug("Start updating subject's batch");
-            int updateBatchSize = 7000;
+//        if (!phoneFidMaps.isEmpty()) {
-            for (int j = 0; j < phoneFidMaps.size(); j += updateBatchSize) {
+//            int updateBatchSize = 7000;
-                int endIndex = Math.min(j + updateBatchSize, phoneFidMaps.size());
+//            for (int j = 0; j < phoneFidMaps.size(); j += updateBatchSize) {
-                mapper.updatePhoneFidMaps(phoneFidMaps.subList(j, endIndex));
+//                int endIndex = Math.min(j + updateBatchSize, phoneFidMaps.size());
-                log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, phoneFidMaps.size(), taskPayload.getPackageId());
+//                mapper.updatePhoneFidMaps(phoneFidMaps.subList(j, endIndex));
-            }
+//                log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, phoneFidMaps.size(), taskPayload.getPackageId());
-        }
+//            }
-        log.debug("Complete updating subject's batch");
+//        }
+//        log.debug("Complete updating subject's batch");
 
         return new DummyPayload(false);
     }

src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.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 IPipeLineTerminator<Q extends StagePayload> {
+    void startPollingQueue(BlockingQueue<Q> queue);
+}