From 00662f9685ec15572b565839e3620a91e53b20ee Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Mon, 10 Nov 2025 19:17:29 +0300 Subject: [PATCH] flag calc --- .../config/task/PipelineStarterConfig.java | 9 +- .../config/task/StageConfig.java | 5 + .../service/indic/FlagsService2.java | 1 + .../service/task/PipelinePackageManager.java | 32 ++++++- .../credit_tracker/service/task/Stage.java | 13 +++ .../service/task/impl/FlagsEnricher.java | 96 +++++++++++++++++++ .../service/task/impl/SaveSubjectHandler.java | 58 ++++++----- .../task/{ => impl}/XmlMergeCoordinator.java | 5 +- .../task/impl/dprctd/FlagsEnricher.java | 46 --------- .../task/setup/IPipeLineTerminator.java | 2 +- .../service/task/setup/StageBuilderV2.java | 8 ++ .../service/task/setup/step/BuildStep.java | 2 + 12 files changed, 202 insertions(+), 75 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java rename src/main/java/ru/nbch/credit_tracker/service/task/{ => impl}/XmlMergeCoordinator.java (98%) delete mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java index 1077171..ab39072 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java +++ b/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 flagEnricherFactory(FlagsService2 flagsService, PhoneFidMapMapper mapper) { + return () -> new FlagsEnricher(flagsService, mapper); + } } 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 0c5e988..f87c3d6 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,4 +23,9 @@ public class StageConfig { return new StageQueueFactory<>(10000); } + @Bean("flagQueueFactory") + public StageQueueFactory flagQueueFactory() { + return new StageQueueFactory<>(15); + } + } diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java index 5997ed3..972b5a8 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService2.java +++ b/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()); 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 4bef0fc..55d7013 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 @@ -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 saveSubjectQueueFactory; private final StageQueueFactory idEnricherQueueFactory; private final StageQueueFactory sendSignalQueueFactory; + private final StageQueueFactory flagQueueFactory; private final SaveSubjectHandler saveSubjectHandler; private final FidIdEnricher fidIdEnricherHandler; private final Supplier> pipelineTerminatorFactory; + private final Supplier flagEnricherFactory; private final BiFunction> pipelineStarterFactory; private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> { Thread t = new Thread(r); @@ -51,8 +56,10 @@ public class PipelinePackageManager { @Qualifier("saveSubjectQueueFactory") StageQueueFactory saveSubjectQueueFactory, @Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, + @Qualifier("flagQueueFactory") StageQueueFactory flagQueueFactory, SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, Supplier> pipelineTerminatorFactory, + Supplier flagEnricherFactory, BiFunction> 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 flagQueue = flagQueueFactory.getObject(); + Consumer 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; 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 4c2bbf3..bf8d372 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 @@ -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 implements Ru private final Z POISON; private final ExecutorService ioPool; private final CompletableFuture done = new CompletableFuture<>(); + private Consumer consumer = null; public Stage(Handler dispatcher, BlockingQueue in, @@ -122,6 +124,9 @@ public class Stage implements Ru void putQuiet(BlockingQueue 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 implements Ru public BlockingQueue getIn() { return in; } + + public Consumer getConsumer() { + return consumer; + } + + public void setConsumer(Consumer consumer) { + this.consumer = consumer; + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java new file mode 100644 index 0000000..dd95a58 --- /dev/null +++ b/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 in; + private final CompletableFuture 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> 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 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 queue) { + this.in = queue; + } + + public CompletableFuture doneFuture() { + return done; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java index 84155e9..f1d38af 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/SaveSubjectHandler.java +++ b/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 { 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,26 +39,38 @@ public class SaveSubjectHandler implements Handler nameByPhone = phoneService2.findNameByNumber(subjects); - phoneFidMapService.register(subjects, nameByPhone, packageId); - - //fill name info //todo сделать один проход для установки FID для вставки в PHONE_FID_MAP и одновременного заполнения инфо из Name для дальнейшей обработки - subjects.stream() + List phoneFidMaps = subjects.stream() .filter(subject -> nameByPhone.containsKey(CTUtil.normalizePhoneNumber(subject.getPhone()))) - .forEach(subject -> { - Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone())); - NameData nameData = NameData.builder() - .middleName(name.getMiddle()) - .firstName(name.getFirst()) - .lastName(name.getLast()) - .fileSinceDate(name.getFileSinceDate()) - .birthDate(name.getBirthDate()) + .map(subject -> { + //заполним инфо из NAME для след. обработчика + { + Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subject.getPhone())); + NameData nameData = NameData.builder() + .middleName(name.getMiddle()) + .firstName(name.getFirst()) + .lastName(name.getLast()) + .fileSinceDate(name.getFileSinceDate()) + .birthDate(name.getBirthDate()) + .build(); + + 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(); - - subject.setFid(name.getFid()); - subject.setNameData(nameData); - }); - + }).toList(); + phoneFidMapMapper.register(phoneFidMaps); //todo стоит ли делать имутабл для subject? return new SubjectPayload(taskPayload.getPackageId(), subjects); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java similarity index 98% rename from src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java rename to src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java index f08508c..f035f0e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/XmlMergeCoordinator.java +++ b/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 { private final Path mergedDir; private final Charset encoding = Charset.forName("windows-1251"); @@ -47,7 +46,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator getDone() { + public CompletableFuture doneFuture() { return done; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java deleted file mode 100644 index 5cd0197..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/dprctd/FlagsEnricher.java +++ /dev/null @@ -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 { - - 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 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); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java index 803f7d0..82a8519 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java +++ b/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 extends Runnable { void setPollingQueue(BlockingQueue queue); - CompletableFuture getDone(); + CompletableFuture doneFuture(); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java index a20a858..f8483aa 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilderV2.java +++ b/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 outputQueue; private Handler handler; private O poison; + Consumer consumer = null; private StageBuilderV2(BlockingQueue inputQueue, BlockingQueue outputQueue) { this.inputQueue = Objects.requireNonNull(inputQueue, "inputQueue is required"); @@ -53,6 +55,11 @@ public class StageBuilderV2 handlerResult(Consumer consumer) { + this.consumer = Objects.requireNonNull(consumer, "consumer is required"); + return this; + } + /** Шаг 3: собрать Stage и вернуться к PipeLineBuilder */ @Override public PipeLineBuilderV2 andThen() { @@ -67,6 +74,7 @@ public class StageBuilderV2 stage = new Stage<>(handler, inputQueue, outputQueue, poison); + stage.setConsumer(consumer); parent.addStage(stage); return parent; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/BuildStep.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/BuildStep.java index b3c7ee3..e3b7e4e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/BuildStep.java +++ b/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 { PipeLineBuilderV2 andThen(); + BuildStep handlerResult(Consumer consumer); }