treshenkov/credit_tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

pipeline builder refactor

Browse source
This commit is contained in:
ialbert 2025-11-01 15:18:58 +03:00
parent 8e53f24aeb
commit d24cd03656
2 changed files with 6 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java
View file

@ -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()

6
src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java
View file

@ -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;
}