remove useless classes

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

@@ -1,56 +0,0 @@
-package ru.nbch.credit_tracker.config.task;
-
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import ru.nbch.credit_tracker.model.PackageToSavePayload;
-import ru.nbch.credit_tracker.model.PartFilePayload;
-import ru.nbch.credit_tracker.model.PhoneFidPayload;
-import ru.nbch.credit_tracker.model.SubjectsProfilePayload;
-import ru.nbch.credit_tracker.service.task.FidIdEnricher;
-import ru.nbch.credit_tracker.service.task.FidNameEnricher;
-import ru.nbch.credit_tracker.service.task.PackageSaver;
-import ru.nbch.credit_tracker.service.task.TaskConsumer;
-
-@Configuration
-public class ConsumerConfig {
-
-    @Bean
-    public BlockingQueue<PackageToSavePayload> savingRequest() {
-        return new LinkedBlockingQueue<>();
-    }
-    @Bean
-    public TaskConsumer<PackageToSavePayload> savingRequestTask(PackageSaver packageSaver,
-                                                                  BlockingQueue<PackageToSavePayload> savingRequest) {
-        return new TaskConsumer<>(packageSaver, savingRequest, Executors.newSingleThreadScheduledExecutor());
-    }
-
-    @Bean
-    public BlockingQueue<PhoneFidPayload> fidNameEnricherQueue() {
-        return new LinkedBlockingQueue<>();
-    }
-
-    @Bean
-    public TaskConsumer<PhoneFidPayload> fidNameEnricherTask(FidNameEnricher fidNameEnricher,
-                                                             BlockingQueue<PhoneFidPayload> fidNameEnricherQueue) {
-        return new TaskConsumer<>(fidNameEnricher, fidNameEnricherQueue, Executors.newFixedThreadPool(10));
-    }
-
-    @Bean
-    public BlockingQueue<SubjectsProfilePayload> fidIdEnricherQueue() {
-        return new LinkedBlockingQueue<>();
-    }
-
-    @Bean
-    public TaskConsumer<SubjectsProfilePayload> fidIdEnricherTask(FidIdEnricher fidIdEnricher,
-                                                   BlockingQueue<SubjectsProfilePayload> fidIdEnricherQueue) {
-        return new TaskConsumer<>(fidIdEnricher, fidIdEnricherQueue, Executors.newFixedThreadPool(10));
-    }
-
-    @Bean
-    public BlockingQueue<PartFilePayload> mergerQueue() {
-        return new LinkedBlockingQueue<>();
-    }
-}

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

@@ -1,5 +1,8 @@
 package ru.nbch.credit_tracker.facade;
 
+import java.io.File;
+import java.io.IOException;
+import javax.xml.stream.XMLStreamException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.io.FileUtils;
 import org.springframework.stereotype.Service;
@@ -11,18 +14,11 @@ import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.enums.XmlErrorFormat;
 import ru.nbch.credit_tracker.exceptions.SignalClientException;
 import ru.nbch.credit_tracker.model.MonitoringValidationError;
-import ru.nbch.credit_tracker.model.PackageToSavePayload;
 import ru.nbch.credit_tracker.model.ValidationResult;
 import ru.nbch.credit_tracker.service.SignalService;
 import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
 import ru.nbch.credit_tracker.utils.CTUtil;
 import ru.nbch.credit_tracker.utils.ZipUtils;
-
-import javax.xml.stream.XMLStreamException;
-import java.io.File;
-import java.io.IOException;
-import java.util.concurrent.BlockingQueue;
-
 import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp;
 import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp;
 
