diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java index 581fe10..96ff9d0 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -7,7 +7,6 @@ import java.nio.file.Path; import java.nio.file.StandardCopyOption; import java.time.LocalDateTime; import java.util.List; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import javax.xml.stream.XMLStreamException; import lombok.extern.slf4j.Slf4j; @@ -41,7 +40,8 @@ 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.PipelinePackageManager; +import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; +import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; import ru.nbch.credit_tracker.service.task.impl.Saver; @@ -50,7 +50,6 @@ 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.credit_tracker.utils.error.ExceptionUtils; import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp; import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp; @@ -62,21 +61,22 @@ public class SignalFacadeV3 { private final CreditTrackerSettings config; private final UserPackageService userPackageService; private final AuthorizationService authorizationService; - private final PipelinePackageManager pipelinePackageManager; + private final PipelinePackageFabric pipelinePackageManager; private final SignalRestService signalRestService; private final Saver saver; private final ExecutorService pipelineExecutorService; private final ExecutorService saverExecutorService; + private final PipelinePackageOrchestrator pipelineOrchestrator; public SignalFacadeV3(MonitoringRequestValidator monitoringRequestValidator, CreditTrackerSettings config, UserPackageService userPackageService, AuthorizationService authorizationService, - PipelinePackageManager pipelinePackageManager, + PipelinePackageFabric pipelinePackageManager, SignalRestService signalRestService, Saver saver, ExecutorService pipelineExecutorService, - ExecutorService saverExecutorService) { + ExecutorService saverExecutorService, PipelinePackageOrchestrator pipelineOrchestrator) { this.monitoringRequestValidator = monitoringRequestValidator; this.config = config; this.userPackageService = userPackageService; @@ -86,6 +86,7 @@ public class SignalFacadeV3 { this.saverExecutorService = saverExecutorService; this.pipelineExecutorService = pipelineExecutorService; this.saver = saver; + this.pipelineOrchestrator = pipelineOrchestrator; } public void setupMonitoringRequest(MultipartFile file) { @@ -150,18 +151,11 @@ public class SignalFacadeV3 { Long packageId = userPackageService.registerPackage(userId, packageName); LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId)); log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name()); - Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT)); - CompletableFuture - .runAsync(() -> { - try { - saver.readAndSave(packageId, parsedFilePath); - CtFileUtils.deleteQuietly(parsedFilePath); - } catch (Throwable t) { - log.error(ExceptionUtils.getStackTrace(t)); - pipeline.close(); - } - }, saverExecutorService) - .thenRun(() -> pipeline.start(pipelineExecutorService)); + Pipeline pipeline = pipelinePackageManager.build(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT)); + pipelineOrchestrator.startOrDelay(() -> { + saver.readAndSave(packageId, parsedFilePath); + CtFileUtils.deleteQuietly(parsedFilePath); + }, pipeline, packageId); } catch (Exception e) { try { Files.delete(parsedFilePath); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java index 3b0f16c..686cbc7 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java @@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.service; import java.nio.file.Path; import java.util.List; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.EnableAspectJAutoProxy; @@ -13,7 +12,8 @@ import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.task.Pipeline; -import ru.nbch.credit_tracker.service.task.PipelinePackageManager; +import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; +import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; import ru.nbch.credit_tracker.service.task.impl.Saver; @@ -31,7 +31,8 @@ public class UpdateProcessingPackagesService { private final UserPackageMapper userPackageMapper; private final PhoneFidMapMapper phoneFidMapMapper; - private final PipelinePackageManager pipelinePackageManager; + private final PipelinePackageFabric pipelinePackageManager; + private final PipelinePackageOrchestrator pipelineOrchestrator; private final CreditTrackerSettings creditTrackerSettings; private final Saver saver; private final ExecutorService pipelineExecutorService; @@ -39,7 +40,7 @@ public class UpdateProcessingPackagesService { public UpdateProcessingPackagesService(UserPackageMapper userPackageMapper, PhoneFidMapMapper phoneFidMapMapper, - PipelinePackageManager pipelinePackageManager, + PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator, CreditTrackerSettings creditTrackerSettings, Saver saver, ExecutorService pipelineExecutorService, @@ -47,6 +48,7 @@ public class UpdateProcessingPackagesService { this.userPackageMapper = userPackageMapper; this.phoneFidMapMapper = phoneFidMapMapper; this.pipelinePackageManager = pipelinePackageManager; + this.pipelineOrchestrator = pipelineOrchestrator; this.creditTrackerSettings = creditTrackerSettings; this.saver = saver; this.pipelineExecutorService = pipelineExecutorService; @@ -68,17 +70,11 @@ public class UpdateProcessingPackagesService { log.trace("Deleted {} rows", count); Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage); log.debug("Try to read and build pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath); - Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT)); - CompletableFuture - .runAsync(() -> { - try { - saver.readAndSave(userPackage.getId(), xmlReqPath); - } catch (Throwable t) { - log.error(ExceptionUtils.getStackTrace(t)); - pipeline.close(); - } - }, saverExecutorService) - .thenRun(() -> pipeline.start(pipelineExecutorService)); + Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT)); + pipelineOrchestrator.startOrDelay( + () -> saver.readAndSave(userPackage.getId(), xmlReqPath), + pipeline, userPackage.getId() + ); } catch (Exception e) { log.error("Error build default pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); } @@ -91,8 +87,8 @@ public class UpdateProcessingPackagesService { log.debug("Found {} packages in status PROCESSING and rejected_status = null", packages.size()); for (UserPackage userPackage : packages) { try { - Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE)); - pipeline.start(pipelineExecutorService); + Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE)); + pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); } catch (Exception e) { log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); } @@ -104,8 +100,8 @@ public class UpdateProcessingPackagesService { log.debug("Found {} packages in status PROCESSING_REPEATED", packages.size()); for (UserPackage userPackage : packages) { try { - Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE)); - pipeline.start(pipelineExecutorService); + Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE)); + pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); } catch (Exception e) { log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index d126b61..48998f0 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -14,7 +14,8 @@ import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.Pipeline; -import ru.nbch.credit_tracker.service.task.PipelinePackageManager; +import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; +import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; @@ -33,7 +34,8 @@ public class UpdatingFlagsService { private final Long daysToUpdateFlags; private final PhoneFidMapMapper phoneFidMapMapper; private final UserPackageMapper userPackageMapper; - private final PipelinePackageManager pipelinePackageManager; + private final PipelinePackageFabric pipelinePackageManager; + private final PipelinePackageOrchestrator pipelineOrchestrator; private final ExecutorService pipelineExecutorService; public UpdatingFlagsService(SignalService signalService, @@ -41,7 +43,7 @@ public class UpdatingFlagsService { MonitoringErrorProcesingService monitoringErrorProcesingService, CreditTrackerSettings config, PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper, - PipelinePackageManager pipelinePackageManager, + PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator, ExecutorService pipelineExecutorService) { this.signalService = signalService; this.signalRestService = signalRestService; @@ -50,6 +52,7 @@ public class UpdatingFlagsService { this.phoneFidMapMapper = phoneFidMapMapper; this.userPackageMapper = userPackageMapper; this.pipelinePackageManager = pipelinePackageManager; + this.pipelineOrchestrator = pipelineOrchestrator; this.pipelineExecutorService = pipelineExecutorService; } @@ -65,8 +68,8 @@ public class UpdatingFlagsService { if (hasOldRow) { try { userPackageMapper.updatePackageStatus(userPackage.getId(), Status.PROCESSING_REPEATED.name()); - Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT)); - pipeline.start(pipelineExecutorService); + Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT)); + pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); } catch (Exception e) { log.error("Failed to build pipeline", e); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java similarity index 66% rename from src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java rename to src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java index 19aaa4e..1bba46f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java @@ -8,23 +8,19 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.task.StageQueueFactory; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService; -import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; -import ru.nbch.credit_tracker.service.task.impl.Saver; 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.PipeLineBuilderV2; @Component @Slf4j -public class PipelinePackageManager implements InitializingBean { +public class PipelinePackageFabric implements InitializingBean { private final FlagsEnricher flagsEnricherV2; private final StageQueueFactory idEnricherQueueFactory; private final StageQueueFactory nameEnricherQueueFactory; @@ -36,18 +32,15 @@ public class PipelinePackageManager implements InitializingBean { private final CreditTrackerSettings creditTrackerSettings; private final PipelineBackoff pipelineBackoff; - public PipelinePackageManager(SignalRestService signalRestService, - @Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, - @Qualifier("nameEnricherQueueFactory") StageQueueFactory nameEnricherQueueFactory, - @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, - UserPackageMapper userPackageMapper, - FidIdEnricher fidIdEnricherHandler, - Supplier> pipelineTerminatorFactory, - Function> pipelineStarterFactory, - FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher, - CreditTrackerSettings creditTrackerSettings, - PipelineBackoff pipelineBackoff, Saver saver, - MonitoringErrorProcesingService monitoringErrorProcesingService) { + public PipelinePackageFabric(@Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, + @Qualifier("nameEnricherQueueFactory") StageQueueFactory nameEnricherQueueFactory, + @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, + FidIdEnricher fidIdEnricherHandler, + Supplier> pipelineTerminatorFactory, + Function> pipelineStarterFactory, + FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher, + CreditTrackerSettings creditTrackerSettings, + PipelineBackoff pipelineBackoff) { this.idEnricherQueueFactory = idEnricherQueueFactory; this.nameEnricherQueueFactory = nameEnricherQueueFactory; this.sendSignalQueueFactory = sendSignalQueueFactory; @@ -60,7 +53,7 @@ public class PipelinePackageManager implements InitializingBean { this.pipelineBackoff = pipelineBackoff; } - public Pipeline buildAndRun(SetupParam setupParam) throws Exception { + public Pipeline build(SetupParam setupParam) throws Exception { return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject()) .handler(fidNameEnricher) .handlerResult(flagsEnricherV2) 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 d3962ac..a71cdda 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 @@ -62,7 +62,7 @@ public class PipelinePackageOrchestrator { innerStartOrDelay(before, pipeline, packageId); } - public void innerStartOrDelay(Runnable before, Pipeline pipeline, Long packageId) { + private void innerStartOrDelay(Runnable before, Pipeline pipeline, Long packageId) { try { if (before != null) { CompletableFuture @@ -89,7 +89,7 @@ public class PipelinePackageOrchestrator { } } - public void addCallback(Pipeline pipeline, Long packageId) { + private void addCallback(Pipeline pipeline, Long packageId) { pipeline.getTerminator() .doneFuture() .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS) @@ -134,7 +134,6 @@ public class PipelinePackageOrchestrator { return; } } - log.info("delayed pipeline found: packageId={}. continue work...", delayedLaunch.packageId); innerStartOrDelay(delayedLaunch.before, delayedLaunch.pipeline, delayedLaunch.packageId); }