From d24cd036562e003e06f73d7b6edcfd1df350d1f7 Mon Sep 17 00:00:00 2001
From: ialbert <ialbert@mfd.ru>
Date: Sat, 1 Nov 2025 15:18:58 +0300
Subject: [PATCH] pipeline builder refactor

---
 .../credit_tracker/service/task/PipelinePackageManager.java | 6 ++----
 .../service/task/setup/PipeLineBuilderV2.java               | 6 ++++--
 2 files changed, 6 insertions(+), 6 deletions(-)

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 7eb958b..4d5edf8 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
@@ -100,13 +100,11 @@ public class PipelinePackageManager {
     }
 
     public void buildAndRun2(Long packageId, String pkgPath) throws Exception {
-        BlockingQueue<SubjectPayload> idEnricherQueue = idEnricherQueueFactory.getObject();
-        BlockingQueue<SendSignalsPayload> signalsQueue = sendSignalQueueFactory.getObject();
-        var startPipeLine = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueue)
+        var startPipeLine = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                 .handler(saveSubjectHandler)
                 .poison(SubjectPayload.POISON)
                 .andThen()
-            .stage(idEnricherQueue, signalsQueue)
+            .stage(sendSignalQueueFactory.getObject())
                 .handler(fidIdEnricherHandler)
                 .poison(SendSignalsPayload.POISON)
                 .andThen()
diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java
index 5455790..daec97f 100644
--- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java
+++ b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java
@@ -11,6 +11,7 @@ import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep;
 public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePayload> {
     private final List<Stage<?, ?>> stages;
     private Stage<FIRST, ?> firstStage;
+    private BlockingQueue<LAST> lastQueue;
 
     private PipeLineBuilderV2() {
         this.stages = new ArrayList<>();
@@ -19,6 +20,7 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
     public static <I extends StagePayload, O extends StagePayload>
     HandlerStep<I, I, O> first(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
         PipeLineBuilderV2<I, O> builder = new PipeLineBuilderV2<>();
+        builder.lastQueue = outputQueue;
         HandlerStep<I, I, O> step = StageBuilderV2.create(inputQueue, outputQueue);
         ((StageBuilderV2<I, I, O>) step).setParent(builder);
         return step;
@@ -26,8 +28,8 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
 
     @SuppressWarnings("unchecked")
     public <NEXT extends StagePayload>
-    HandlerStep<FIRST, LAST, NEXT> stage(BlockingQueue<LAST> inputQueue, BlockingQueue<NEXT> outputQueue) {
-        HandlerStep<FIRST, LAST, NEXT> step = StageBuilderV2.create(inputQueue, outputQueue);
+    HandlerStep<FIRST, LAST, NEXT> stage(BlockingQueue<NEXT> outputQueue) {
+        HandlerStep<FIRST, LAST, NEXT> step = StageBuilderV2.create(lastQueue, outputQueue);
         ((StageBuilderV2<FIRST, LAST, NEXT>) step).setParent((PipeLineBuilderV2<FIRST, NEXT>) this);
         return step;
     }