@@ -34,17 +30,14 @@ public class SignalFacadeV2 {
     private final CreditTrackerSettings config;
     private final SignalService signalService;
     private final AuthorizationService authorizationService;
-    private final BlockingQueue<PackageToSavePayload> savingRequest;
     public SignalFacadeV2(MonitoringRequestValidator monitoringRequestValidator,
                           CreditTrackerSettings config,
                           SignalService signalService,
-                          AuthorizationService authorizationService,
+                          AuthorizationService authorizationService) {
-                          BlockingQueue<PackageToSavePayload> savingRequest) {
         this.monitoringRequestValidator = monitoringRequestValidator;
         this.config = config;
         this.signalService = signalService;
         this.authorizationService = authorizationService;
-        this.savingRequest = savingRequest;
     }
 
     public void setupMonitoringRequest(MultipartFile file) {
@@ -99,7 +92,6 @@ public class SignalFacadeV2 {
             Long packageId = signalService.registerPackage(userId, packageName);
             log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name());
             //create and start pipilene
-            savingRequest.add(new PackageToSavePayload(packageId, filePath));
         } catch (Exception e) {
             CTUtil.removeFile(filePath);
             if (e instanceof SignalClientException) {

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

@@ -1,21 +0,0 @@
-package ru.nbch.credit_tracker.model;
-
-import ru.nbch.credit_tracker.service.task.TaskPayload;
-
-public class PackageToSavePayload implements TaskPayload {
-    private final Long packageId;
-    private final String filePath;
-
-    public PackageToSavePayload(Long packageId, String filePath) {
-        this.packageId = packageId;
-        this.filePath = filePath;
-    }
-
-    public Long getPackageId() {
-        return packageId;
-    }
-
-    public String getFilePath() {
-        return filePath;
-    }
-}

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

@@ -3,7 +3,7 @@ package ru.nbch.credit_tracker.model;
 import ru.nbch.credit_tracker.service.task.TaskPayloadV2;
 
 public class PackageToSavePayloadV2 implements TaskPayloadV2 {
-    private final Integer packageId;
+    private final Long packageId;
     private final String filePath;
     private final boolean poison;
 
@@ -12,13 +12,13 @@ public class PackageToSavePayloadV2 implements TaskPayloadV2 {
         this.filePath = null;
         this.poison = true;
     }
-    public PackageToSavePayloadV2(Integer packageId, String filePath) {
+    public PackageToSavePayloadV2(Long packageId, String filePath) {
         this.packageId = packageId;
         this.filePath = filePath;
         this.poison = false;
     }
 
-    public Integer getPackageId() {
+    public Long getPackageId() {
         return packageId;
     }
 

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

@@ -1,24 +0,0 @@
-package ru.nbch.credit_tracker.model;
-
-import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.service.task.TaskPayload;
-
-import java.util.List;
-
-public class PhoneFidPayload implements TaskPayload {
-    private final Long packageId;
-    private final List<PhoneFidMap> phoneFidsBatch;
-
-    public PhoneFidPayload(Long packageId, List<PhoneFidMap> phoneFidsBatch) {
-        this.packageId = packageId;
-        this.phoneFidsBatch = phoneFidsBatch;
-    }
-
-    public Long getPackageId() {
-        return packageId;
-    }
-
-    public List<PhoneFidMap> getPhoneFidsBatch() {
-        return phoneFidsBatch;
-    }
-}

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

@@ -1,24 +0,0 @@
-package ru.nbch.credit_tracker.model;
-
-import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
-import ru.nbch.credit_tracker.service.task.TaskPayload;
-
-import java.util.List;
-
-public class SubjectsProfilePayload implements TaskPayload {
-    private final Long packageId;
-    private final List<SubjectProfile> subjectProfiles;
-
-    public SubjectsProfilePayload(Long packageId, List<SubjectProfile> subjectProfiles) {
-        this.packageId = packageId;
-        this.subjectProfiles = subjectProfiles;
-    }
-
-    public Long getPackageId() {
-        return packageId;
-    }
-
-    public List<SubjectProfile> getSubjectProfiles() {
-        return subjectProfiles;
-    }
-}

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

@@ -1,85 +0,0 @@
-package ru.nbch.credit_tracker.service.task;
-
-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.dto.SubjectProfile;
-import ru.nbch.credit_tracker.entities.external_request.signals.Person;
-import ru.nbch.credit_tracker.entities.external_request.signals.Signals;
-import ru.nbch.credit_tracker.model.PartFilePayload;
-import ru.nbch.credit_tracker.model.SubjectsProfilePayload;
-import ru.nbch.credit_tracker.service.SignalService;
-import ru.nbch.credit_tracker.service.indic.PersonService;
-import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.file.*;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.BlockingQueue;
-
-@Service
-@Slf4j
-public class FidIdEnricher implements Dispatcher<SubjectsProfilePayload>{
-    private final PersonService personService;
-    private final SignalService signalService;
-    private final CreditTrackerSettings config;
-    private final BlockingQueue<PartFilePayload> mergerQueue;
-    public FidIdEnricher(PersonService personService,
-                         SignalService signalService,
-                         CreditTrackerSettings config,
-                         BlockingQueue<PartFilePayload> mergerQueue) {
-        this.personService = personService;
-        this.signalService = signalService;
-        this.config = config;
-        this.mergerQueue = mergerQueue;
-    }
-
-    @Override
-    public void handle(SubjectsProfilePayload taskPayload) {
-        List<SubjectProfile> subjectsProfile = taskPayload.getSubjectProfiles();
-        log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
-        personService.filledPassportsData(subjectsProfile);
-        List<Person> personOnProcessing = subjectsProfile.stream()
-            .filter(SubjectProfile::isFilled)
-            .map(subjectProfile -> PersonFactoryBuilderV2.builder().personData(subjectProfile).build())
-            .toList();
-        log.debug("personOnProcessing size: {}", personOnProcessing.size());
-        writeSignals(personOnProcessing, taskPayload.getPackageId());
-    }
-
-    private void writeSignals(List<Person> persons, Long packageId){
-        Path dir = Paths.get(config.getLoadDir(), "tmp_folder");
-        try {
-            Files.createDirectories(dir);
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-
-        String base = UUID.randomUUID() + "_output.xml";
-        log.debug("Write tmp xml by packageId: {}", packageId);
-        Path tmp = dir.resolve(base + ".part");
-        Path done = dir.resolve(base);
-
-        Signals signals = signalService.buildSignals(packageId);
-
-        try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW)) {
-            SignalsStaxXmlWriterV2 writer = new SignalsStaxXmlWriterV2(persons, signals, config.getEncoding());
-            writer.writeWithStax(os); // перегрузи на OutputStream, если нужно
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-        try {
-            Files.move(tmp, done, StandardCopyOption.ATOMIC_MOVE);
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-        try {
-            mergerQueue.put(new PartFilePayload(done, packageId, signals));
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }
-    }
-}

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

@@ -1,61 +0,0 @@
-package ru.nbch.credit_tracker.service.task;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.entities.dto.NameProfile;
-import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.entities.dto.SubjectProfile;
-import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
-import ru.nbch.credit_tracker.model.PhoneFidPayload;
-import ru.nbch.credit_tracker.model.SubjectsProfilePayload;
-import ru.nbch.credit_tracker.service.indic.PhoneService2;
-import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
-
-@Service
-@Slf4j
-public class FidNameEnricher implements Dispatcher<PhoneFidPayload> {
-    private final PhoneService2 phoneService2;
-    private final PhoneFidMapService phoneFidMapService;
-    private final PhoneFidMapMapper mapper;
-    private final Queue<SubjectsProfilePayload> fidIdEnricherQueue;
-
-    public FidNameEnricher(PhoneService2 phoneService2,
-                           PhoneFidMapService phoneFidMapService,
-                           PhoneFidMapMapper mapper,
-                           Queue<SubjectsProfilePayload> fidIdEnricherQueue) {
-        this.phoneService2 = phoneService2;
-        this.phoneFidMapService = phoneFidMapService;
-        this.mapper = mapper;
-        this.fidIdEnricherQueue = fidIdEnricherQueue;
-    }
-
-    @Override
-    public void handle(PhoneFidPayload taskPayload) {
-        log.debug("FidNameEnricher handle by packageId: {}", taskPayload.getPackageId());
-        List<PhoneFidMap> phoneFidMap = taskPayload.getPhoneFidsBatch();
-        phoneFidMapService.register(phoneFidMap, taskPayload.getPackageId());
-        Map<Integer, PhoneFidMap> fidsMapByBatch = phoneService2.defineFidsAndNameByPhone(phoneFidMap);
-        log.debug("Found {} fids by batch", fidsMapByBatch.size());
-        //todo по phoneFidMap может быть найден не полный набор фидов, поэтому стоит это контролировать
-        mapper.updateFid(fidsMapByBatch.values());
-
-        List<SubjectProfile> subjectProfiles = fidsMapByBatch.values().stream().map(pf -> {
-            SubjectProfile subjectProfile = new SubjectProfile();
-            subjectProfile.setFid(pf.getFid());
-            subjectProfile.setNameProfile(NameProfile.builder()
-                .firstName(pf.getFirstName())
-                .middleName(pf.getMiddleName())
-                .lastName(pf.getLastName())
-                .birthDate(pf.getBirthDate())
-                .fileSinceDate(pf.getFileSinceDate())
-                .build());
-            return subjectProfile;
-        }).toList();
-
-        SubjectsProfilePayload subjectsProfilePayload = new SubjectsProfilePayload(taskPayload.getPackageId(), subjectProfiles);
-        fidIdEnricherQueue.add(subjectsProfilePayload);
-    }
-}

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

@@ -1,46 +1,36 @@
 package ru.nbch.credit_tracker.service.task;
 
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
-import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
-import ru.nbch.credit_tracker.model.PackageToSavePayload;
-import ru.nbch.credit_tracker.model.PhoneFidPayload;
-import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
-import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
-import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
-
 import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
-import java.util.Queue;
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
+import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
+import ru.nbch.credit_tracker.model.PackageToSavePayloadV2;
+import ru.nbch.credit_tracker.model.PhoneFidPayloadV2;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 
-@Service
 @Slf4j
-public class PackageSaver implements Dispatcher<PackageToSavePayload> {
+public class PipelinePackageProcessor implements Runnable {
     private final StaxXmlProcessor xmlProcessor;
-    private final PhoneFidMapService phoneFidMapService;
-    private final Queue<PhoneFidPayload> fidNameEnricherQueue;
-    private final BlockingQueue<PhoneFidPayloadV2> savingRequest2;
     private final static int BATCH_SIZE = 10_000;
+    private final PackageToSavePayloadV2 taskPayload;
 
-    public PackageSaver(StaxXmlProcessor xmlProcessor,
+    public PipelinePackageProcessor(StaxXmlProcessor xmlProcessor,
-                        PhoneFidMapService phoneFidMapService,
+                                    PackageToSavePayloadV2 taskPayload) {
-                        Queue<PhoneFidPayload> fidNameEnricherQueue,
-                        BlockingQueue<PhoneFidPayloadV2> savingRequest2) {
         this.xmlProcessor = xmlProcessor;
-        this.phoneFidMapService = phoneFidMapService;
+        this.taskPayload = taskPayload;
-        this.fidNameEnricherQueue = fidNameEnricherQueue;
-        this.savingRequest2 = savingRequest2;
     }
     @Override
-    public void handle(PackageToSavePayload taskPayload) {
+    public void run() {
+        BlockingQueue<PhoneFidPayloadV2> savingRequest2 = new LinkedBlockingQueue<>();
         Long packageId = taskPayload.getPackageId();
         Path pkgPath = Paths.get(taskPayload.getFilePath());
         log.info("Start registering subjects. PackageId: {}", taskPayload.getPackageId());

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

@@ -1,59 +0,0 @@
-package ru.nbch.credit_tracker.service.task;
-
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.springframework.context.SmartLifecycle;
-
-@Slf4j
-public class TaskConsumer<X extends TaskPayload> implements SmartLifecycle {
-    private final Dispatcher<X> dispatcher;
-    private final BlockingQueue<X> taskQueue;
-    private final ExecutorService worker;
-    private volatile boolean running;
-
-    public TaskConsumer(Dispatcher<X> dispatcher,
-                        BlockingQueue<X> taskQueue,
-                        ExecutorService worker) {
-        this.dispatcher = dispatcher;
-        this.taskQueue = taskQueue;
-        this.worker = worker;
-    }
-
-    @Override
-    public void start() {
-        running = true;
-        ExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
-        executorService.submit(() -> {
-            while (running && !Thread.currentThread().isInterrupted()) {
-                try {
-                    X job = taskQueue.take();
-                    worker.submit(() -> {
-                        try {
-                            dispatcher.handle(job);
-                        } catch (Exception e) {
-                            log.error("Handle Error: {}", ExceptionUtils.getStackTrace(e));
-                        }
-                    });
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                } catch (Exception e) {
-                    log.error("Error: {}", ExceptionUtils.getStackTrace(e));
-                }
-            }
-        });
-    }
-
-    @Override
-    public void stop() {
-        running = false;
-        worker.shutdownNow();
-    }
-
-    @Override
-    public boolean isRunning() {
-        return running;
-    }
-}

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

@@ -15,12 +15,11 @@ import org.apache.commons.lang3.exception.ExceptionUtils;
 
 @Slf4j
 public class TaskConsumerV2<X extends TaskPayloadV2, Z extends TaskPayloadV2>
-//    implements SmartLifecycle
 {
     private final DispatcherV2<X, Z> dispatcher;
     private final BlockingQueue<X> in;
     private final BlockingQueue<Z> out;
-    private Z POISON;
+    private final Z POISON;
     private final ExecutorService ioPool;
 
     public TaskConsumerV2(DispatcherV2<X, Z> dispatcher,
@@ -34,7 +33,6 @@ public class TaskConsumerV2<X extends TaskPayloadV2, Z extends TaskPayloadV2>
         this.POISON = POISON;
     }
 
-//    @Override
     public void start() {
         List<Future<Z>> inflight = new ArrayList<>();
         AtomicInteger inFlightCount = new AtomicInteger();
@@ -54,7 +52,6 @@ public class TaskConsumerV2<X extends TaskPayloadV2, Z extends TaskPayloadV2>
 
                 Future<Z> f = ioPool.submit(() -> {
                         Z p = dispatcher.handle(b);
-                        //
                         putQuiet(out, p);
                         return p;
                     }

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

@@ -1,4 +0,0 @@
-package ru.nbch.credit_tracker.service.task;
-
-public interface TaskPayload {
-}