diff --git a/pom.xml b/pom.xml index 57ac355..0ad9770 100644 --- a/pom.xml +++ b/pom.xml @@ -38,6 +38,11 @@ mybatis-spring-boot-starter 3.0.4 + + ru.nbch + pipeline-engine + 1.0 + org.postgresql postgresql 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 f7e27c0..8d00b9d 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 @@ -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> noOpTerminatorFactory() { return NoOpTerminator::new; } + + @Bean + public PipelineBackoff pipelineBackoff(CreditTrackerSettings creditTrackerSettings) { + return new PipelineBackoff( + creditTrackerSettings.getPipeline().errorAttemptCount(), + creditTrackerSettings.getPipeline().errorAttemptDelay(), + creditTrackerSettings.getPipeline().retryableExceptions() + ); + } } diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java index 73e0fd6..bd0b170 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageQueueFactory.java @@ -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 implements FactoryBean> { diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java index f946516..68980ec 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -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; diff --git a/src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java b/src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java deleted file mode 100644 index 286619b..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/PipelineContext.java +++ /dev/null @@ -1,21 +0,0 @@ -package ru.nbch.credit_tracker.model; - -import java.util.EnumMap; - -public class PipelineContext> { - private final EnumMap storage; - - public PipelineContext(Class keyClass) { - this.storage = new EnumMap<>(keyClass); - } - - public void store(T key, Object value) { - storage.put(key, value); - } - - @SuppressWarnings("unchecked") - public S get(T key) { - Object o = storage.get(key); - return (S) o; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java b/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java deleted file mode 100644 index 0dc7b42..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/StagePayload.java +++ /dev/null @@ -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;} -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java b/src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java deleted file mode 100644 index f115291..0000000 --- a/src/main/java/ru/nbch/credit_tracker/model/TerminatePayload.java +++ /dev/null @@ -1,4 +0,0 @@ -package ru.nbch.credit_tracker.model; - -public interface TerminatePayload { -} diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java index 5db5767..a8989f0 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java @@ -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 { } diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java index f2cb147..4a2902d 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/PartFilePayload.java @@ -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 { diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java index 9955681..a69591f 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SignalXmlFilePayload.java @@ -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 { } diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java index 4eaf567..3d7b2f9 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java @@ -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 { diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java index 2eb4e8b..f1a6a88 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java @@ -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 { diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java index a692d50..eb223e8 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java @@ -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 diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java index 7c52e55..c456bc0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidAndPdService.java @@ -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 diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java index 5c039fd..d4d59ba 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/migration/MigrationService.java @@ -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 diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java deleted file mode 100644 index cb68c32..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java +++ /dev/null @@ -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 { - Z handle(X taskPayload); - - @SuppressWarnings("unchecked") - default > PipelineContext obtainContext(Class keyClass, X input) { - PipelineContext ctx = input.getContext(); - return (PipelineContext) ctx; - } - - @Suspicios("Удалить?") - default void handleOnPoison(X poison){ - //do nothing - } - - default int preferredConcurrency() {return 10;} - - Z cookPoison(); -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java b/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java deleted file mode 100644 index f765381..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java +++ /dev/null @@ -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> 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> getStages() { - return stages; - } - - public void setStages(List> stages) { - this.stages = stages; - } - - public long getStartTime() { - return startTime; - } - - public void setContext(PipelineContext context) { - this.context = context; - } - - public > PipelineContext obtainContext(Class keyClass) { - return (PipelineContext) context; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java deleted file mode 100644 index 327f565..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineBackoff.java +++ /dev/null @@ -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> 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())); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java index 046e722..cbbc4cf 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineMigrationPackageFabric.java @@ -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()) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java index 73c4bc1..f25f982 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java @@ -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()) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java index fabd1f0..dd4c632 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java @@ -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 { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java index a2964ef..1c49013 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java @@ -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 { 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 deleted file mode 100644 index be6155e..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java +++ /dev/null @@ -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 implements Runnable { - private final Handler dispatcher; - private final BlockingQueue in; - private final BlockingQueue out; - private final ExecutorService ioPool; - private Consumer 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 dispatcher, - BlockingQueue in, - BlockingQueue 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 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> 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 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 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 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 getIn() { - return in; - } - - public void setConsumer(Consumer 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; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java index fe5f66e..7d64305 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java @@ -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; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java index 6b623b3..f82c35e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java @@ -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; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java index 7ed9462..2feda07 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameEnricher.java @@ -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; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java index f2e211d..16918be 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java @@ -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; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java index 5013620..543965e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java @@ -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; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java index 09ae8f8..8ab5563 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java @@ -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 { private final Logger log = LoggerFactory.getLogger(getClass()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java index e07f862..f062e2e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java @@ -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 { private final Logger log = LoggerFactory.getLogger(getClass()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java index f736916..89644b2 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java @@ -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 { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java index 976fd38..b0e08a0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java @@ -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 { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java deleted file mode 100644 index 48a96e7..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineStarter.java +++ /dev/null @@ -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 { - void startFillingQueue(BlockingQueue queue); - - default void setContext(PipelineContext context) {} -} 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 deleted file mode 100644 index 82a8519..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/IPipeLineTerminator.java +++ /dev/null @@ -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 extends Runnable { - void setPollingQueue(BlockingQueue queue); - CompletableFuture doneFuture(); -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilder.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilder.java deleted file mode 100644 index 2eb8a0b..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilder.java +++ /dev/null @@ -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 { - private final List> stages; - private Stage firstStage; - private BlockingQueue lastQueue; - private IPipeLineStarter pipeLineStarter; - private IPipeLineTerminator pipeLineTerminator; - private PipelineBackoff backoff; - private PipelineContext context; - - private PipeLineBuilder() { - this.stages = new ArrayList<>(); - } - - public static - HandlerStep first(BlockingQueue inputQueue, BlockingQueue outputQueue) { - PipeLineBuilder builder = new PipeLineBuilder<>(); - builder.lastQueue = outputQueue; - HandlerStep step = StageBuilder.create(inputQueue, outputQueue); - ((StageBuilder) step).setParent(builder); - return step; - } - - @SuppressWarnings("unchecked") - public - HandlerStep stage(BlockingQueue outputQueue) { - HandlerStep step = StageBuilder.create(lastQueue, outputQueue); - lastQueue = (BlockingQueue) outputQueue; - ((StageBuilder) step).setParent((PipeLineBuilder) this); - return step; - } - - public PipeLineBuilder fillInitialQueue(IPipeLineStarter pipeLineStarter) { - Objects.requireNonNull(pipeLineStarter, "pipeLineStarter is required"); - if (firstStage == null) { - throw new IllegalStateException("Pipeline has no stages"); - } - this.pipeLineStarter = pipeLineStarter; - return this; - } - - public PipeLineBuilder pollFinishingQueue(IPipeLineTerminator pipeLineTerminator) { - Objects.requireNonNull(pipeLineTerminator, "pipeLineTerminator is required"); - this.pipeLineTerminator = pipeLineTerminator; - return this; - } - - public PipeLineBuilder backoff(PipelineBackoff backoff) { - this.backoff = backoff; - return this; - } - - public > - PipeLineBuilder context(Class 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") - - void addStage(Stage stage) { - Objects.requireNonNull(stage, "stage is required"); - if (stages.isEmpty()) { - this.firstStage = (Stage) stage; - } - stages.add(stage); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilder.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilder.java deleted file mode 100644 index c1799d0..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/StageBuilder.java +++ /dev/null @@ -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 - implements HandlerStep, BuildStep { - - private PipeLineBuilder parent; - private final BlockingQueue inputQueue; - private final BlockingQueue outputQueue; - private Handler handler; - private Consumer consumer = null; - - private StageBuilder(BlockingQueue inputQueue, BlockingQueue outputQueue) { - this.inputQueue = Objects.requireNonNull(inputQueue, "inputQueue is required"); - this.outputQueue = Objects.requireNonNull(outputQueue, "outputQueue is required"); - } - - /** Стартовая фабрика возвращает первый шаг — HandlerStep */ - public static - HandlerStep create(BlockingQueue inputQueue, BlockingQueue outputQueue) { - return new StageBuilder<>(inputQueue, outputQueue); - } - - void setParent(PipeLineBuilder parent) { - this.parent = Objects.requireNonNull(parent, "parent is required"); - } - - // ===================== ШАГИ ===================== - - /** Шаг 1: задать обработчик */ - @Override - public BuildStep handler(Handler handler) { - this.handler = Objects.requireNonNull(handler, "handler is required"); - return this; - } - - @Suspicios("Удалить?") - @Override - public BuildStep handlerResult(Consumer consumer) { - this.consumer = Objects.requireNonNull(consumer, "consumer is required"); - return this; - } - - /** Шаг 3: собрать Stage и вернуться к PipeLineBuilder */ - @Override - public PipeLineBuilder 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 stage = new Stage<>(handler, inputQueue, outputQueue); - stage.setConsumer(consumer); - parent.addStage(stage); - return parent; - } - - // ===================== Step-interfaces ===================== - -} 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 deleted file mode 100644 index 05d3bc0..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/BuildStep.java +++ /dev/null @@ -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 { - PipeLineBuilder andThen(); - - @Suspicios("Удалить?") - BuildStep handlerResult(Consumer consumer); -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java b/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java deleted file mode 100644 index 0f8ece4..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/setup/step/HandlerStep.java +++ /dev/null @@ -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 { - BuildStep handler(Handler handler); -} diff --git a/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java index 883bac7..267ab42 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java @@ -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"; diff --git a/src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java b/src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java deleted file mode 100644 index 7c4ec51..0000000 --- a/src/main/java/ru/nbch/credit_tracker/utils/TaskTimer.java +++ /dev/null @@ -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 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 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); - } - } -}