flag calc

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

@@ -5,11 +5,14 @@ import java.util.function.Supplier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.model.impl.PartFilePayload;
 import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
-import ru.nbch.credit_tracker.service.task.XmlMergeCoordinator;
+import ru.nbch.credit_tracker.service.indic.FlagsService2;
+import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
 import ru.nbch.credit_tracker.service.task.impl.KickPipeLineImpl;
+import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@@ -26,4 +29,8 @@ public class PipelineStarterConfig {
         return () -> new XmlMergeCoordinator(creditTrackerSettings);
     }
 
+    @Bean
+    public Supplier<FlagsEnricher> flagEnricherFactory(FlagsService2 flagsService, PhoneFidMapMapper mapper) {
+        return () -> new FlagsEnricher(flagsService, mapper);
+    }
 }

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

@@ -23,4 +23,9 @@ public class StageConfig {
         return new StageQueueFactory<>(10000);
     }
 
+    @Bean("flagQueueFactory")
+    public StageQueueFactory<SubjectPayload> flagQueueFactory() {
+        return new StageQueueFactory<>(15);
+    }
+
 }

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

@@ -47,6 +47,7 @@ public class FlagsService2 {
 
         fidsflagsDataMap.forEach((fid, data) -> {
             subjects.stream().filter(s-> s.getFid() != null && s.getFid().equals(fid)).forEach(phoneFidMap -> {
+
 //                phoneFidMap.setFlagDebt5000(data.getFlagDebt5000());
 //                phoneFidMap.setFlag90days(data.getFlag90days());
 //                phoneFidMap.setCalculatedAt(LocalDateTime.now());

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

@@ -1,10 +1,12 @@
 package ru.nbch.credit_tracker.service.task;
 
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -16,6 +18,7 @@ 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.SaveSubjectHandler;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
@@ -32,9 +35,11 @@ public class PipelinePackageManager {
     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 ExecutorService executorService = Executors.newFixedThreadPool(10, r -> {
         Thread t = new Thread(r);
@@ -51,8 +56,10 @@ public class PipelinePackageManager {
                                   @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) {
         this.applicationContext = applicationContext;
         this.xmlProcessor = xmlProcessor;
@@ -60,9 +67,11 @@ public class PipelinePackageManager {
         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.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS);
     }
@@ -72,9 +81,18 @@ public class PipelinePackageManager {
             throw new RuntimeException("Too many concurrent requests");
         }
         try {
+            BlockingQueue<SubjectPayload> flagQueue = flagQueueFactory.getObject();
+            Consumer<SubjectPayload> consumer = subjectPayload -> {
+                try {
+                    flagQueue.put(subjectPayload);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+            };
             Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                     .handler(saveSubjectHandler)
                     .poison(SubjectPayload.POISON)
+                    .handlerResult(consumer)
                     .andThen()
                 .stage(sendSignalQueueFactory.getObject())
                     .handler(fidIdEnricherHandler)
@@ -84,11 +102,15 @@ public class PipelinePackageManager {
                 .pollFinishingQueue(pipelineTerminatorFactory.get())
                 .build();
 
+            FlagsEnricher flagEnricher = flagEnricherFactory.get();
+            flagEnricher.setPollingQueue(flagQueue);
+
+            executorService.submit(flagEnricher);
             executorService.submit(pipeline.getStarter());
             executorService.submit(pipeline.getTerminator());
 
             pipeline.getTerminator()
-                .getDone()
+                .doneFuture()
                 .orTimeout(5, TimeUnit.MINUTES)
                 .whenComplete((v, e) -> {
                     if (e != null) {
@@ -106,6 +128,14 @@ public class PipelinePackageManager {
                     semaphore.release();
             });
 
+            flagEnricher.doneFuture()
+                .orTimeout(10, TimeUnit.MINUTES)
+                .whenComplete((v, e) -> {
+                    if (e != null) {
+                        log.error(ExceptionUtils.getStackTrace(e));
+                    }
+                    log.debug("Calculation flags completed");
+                });
         } catch (Exception e) {
             semaphore.release();
             throw e;

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

@@ -10,6 +10,7 @@ import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import ru.nbch.credit_tracker.model.StagePayload;
@@ -22,6 +23,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     private final Z POISON;
     private final ExecutorService ioPool;
     private final CompletableFuture<Void> done = new CompletableFuture<>();
+    private Consumer<Z> consumer = null;
 
     public Stage(Handler<X, Z> dispatcher,
                  BlockingQueue<X> in,
@@ -122,6 +124,9 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     void putQuiet(BlockingQueue<Z> q, Z b) {
         try {
             q.put(b);
+            if (consumer != null) {
+                consumer.accept(b);
+            }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
         }
@@ -139,4 +144,12 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     public BlockingQueue<X> getIn() {
         return in;
     }
+
+    public Consumer<Z> getConsumer() {
+        return consumer;
+    }
+
+    public void setConsumer(Consumer<Z> consumer) {
+        this.consumer = consumer;
+    }
 }

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

@@ -0,0 +1,96 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+import ru.nbch.credit_tracker.service.indic.FlagsService2;
+
+@Slf4j
+public class FlagsEnricher implements Runnable {
+
+    private final FlagsService2 flagsService;
+    private final PhoneFidMapMapper mapper;
+    private BlockingQueue<SubjectPayload> in;
+    private final CompletableFuture<Void> done = new CompletableFuture<>();
+    private final ExecutorService ioPool;
+
+    public FlagsEnricher(FlagsService2 flagsService,
+                         PhoneFidMapMapper mapper) {
+        this.flagsService = flagsService;
+        this.mapper = mapper;
+        this.ioPool = new ThreadPoolExecutor(10,
+            10,
+            0L, TimeUnit.MINUTES,
+            new LinkedBlockingQueue<>(),
+            r -> {
+                Thread t = new Thread(r);
+                t.setName("queue-handler-%s-%d".formatted(this.getClass().getSimpleName(), t.threadId()));
+                return t;
+            }
+        );
+    }
+
+    @Override
+    public void run() {
+        List<Future<?>> inFlight = new ArrayList<>();
+        while (!Thread.currentThread().isInterrupted()) {
+            try {
+                SubjectPayload subjectPayload = in.poll(5, TimeUnit.SECONDS);
+                if (subjectPayload == null) {
+                    continue;
+                }
+                if (subjectPayload.isPoison()) {
+                    log.debug("Got poison");
+                    for (Future<?> f : inFlight) {
+                        f.get();
+                    }
+                    done.complete(null);
+                    return;
+                }
+                Future<?> future = ioPool.submit(() -> process(subjectPayload));
+                inFlight.add(future);
+            } catch (InterruptedException ie) {
+                Thread.currentThread().interrupt();
+            } catch (Exception e) {
+                log.error("Error: {}", ExceptionUtils.getStackTrace(e));
+            }
+        }
+    }
+
+    public void process(SubjectPayload subjectPayload) {
+        List<SubjectProfile> phoneFidMaps = subjectPayload.getSubjects();
+        log.debug("FlagsEnricher handle by packageId: {}, size: {}", subjectPayload.getPackageId(), phoneFidMaps.size());
+        flagsService.fillFlagsData(phoneFidMaps);
+
+        //todo по какому ключу апдейтить записи, потому что id у нас тут нет, по subject_id? или все же сохранять id при вставке в SaveSubjectHandler
+//        log.debug("Start updating subject's batch");
+//        if (!phoneFidMaps.isEmpty()) {
+//            int updateBatchSize = 7000;
+//            for (int j = 0; j < phoneFidMaps.size(); j += updateBatchSize) {
+//                int endIndex = Math.min(j + updateBatchSize, phoneFidMaps.size());
+//                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");
+    }
+
+    public void setPollingQueue(BlockingQueue<SubjectPayload> queue) {
+        this.in = queue;
+    }
+
+    public CompletableFuture<Void> doneFuture() {
+        return done;
+    }
+}

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

@@ -2,15 +2,16 @@ package ru.nbch.credit_tracker.service.task.impl;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.annotation.Scope;
 import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.indic.Name;
+import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.model.NameData;
 import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.indic.PhoneService2;
-import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
 import ru.nbch.credit_tracker.service.task.Handler;
 import ru.nbch.credit_tracker.utils.CTUtil;
 
@@ -19,17 +20,16 @@ import ru.nbch.credit_tracker.utils.CTUtil;
  * Так же выполняется поиск и обогащение перс.данными из NAME для дальнейшей обработки.
  */
 @Service
-@Scope("prototype")
 @Slf4j
 //todo опять нейминг класса, варианты: ResolveSubject; DefineIcrsSubject; IcrsMapping/Resolving; PackageSaver; Filler...
 public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayload> {
     private final PhoneService2 phoneService2;
-    private final PhoneFidMapService phoneFidMapService;
+    private final PhoneFidMapMapper phoneFidMapMapper;
 
     public SaveSubjectHandler(PhoneService2 phoneService2,
-                              PhoneFidMapService phoneFidMapService) {
+                              PhoneFidMapMapper phoneFidMapMapper) {
         this.phoneService2 = phoneService2;
-        this.phoneFidMapService = phoneFidMapService;
+        this.phoneFidMapMapper = phoneFidMapMapper;
     }
 
     @Override
@@ -39,13 +39,12 @@ public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayloa
         log.debug("SaveSubjectStage handle by packageId: {}, size of bucket for save: {}", taskPayload.getPackageId(), subjects.size());
 
         Map<String, Name> nameByPhone = phoneService2.findNameByNumber(subjects);
-        phoneFidMapService.register(subjects, nameByPhone, packageId);
-
-        //fill name info
         //todo сделать один проход для установки FID для вставки в PHONE_FID_MAP и одновременного заполнения инфо из Name для дальнейшей обработки
-        subjects.stream()
+        List<PhoneFidMap> phoneFidMaps = subjects.stream()
             .filter(subject -> nameByPhone.containsKey(CTUtil.normalizePhoneNumber(subject.getPhone())))
-            .forEach(subject -> {
+            .map(subject -> {
+                //заполним инфо из NAME для след. обработчика
+                {
                     Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone()));
                     NameData nameData = NameData.builder()
                         .middleName(name.getMiddle())
@@ -57,8 +56,21 @@ public class SaveSubjectHandler implements Handler<SubjectPayload, SubjectPayloa
 
                     subject.setFid(name.getFid());
                     subject.setNameData(nameData);
-            });
+                }
 
+                Integer fid = Optional
+                    .ofNullable(nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone())))
+                    .map(Name::getFid)
+                    .orElse(null);
+
+                return PhoneFidMap.builder()
+                    .phone(subject.getPhone())
+                    .subjectId(subject.getIdentifier())
+                    .packageId(packageId)
+                    .fid(fid)
+                    .build();
+            }).toList();
+        phoneFidMapMapper.register(phoneFidMaps);
         //todo стоит ли делать имутабл для subject?
         return new SubjectPayload(taskPayload.getPackageId(), subjects);
     }

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

@@ -1,4 +1,4 @@
-package ru.nbch.credit_tracker.service.task;
+package ru.nbch.credit_tracker.service.task.impl;
 
 import java.io.File;
 import java.io.IOException;
@@ -27,7 +27,6 @@ import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
 
 @Slf4j
-//@Service
 public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload> {
     private final Path mergedDir;
     private final Charset encoding = Charset.forName("windows-1251");
@@ -47,7 +46,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
     }
 
     @Override
-    public CompletableFuture<SignalXmlFilePayload> getDone() {
+    public CompletableFuture<SignalXmlFilePayload> doneFuture() {
         return done;
     }
 

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

@@ -1,46 +0,0 @@
-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.mapper.signal.PhoneFidMapMapper;
-import ru.nbch.credit_tracker.model.DummyPayload;
-import ru.nbch.credit_tracker.model.SubjectProfile;
-import ru.nbch.credit_tracker.model.impl.SubjectPayload;
-import ru.nbch.credit_tracker.service.indic.FlagsService2;
-import ru.nbch.credit_tracker.service.task.Handler;
-
-@Service
-@Slf4j
-public class FlagsEnricher implements Handler<SubjectPayload, DummyPayload> {
-
-    private final FlagsService2 flagsService;
-    private final PhoneFidMapMapper mapper;
-
-    public FlagsEnricher(FlagsService2 flagsService,
-                         PhoneFidMapMapper mapper) {
-        this.flagsService = flagsService;
-        this.mapper = mapper;
-    }
-
-    @Override
-    public DummyPayload handle(SubjectPayload taskPayload) {
-        List<SubjectProfile> phoneFidMaps = taskPayload.getSubjects();
-        log.debug("FlagsEnricher handle by packageId: {}, size: {}", taskPayload.getPackageId(), phoneFidMaps.size());
-        flagsService.fillFlagsData(phoneFidMaps);
-
-        //todo по какому ключу апдейтить записи, потому что id у нас тут нет, по subject_id? или все же сохранять id при вставке в SaveSubjectHandler
-//        log.debug("Start updating subject's batch");
-//        if (!phoneFidMaps.isEmpty()) {
-//            int updateBatchSize = 7000;
-//            for (int j = 0; j < phoneFidMaps.size(); j += updateBatchSize) {
-//                int endIndex = Math.min(j + updateBatchSize, phoneFidMaps.size());
-//                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");
-
-        return new DummyPayload(false);
-    }
-}

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

@@ -7,5 +7,5 @@ import ru.nbch.credit_tracker.model.TerminatePayload;
 
 public interface IPipeLineTerminator<Q extends StagePayload, X extends TerminatePayload> extends Runnable {
     void setPollingQueue(BlockingQueue<Q> queue);
-    CompletableFuture<X> getDone();
+    CompletableFuture<X> doneFuture();
 }

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

@@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.service.task.setup;
 
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
+import java.util.function.Consumer;
 import ru.nbch.credit_tracker.model.StagePayload;
 import ru.nbch.credit_tracker.service.task.Handler;
 import ru.nbch.credit_tracker.service.task.Stage;
@@ -21,6 +22,7 @@ public class StageBuilderV2<FIRST extends StagePayload, I extends StagePayload,
     private final BlockingQueue<O> outputQueue;
     private Handler<I, O> handler;
     private O poison;
+    Consumer<O> consumer = null;
 
     private StageBuilderV2(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
         this.inputQueue = Objects.requireNonNull(inputQueue, "inputQueue is required");
@@ -53,6 +55,11 @@ public class StageBuilderV2<FIRST extends StagePayload, I extends StagePayload,
         return this;
     }
 
+    public BuildStep<FIRST, O> handlerResult(Consumer<O> consumer) {
+        this.consumer = Objects.requireNonNull(consumer, "consumer is required");
+        return this;
+    }
+
     /** Шаг 3: собрать Stage и вернуться к PipeLineBuilder */
     @Override
     public PipeLineBuilderV2<FIRST, O> andThen() {
@@ -67,6 +74,7 @@ public class StageBuilderV2<FIRST extends StagePayload, I extends StagePayload,
         }
 
         Stage<I, O> stage = new Stage<>(handler, inputQueue, outputQueue, poison);
+        stage.setConsumer(consumer);
         parent.addStage(stage);
         return parent;
     }

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

@@ -1,8 +1,10 @@
 package ru.nbch.credit_tracker.service.task.setup.step;
 
+import java.util.function.Consumer;
 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();
+    BuildStep<FIRST, O> handlerResult(Consumer<O> consumer);
 }