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

Use pipline engine as library

Browse source
This commit is contained in:
Mikhail Trofimov 2026-04-23 17:28:27 +03:00
parent 4cc3f6eb3f
commit 45e6447d2e
40 changed files with 114 additions and 874 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
pom.xml
View file

@ -38,6 +38,11 @@
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>3.0.4</version>
</dependency>
<dependency>
<groupId>ru.nbch</groupId>
<artifactId>pipeline-engine</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>

31
src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java
View file

@ -1,26 +1,20 @@
package ru.nbch.credit_tracker.config.task;
import java.util.function.Function;
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.PhonesMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload;
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.model.impl.VoidPayload;
import ru.nbch.credit_tracker.model.impl.*;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.KickPipeLineDefaultImpl;
import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRepeatedRestoreImpl;
import ru.nbch.credit_tracker.service.task.impl.KickRecalcPipeLine;
import ru.nbch.credit_tracker.service.task.impl.NoOpTerminator;
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.task.impl.*;
import ru.nbch.pipeline.engine.core.PipelineBackoff;
import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator;
import java.util.function.Function;
import java.util.function.Supplier;
@Configuration
public class PipelineStarterConfig {
@ -52,4 +46,13 @@ public class PipelineStarterConfig {
public Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory() {
return NoOpTerminator::new;
}
@Bean
public PipelineBackoff pipelineBackoff(CreditTrackerSettings creditTrackerSettings) {
return new PipelineBackoff(
creditTrackerSettings.getPipeline().errorAttemptCount(),
creditTrackerSettings.getPipeline().errorAttemptDelay(),
creditTrackerSettings.getPipeline().retryableExceptions()
);
}
}

5
src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java
View file

@ -1,9 +1,10 @@
package ru.nbch.credit_tracker.config.task;
import org.springframework.beans.factory.FactoryBean;
import ru.nbch.pipeline.engine.core.StagePayload;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.springframework.beans.factory.FactoryBean;
import ru.nbch.credit_tracker.model.StagePayload;
public class StageQueueFactory<T extends StagePayload> implements FactoryBean<BlockingQueue<T>> {

2
src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java
View file

@ -30,7 +30,6 @@ import ru.nbch.credit_tracker.model.ValidationResult;
import ru.nbch.credit_tracker.service.UserPackageService;
import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
@ -40,6 +39,7 @@ import ru.nbch.credit_tracker.utils.CtFileUtils;
import ru.nbch.credit_tracker.utils.EnumUtil;
import ru.nbch.credit_tracker.utils.LogUtil;
import ru.nbch.credit_tracker.utils.ZipUtils;
import ru.nbch.pipeline.engine.core.Pipeline;
import javax.xml.stream.XMLStreamException;
import java.io.File;

21
src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java
View file

@ -1,21 +0,0 @@
package ru.nbch.credit_tracker.model;
import java.util.EnumMap;
public class PipelineContext<T extends Enum<T>> {
private final EnumMap<T, Object> storage;
public PipelineContext(Class<T> keyClass) {
this.storage = new EnumMap<>(keyClass);
}
public void store(T key, Object value) {
storage.put(key, value);
}
@SuppressWarnings("unchecked")
public <S> S get(T key) {
Object o = storage.get(key);
return (S) o;
}
}

7
src/main/java/ru/nbch/credit_tracker/model/StagePayload.java
View file

@ -1,7 +0,0 @@
package ru.nbch.credit_tracker.model;
public interface StagePayload {
boolean isPoison();
default void setContext(PipelineContext<?> context) {}
default PipelineContext<?> getContext() {return null;}
}

4
src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java
View file

@ -1,4 +0,0 @@
package ru.nbch.credit_tracker.model;
public interface TerminatePayload {
}

2
src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java
View file

@ -1,6 +1,6 @@
package ru.nbch.credit_tracker.model.impl;
import ru.nbch.credit_tracker.model.TerminatePayload;
import ru.nbch.pipeline.engine.core.TerminatePayload;
public class EmptyTerminatePayload implements TerminatePayload {
}

5
src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java
View file

@ -1,9 +1,10 @@
package ru.nbch.credit_tracker.model.impl;
import java.nio.file.Path;
import lombok.Getter;
import ru.nbch.credit_tracker.enums.PoisonType;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.pipeline.engine.core.StagePayload;
import java.nio.file.Path;
@Getter
public class PartFilePayload implements StagePayload {

3
src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java
View file

@ -1,7 +1,8 @@
package ru.nbch.credit_tracker.model.impl;
import ru.nbch.pipeline.engine.core.TerminatePayload;
import java.nio.file.Path;
import ru.nbch.credit_tracker.model.TerminatePayload;
public record SignalXmlFilePayload(Path signalXmlFilePath) implements TerminatePayload {
}

7
src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java
View file

@ -1,10 +1,11 @@
package ru.nbch.credit_tracker.model.impl;
import java.util.List;
import ru.nbch.credit_tracker.enums.PoisonType;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.pipeline.engine.core.PipelineContext;
import ru.nbch.pipeline.engine.core.StagePayload;
import java.util.List;
public class SubjectPayload implements StagePayload {

2
src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java
View file

@ -2,7 +2,7 @@ package ru.nbch.credit_tracker.model.impl;
import lombok.Getter;
import ru.nbch.credit_tracker.enums.PoisonType;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.pipeline.engine.core.StagePayload;
@Getter
public class VoidPayload implements StagePayload {

7
src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java
View file

@ -1,7 +1,5 @@
package ru.nbch.credit_tracker.service.mode.common;
import java.nio.file.Path;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.stereotype.Service;
@ -11,7 +9,6 @@ import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
@ -19,6 +16,10 @@ import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.credit_tracker.service.task.impl.Saver;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import ru.nbch.pipeline.engine.core.Pipeline;
import java.nio.file.Path;
import java.util.List;
/**
* Сервис для завершения обработки пакетов находящихся в статусе PROCESSING

7
src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java
View file

@ -1,7 +1,5 @@
package ru.nbch.credit_tracker.service.mode.common;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
@ -10,12 +8,15 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.signal.PhonesMapService;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.pipeline.engine.core.Pipeline;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@Slf4j
@Service

5
src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java
View file

@ -1,16 +1,17 @@
package ru.nbch.credit_tracker.service.mode.migration;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelineMigrationPackageFabric;
import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.pipeline.engine.core.Pipeline;
import java.util.concurrent.CompletableFuture;
@Slf4j
@Service

24
src/main/java/ru/nbch/credit_tracker/service/task/Handler.java
View file

@ -1,24 +0,0 @@
package ru.nbch.credit_tracker.service.task;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.refactoring.Suspicios;
public interface Handler<X extends StagePayload, Z extends StagePayload> {
Z handle(X taskPayload);
@SuppressWarnings("unchecked")
default <T extends Enum<T>> PipelineContext<T> obtainContext(Class<T> keyClass, X input) {
PipelineContext<?> ctx = input.getContext();
return (PipelineContext<T>) ctx;
}
@Suspicios("Удалить?")
default void handleOnPoison(X poison){
//do nothing
}
default int preferredConcurrency() {return 10;}
Z cookPoison();
}

81
src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java
View file

@ -1,81 +0,0 @@
package ru.nbch.credit_tracker.service.task;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
import ru.nbch.credit_tracker.utils.LogUtil;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
public class Pipeline {
private final Logger log = LoggerFactory.getLogger(getClass());
private Runnable starter;
private IPipeLineTerminator<?,?> terminator;
private List<Stage<?, ?>> stages;
private Future<?> starterFuture;
private Future<?> terminatorFuture;
private final AtomicBoolean closed = new AtomicBoolean(false);
private PipelineContext<?> context;
private volatile long startTime;
public void start(ExecutorService executorService) {
this.startTime = System.currentTimeMillis();
this.starterFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), starter));
this.terminatorFuture = executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), terminator));
}
public void close() {
if (closed.compareAndSet(false, true)) {
log.info("closing the pipeline...");
stages.forEach(Stage::close);
this.starterFuture.cancel(true);
this.terminatorFuture.cancel(true);
} else {
log.info("pipeline already closed.");
}
}
public Stage<?, ?> getLast() {
return stages.getLast();
}
public Runnable getStarter() {
return starter;
}
public void setStarter(Runnable starter) {
this.starter = starter;
}
public IPipeLineTerminator<?, ?> getTerminator() {
return terminator;
}
public void setTerminator(IPipeLineTerminator<?, ?> terminator) {
this.terminator = terminator;
}
public List<Stage<?, ?>> getStages() {
return stages;
}
public void setStages(List<Stage<?, ?>> stages) {
this.stages = stages;
}
public long getStartTime() {
return startTime;
}
public void setContext(PipelineContext<?> context) {
this.context = context;
}
public <T extends Enum<T>> PipelineContext<T> obtainContext(Class<T> keyClass) {
return (PipelineContext<T>) context;
}
}

33
src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java
View file

@ -1,33 +0,0 @@
package ru.nbch.credit_tracker.service.task;
import java.time.Duration;
import java.util.List;
import lombok.Getter;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@Getter
@Service
public class PipelineBackoff {
private final Integer retryCount;
private final Duration delay;
private final List<Class<? extends Throwable>> retryableExceptions;
public PipelineBackoff(CreditTrackerSettings creditTrackerSettings) {
this.retryCount = creditTrackerSettings.getPipeline().errorAttemptCount();
this.delay = creditTrackerSettings.getPipeline().errorAttemptDelay();
this.retryableExceptions = creditTrackerSettings.getPipeline().retryableExceptions();
}
public void backoff() {
try {
Thread.sleep(delay.toMillis());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public boolean isRetryable(final Throwable throwable) {
return retryableExceptions.stream().anyMatch(aClass -> aClass.isAssignableFrom(throwable.getClass()));
}
}

10
src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java
View file

@ -12,9 +12,11 @@ import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.model.impl.VoidPayload;
import ru.nbch.credit_tracker.service.task.impl.FidIdUpdater;
import ru.nbch.credit_tracker.service.task.impl.FidNameUpdaterMigration;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilder;
import ru.nbch.pipeline.engine.core.Pipeline;
import ru.nbch.pipeline.engine.core.PipelineBackoff;
import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator;
import ru.nbch.pipeline.engine.core.setup.PipeLineBuilderV2;
import java.util.function.Function;
import java.util.function.Supplier;
@ -53,7 +55,7 @@ public class PipelineMigrationPackageFabric implements InitializingBean {
}
public Pipeline buildRecalculationPDMigration(SetupParam setupParam) throws Exception {
return PipeLineBuilder.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject())
return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject())
.handler(fidNameUpdaterMigration)
.andThen()
.stage(noOpQueueFactory.getObject())

14
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java
View file

@ -8,9 +8,11 @@ import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.config.task.StageQueueFactory;
import ru.nbch.credit_tracker.model.impl.*;
import ru.nbch.credit_tracker.service.task.impl.*;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilder;
import ru.nbch.pipeline.engine.core.Pipeline;
import ru.nbch.pipeline.engine.core.PipelineBackoff;
import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator;
import ru.nbch.pipeline.engine.core.setup.PipeLineBuilderV2;
import java.util.function.Function;
import java.util.function.Supplier;
@ -70,7 +72,7 @@ public class PipelinePackageFabric implements InitializingBean {
}
public Pipeline build(SetupParam setupParam) throws Exception {
return PipeLineBuilder.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
.handler(fidNameEnricher)
.andThen()
.stage(sendSignalQueueFactory.getObject())
@ -84,7 +86,7 @@ public class PipelinePackageFabric implements InitializingBean {
}
public Pipeline buildRecalculationPD(SetupParam setupParam) throws Exception {
return PipeLineBuilder.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject())
return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject())
.handler(fidNameUpdater)
.andThen()
.stage(noOpQueueFactory.getObject())
@ -98,7 +100,7 @@ public class PipelinePackageFabric implements InitializingBean {
}
public Pipeline buildRecalculationPDMigration(SetupParam setupParam) throws Exception {
return PipeLineBuilder.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject())
return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject())
.handler(fidNameUpdaterMigration)
.andThen()
.stage(noOpQueueFactory.getObject())

14
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java
View file

@ -1,11 +1,5 @@
package ru.nbch.credit_tracker.service.task;
import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@ -19,6 +13,14 @@ import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import ru.nbch.pipeline.engine.core.Pipeline;
import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@Service
public class PipelinePackageOrchestrator {

10
src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java
View file

@ -1,14 +1,16 @@
package ru.nbch.credit_tracker.service.task;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import ru.nbch.pipeline.engine.core.Pipeline;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@Service
public class PipelineRecalcPDOrchestrator {

227
src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
View file

@ -1,227 +0,0 @@
package ru.nbch.credit_tracker.service.task;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.utils.LogUtil;
import ru.nbch.credit_tracker.utils.TaskTimer;
@Slf4j
public class Stage<X extends StagePayload, Z extends StagePayload> implements Runnable {
private final Handler<X, Z> dispatcher;
private final BlockingQueue<X> in;
private final BlockingQueue<Z> out;
private final ExecutorService ioPool;
private Consumer<Z> consumer = null;
private final String dispatcherName;
private Pipeline pipeline;
private PipelineBackoff pipelineBackoff;
private final TaskTimer stopWatch = new TaskTimer();
private static final ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy
= new ThreadPoolExecutor.CallerRunsPolicy();
private PipelineContext<?> context;
public Stage(Handler<X, Z> dispatcher,
BlockingQueue<X> in,
BlockingQueue<Z> out) {
if (dispatcher.preferredConcurrency() < 2) {
throw new IllegalStateException("preferred concurrency must be at least 2");
}
this.dispatcher = dispatcher;
this.in = in;
this.out = out;
this.dispatcherName = dispatcher.getClass().getSimpleName();
var mdcMap = LogUtil.extractMdcMap();
BlockingQueue<Runnable> handoffQueue = new SynchronousQueue<>();
this.ioPool = new ThreadPoolExecutor(dispatcher.preferredConcurrency(),
dispatcher.preferredConcurrency(),
0L, TimeUnit.MINUTES,
handoffQueue,
r -> {
Thread t = new Thread(LogUtil.wrap(mdcMap, r));
t.setName("pipe-handler-%s-%d".formatted(dispatcherName, t.threadId()));
return t;
},
(r, executor) -> {
log.info("stage {}: all threads are working; switch to CallerRunsPolicy", dispatcherName);
try {
//блокировать отправителя, пока какой‑то рабочий поток не возьмёт задачу
handoffQueue.put(r);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// callerRunsPolicy.rejectedExecution(r, executor);
}
);
}
@Override
public void run() {
List<Future<Void>> inflight = new ArrayList<>();
log.info("this is main thread for {}", dispatcherName);
try {
int counter = 0;
while (true) {
X b = takeQuiet();
counter++;
log.trace("Take item from queue counter: {}", counter);
if (wasInterrupted()) {
break;
}
if (b.isPoison()) {
log.info("Got poison");
// дождаться всех in-flight задач
for (Future<Void> f : inflight) {
if (wasInterrupted()) {
log.info("stage {} was interrupted while waiting for tasks to finish", dispatcherName);
break;
}
f.get();
}
dispatcher.handleOnPoison(b);
putQuiet(out, dispatcher.cookPoison());
log.info("stage {} finished.", dispatcherName);
return;
}
int finalCounter = counter;
AtomicInteger retryForSingleTask = new AtomicInteger(0);
log.trace("submit task");
Future<Void> f = ioPool.submit(() -> {
while (retryForSingleTask.incrementAndGet() < pipelineBackoff.getRetryCount()) {
try {
if (Thread.currentThread().isInterrupted()) {
log.info("stage {} was interrupted. skipping task {}...", dispatcherName, finalCounter);
return null;
}
log.info("stage {} task {} started.", dispatcher.getClass().getSimpleName(), finalCounter);
Z p;
try (var m = stopWatch.start(dispatcherName + "-handler")) {
if (b.getContext() == null) {
b.setContext(context);
}
p = dispatcher.handle(b);
p.setContext(b.getContext()); // нужно т.к. в конце придет в терминатор
}
if (Thread.currentThread().isInterrupted()) {
log.info("after {} task {} finished, worker has been interrupted. not putting result in queue.",
dispatcherName, finalCounter);
return null;
}
putQuiet(out, p);
log.info("stage {} task {} ended.", dispatcher.getClass().getSimpleName(), finalCounter);
return null;
} catch (Exception e) {
boolean isRetryable = pipelineBackoff.isRetryable(e);
if (!isRetryable) {
log.error("{} ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
pipeline.close();
return null;
}
log.warn("{} RETRY AFTER ERROR: {}", dispatcherName, ExceptionUtils.getStackTrace(e));
pipelineBackoff.backoff();
}
}
log.error("tried for {} times, failed batch...", retryForSingleTask.get());
pipeline.close();
return null;
});
inflight.add(f);
if ((counter % 20) == 0) {
log.debug("Clean inflight future");
inflight.removeIf(Future::isDone);
}
}
} catch (Exception e) {
log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
putQuiet(out, dispatcher.cookPoison());
}
}
private String info() {
ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool;
return "executor active tasks: %d; queue size: %d, completed tasks: %d".formatted(
casted.getActiveCount(),
casted.getQueue().size(),
casted.getCompletedTaskCount()
);
}
private boolean wasInterrupted() {
if (Thread.interrupted()) {
log.warn("pipeline was interrupted...");
//putQuiet(out, dispatcher.cookPoison());
return true;
}
return false;
}
public void startRunning() {
ioPool.submit(this);
}
public void close() {
log.info("Stage {} is getting shutdown", dispatcher.getClass().getSimpleName());
log.trace("Stage time stats: {}", stopWatch.info());
ioPool.shutdownNow(); // прерываем обработчики
}
void putQuiet(BlockingQueue<Z> q, Z b) {
try {
if (consumer != null) {
consumer.accept(b);
}
try (var m = stopWatch.start(dispatcherName + "-put")) {
q.put(b);
}
} catch (InterruptedException e) {
log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
Thread.currentThread().interrupt();
}
}
X takeQuiet() {
try (var m = stopWatch.start(dispatcherName + "-take")) {
return in.take();
} catch (InterruptedException e) {
log.error("Main thread for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e));
Thread.currentThread().interrupt();
return null;
}
}
public BlockingQueue<X> getIn() {
return in;
}
public void setConsumer(Consumer<Z> consumer) {
this.consumer = consumer;
}
public void setPipeline(Pipeline pipeline) {
this.pipeline = pipeline;
}
public void setBackoff(PipelineBackoff pipelineBackoff) {
this.pipelineBackoff = pipelineBackoff;
}
public void setContext(PipelineContext<?> context) {
this.context = context;
}
public PipelineContext<?> getContext() {
return context;
}
}

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java
View file

@ -12,9 +12,9 @@ import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.indic.PersonService;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.Handler;
import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriterV2;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.pipeline.engine.core.Handler;
import java.io.IOException;
import java.io.OutputStream;

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java
View file

@ -10,8 +10,8 @@ import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.model.impl.VoidPayload;
import ru.nbch.credit_tracker.service.indic.PersonService;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.Handler;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.pipeline.engine.core.Handler;
import java.util.List;
import java.util.Map;

8
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java
View file

@ -5,13 +5,17 @@ import org.springframework.stereotype.Service;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.entities.app.PhonesMap;
import ru.nbch.credit_tracker.entities.indic.Name;
import ru.nbch.credit_tracker.model.*;
import ru.nbch.credit_tracker.model.FlagData;
import ru.nbch.credit_tracker.model.IdData;
import ru.nbch.credit_tracker.model.NameData;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.indic.PersonService;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.Handler;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.pipeline.engine.core.Handler;
import ru.nbch.pipeline.engine.core.PipelineContext;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java
View file

@ -11,8 +11,8 @@ import ru.nbch.credit_tracker.service.UserPackageService;
import ru.nbch.credit_tracker.service.indic.PersonService;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.signal.PhonesMapService;
import ru.nbch.credit_tracker.service.task.Handler;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.pipeline.engine.core.Handler;
import java.util.ArrayList;
import java.util.List;

2
src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java
View file

@ -9,8 +9,8 @@ import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.indic.PersonService;
import ru.nbch.credit_tracker.service.signal.PhonesMapService;
import ru.nbch.credit_tracker.service.task.Handler;
import ru.nbch.credit_tracker.utils.CTUtil;
import ru.nbch.pipeline.engine.core.Handler;
import java.util.ArrayList;
import java.util.List;

11
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java
View file

@ -1,8 +1,5 @@
package ru.nbch.credit_tracker.service.task.impl;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@ -11,14 +8,18 @@ import ru.nbch.credit_tracker.entities.app.UserPackage;
import ru.nbch.credit_tracker.enums.PoisonType;
import ru.nbch.credit_tracker.exceptions.SignalClientException;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import ru.nbch.pipeline.engine.core.PipelineContext;
import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
private final Logger log = LoggerFactory.getLogger(getClass());

9
src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java
View file

@ -1,7 +1,5 @@
package ru.nbch.credit_tracker.service.task.impl;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@ -12,14 +10,17 @@ import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.IdData;
import ru.nbch.credit_tracker.model.NameData;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.SubjectProfile;
import ru.nbch.credit_tracker.model.impl.PackageVariable;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import ru.nbch.pipeline.engine.core.PipelineContext;
import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
import java.util.List;
import java.util.concurrent.BlockingQueue;
public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
private final Logger log = LoggerFactory.getLogger(getClass());

9
src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java
View file

@ -1,13 +1,14 @@
package ru.nbch.credit_tracker.service.task.impl;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload;
import ru.nbch.credit_tracker.model.impl.VoidPayload;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@Slf4j
public final class NoOpTerminator implements IPipeLineTerminator<VoidPayload, EmptyTerminatePayload> {

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

@ -1,10 +1,15 @@
package ru.nbch.credit_tracker.service.task.impl;
import java.io.BufferedWriter;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
import ru.nbch.pipeline.engine.core.setup.IPipeLineTerminator;
import ru.nbch.pipeline.engine.utils.TaskTimer;
import javax.xml.stream.*;
import java.io.*;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
@ -12,19 +17,6 @@ import java.nio.file.StandardOpenOption;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLOutputFactory;
import javax.xml.stream.XMLStreamConstants;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.XMLStreamReader;
import javax.xml.stream.XMLStreamWriter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
import ru.nbch.credit_tracker.utils.TaskTimer;
@Slf4j
public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload> {

11
src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java
View file

@ -1,11 +0,0 @@
package ru.nbch.credit_tracker.service.task.setup;
import java.util.concurrent.BlockingQueue;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.StagePayload;
public interface IPipeLineStarter<Q extends StagePayload> {
void startFillingQueue(BlockingQueue<Q> queue);
default void setContext(PipelineContext<?> context) {}
}

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

@ -1,11 +0,0 @@
package ru.nbch.credit_tracker.service.task.setup;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import ru.nbch.credit_tracker.model.StagePayload;
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> doneFuture();
}

100
src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilder.java
View file

@ -1,100 +0,0 @@
package ru.nbch.credit_tracker.service.task.setup;
import ru.nbch.credit_tracker.model.PipelineContext;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelineBackoff;
import ru.nbch.credit_tracker.service.task.Stage;
import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
public class PipeLineBuilder<FIRST extends StagePayload, LAST extends StagePayload> {
private final List<Stage<?, ?>> stages;
private Stage<FIRST, ?> firstStage;
private BlockingQueue<LAST> lastQueue;
private IPipeLineStarter<FIRST> pipeLineStarter;
private IPipeLineTerminator<LAST, ?> pipeLineTerminator;
private PipelineBackoff backoff;
private PipelineContext<?> context;
private PipeLineBuilder() {
this.stages = new ArrayList<>();
}
public static <I extends StagePayload, O extends StagePayload>
HandlerStep<I, I, O> first(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
PipeLineBuilder<I, O> builder = new PipeLineBuilder<>();
builder.lastQueue = outputQueue;
HandlerStep<I, I, O> step = StageBuilder.create(inputQueue, outputQueue);
((StageBuilder<I, I, O>) step).setParent(builder);
return step;
}
@SuppressWarnings("unchecked")
public <NEXT extends StagePayload>
HandlerStep<FIRST, LAST, NEXT> stage(BlockingQueue<NEXT> outputQueue) {
HandlerStep<FIRST, LAST, NEXT> step = StageBuilder.create(lastQueue, outputQueue);
lastQueue = (BlockingQueue<LAST>) outputQueue;
((StageBuilder<FIRST, LAST, NEXT>) step).setParent((PipeLineBuilder<FIRST, NEXT>) this);
return step;
}
public PipeLineBuilder<FIRST, LAST> fillInitialQueue(IPipeLineStarter<FIRST> pipeLineStarter) {
Objects.requireNonNull(pipeLineStarter, "pipeLineStarter is required");
if (firstStage == null) {
throw new IllegalStateException("Pipeline has no stages");
}
this.pipeLineStarter = pipeLineStarter;
return this;
}
public PipeLineBuilder<FIRST, LAST> pollFinishingQueue(IPipeLineTerminator<LAST, ?> pipeLineTerminator) {
Objects.requireNonNull(pipeLineTerminator, "pipeLineTerminator is required");
this.pipeLineTerminator = pipeLineTerminator;
return this;
}
public PipeLineBuilder<FIRST, LAST> backoff(PipelineBackoff backoff) {
this.backoff = backoff;
return this;
}
public <T extends Enum<T>>
PipeLineBuilder<FIRST, LAST> context(Class<T> contextKeyClass) {
this.context = new PipelineContext<>(contextKeyClass);
return this;
}
public Pipeline build() {
Pipeline p = new Pipeline();
p.setStarter(() -> {
stages.forEach(stage -> {
stage.setPipeline(p);
stage.setBackoff(backoff);
stage.setContext(context);
stage.startRunning();
});
pipeLineStarter.setContext(context);
pipeLineStarter.startFillingQueue(firstStage.getIn());
});
pipeLineTerminator.setPollingQueue(lastQueue);
p.setTerminator(pipeLineTerminator);
p.setStages(stages);
p.setContext(context);
return p;
}
@SuppressWarnings("unchecked")
<I extends StagePayload, O extends StagePayload>
void addStage(Stage<I, O> stage) {
Objects.requireNonNull(stage, "stage is required");
if (stages.isEmpty()) {
this.firstStage = (Stage<FIRST, ?>) stage;
}
stages.add(stage);
}
}

76
src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilder.java
View file

@ -1,76 +0,0 @@
package ru.nbch.credit_tracker.service.task.setup;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.refactoring.Suspicios;
import ru.nbch.credit_tracker.service.task.Handler;
import ru.nbch.credit_tracker.service.task.Stage;
import ru.nbch.credit_tracker.service.task.setup.step.BuildStep;
import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.function.Consumer;
/**
* Step Builder для Stage:
* Обязательная последовательность: handler(...) -> poison(...) -> buildStage()
*/
public class StageBuilder<FIRST extends StagePayload, I extends StagePayload, O extends StagePayload>
implements HandlerStep<FIRST, I, O>, BuildStep<FIRST, O> {
private PipeLineBuilder<FIRST, O> parent;
private final BlockingQueue<I> inputQueue;
private final BlockingQueue<O> outputQueue;
private Handler<I, O> handler;
private Consumer<O> consumer = null;
private StageBuilder(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
this.inputQueue = Objects.requireNonNull(inputQueue, "inputQueue is required");
this.outputQueue = Objects.requireNonNull(outputQueue, "outputQueue is required");
}
/** Стартовая фабрика возвращает первый шаг — HandlerStep */
public static <FIRST extends StagePayload, I extends StagePayload, O extends StagePayload>
HandlerStep<FIRST, I, O> create(BlockingQueue<I> inputQueue, BlockingQueue<O> outputQueue) {
return new StageBuilder<>(inputQueue, outputQueue);
}
void setParent(PipeLineBuilder<FIRST, O> parent) {
this.parent = Objects.requireNonNull(parent, "parent is required");
}
// ===================== ШАГИ =====================
/** Шаг 1: задать обработчик */
@Override
public BuildStep<FIRST, O> handler(Handler<I, O> handler) {
this.handler = Objects.requireNonNull(handler, "handler is required");
return this;
}
@Suspicios("Удалить?")
@Override
public BuildStep<FIRST, O> handlerResult(Consumer<O> consumer) {
this.consumer = Objects.requireNonNull(consumer, "consumer is required");
return this;
}
/** Шаг 3: собрать Stage и вернуться к PipeLineBuilder */
@Override
public PipeLineBuilder<FIRST, O> andThen() {
if (parent == null) {
throw new IllegalStateException("parent PipeLineBuilder is not set");
}
if (handler == null) {
throw new IllegalStateException("handler must be set before buildStage()");
}
Stage<I, O> stage = new Stage<>(handler, inputQueue, outputQueue);
stage.setConsumer(consumer);
parent.addStage(stage);
return parent;
}
// ===================== Step-interfaces =====================
}

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

@ -1,14 +0,0 @@
package ru.nbch.credit_tracker.service.task.setup.step;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.refactoring.Suspicios;
import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilder;
import java.util.function.Consumer;
public interface BuildStep<FIRST extends StagePayload, O extends StagePayload> {
PipeLineBuilder<FIRST, O> andThen();
@Suspicios("Удалить?")
BuildStep<FIRST, O> handlerResult(Consumer<O> consumer);
}

8
src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java
View file

@ -1,8 +0,0 @@
package ru.nbch.credit_tracker.service.task.setup.step;
import ru.nbch.credit_tracker.model.StagePayload;
import ru.nbch.credit_tracker.service.task.Handler;
public interface HandlerStep<FIRST extends StagePayload, I extends StagePayload, O extends StagePayload> {
BuildStep<FIRST, O> handler(Handler<I, O> handler);
}

3
src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java
View file

@ -1,12 +1,13 @@
package ru.nbch.credit_tracker.utils;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import java.util.Map;
// TODO update library to use provided LogUtil
public class LogUtil {
private final static Logger log = LoggerFactory.getLogger(LogUtil.class);
public static final String mdcKey = "ru.nbch.credit_tracker.log.mdc.key";

166
src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java
View file

@ -1,166 +0,0 @@
package ru.nbch.credit_tracker.utils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;
public class TaskTimer {
private final ConcurrentMap<String, TaskStats> stats = new ConcurrentHashMap<>();
/**
* Начать измерение времени для задачи с именем taskName.
* Используется обычно через try-with-resources:
*
* try (var m = timer.start("dispatcher.handle")) {
* dispatcher.handle(b);
* }
*/
public Measurement start(String taskName) {
return new Measurement(this, taskName, System.nanoTime());
}
void record(String taskName, long durationNanos) {
TaskStats s = stats.computeIfAbsent(taskName, key -> new TaskStats());
s.record(durationNanos);
}
/**
* Снимок статистики по всем задачам на текущий момент.
*/
public Map<String, StatsSnapshot> snapshot() {
return stats.entrySet().stream()
.collect(Collectors.toUnmodifiableMap(
Map.Entry::getKey,
e -> e.getValue().snapshot()
));
}
/**
* Снимок статистики по одной задаче.
*/
public StatsSnapshot snapshot(String taskName) {
TaskStats s = stats.get(taskName);
return s == null ? StatsSnapshot.empty() : s.snapshot();
}
public String info() {
return snapshot().entrySet().stream().map(stringStatsSnapshotEntry ->
"%s: %s".formatted(stringStatsSnapshotEntry.getKey(), stringStatsSnapshotEntry.getValue())).collect(Collectors.joining("\n", "\n", ""));
}
// ---------- Вспомогательные классы ----------
public static final class StatsSnapshot {
public final long count;
public final long totalNanos;
public final long minNanos;
public final long maxNanos;
public StatsSnapshot(long count, long totalNanos, long minNanos, long maxNanos) {
this.count = count;
this.totalNanos = totalNanos;
this.minNanos = minNanos;
this.maxNanos = maxNanos;
}
public static StatsSnapshot empty() {
return new StatsSnapshot(0, 0, 0, 0);
}
public double avgMillis() {
return count == 0 ? 0.0 : (totalNanos / 1_000_000.0) / count;
}
public double totalMillis() {
return totalNanos / 1_000_000.0;
}
public double minMillis() {
return minNanos == 0 ? 0.0 : minNanos / 1_000_000.0;
}
public double maxMillis() {
return maxNanos == 0 ? 0.0 : maxNanos / 1_000_000.0;
}
@Override
public String toString() {
return "StatsSnapshot{" +
"count=" + count +
", totalMillis=" + totalMillis() +
", avgMillis=" + avgMillis() +
", minMillis=" + minMillis() +
", maxMillis=" + maxMillis() +
'}';
}
}
private static final class TaskStats {
private final LongAdder count = new LongAdder();
private final LongAdder totalNanos = new LongAdder();
private final AtomicLong minNanos = new AtomicLong(Long.MAX_VALUE);
private final AtomicLong maxNanos = new AtomicLong(Long.MIN_VALUE);
void record(long durationNanos) {
count.increment();
totalNanos.add(durationNanos);
// min
for (;;) {
long current = minNanos.get();
if (durationNanos >= current) {
break;
}
if (minNanos.compareAndSet(current, durationNanos)) {
break;
}
}
// max
for (;;) {
long current = maxNanos.get();
if (durationNanos <= current) {
break;
}
if (maxNanos.compareAndSet(current, durationNanos)) {
break;
}
}
}
StatsSnapshot snapshot() {
long c = count.sum();
if (c == 0) {
return StatsSnapshot.empty();
}
long total = totalNanos.sum();
long min = minNanos.get();
long max = maxNanos.get();
if (min == Long.MAX_VALUE) min = 0;
if (max == Long.MIN_VALUE) max = 0;
return new StatsSnapshot(c, total, min, max);
}
}
public final class Measurement implements AutoCloseable {
private final String taskName;
private final long startNanos;
private boolean closed = false;
private Measurement(TaskTimer timer, String taskName, long startNanos) {
this.taskName = taskName;
this.startNanos = startNanos;
}
@Override
public void close() {
if (closed) return;
closed = true;
long duration = System.nanoTime() - startNanos;
TaskTimer.this.record(taskName, duration);
}
}
}