diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java index 5934d6c..3b68321 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java @@ -4,17 +4,18 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.ConfigurableApplicationContext; import ru.nbch.credit_tracker.service.MonitoringService; +import ru.nbch.credit_tracker.service.UpdateProcessingPackagesService; @SpringBootApplication public class CreditTrackerApplication { public static void main(String[] args) { ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args); -// UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class); + UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class); MonitoringService monitoringService = ctx.getBean(MonitoringService.class); // UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class); -// updateProcessingPackagesService.run(); - monitoringService.run(); + updateProcessingPackagesService.run(); +// monitoringService.run(); // updatingFlagsService.run(); } diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java new file mode 100644 index 0000000..158cc10 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/task/ExecutorsConfig.java @@ -0,0 +1,29 @@ +package ru.nbch.credit_tracker.config.task; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; + +@Configuration +public class ExecutorsConfig { + + @Bean + public ExecutorService pipelineExecutorService() { + return Executors.newFixedThreadPool(10, r -> { + Thread t = new Thread(r); + t.setName("pipe-starter-%d".formatted(t.threadId())); + return t; + }); + } + + @Bean + public ExecutorService saverExecutorService(CreditTrackerSettings creditTrackerSettings) { + return Executors.newFixedThreadPool(creditTrackerSettings.getPipeline().parallelRequests(), r -> { + Thread t = new Thread(r); + t.setName("subject-saver-%d".formatted(t.threadId())); + return t; + }); + } +} 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 d0a9b05..d42f3ee 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,7 +1,6 @@ package ru.nbch.credit_tracker.config.task; -import java.nio.file.Path; -import java.util.function.BiFunction; +import java.util.function.Function; import java.util.function.Supplier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -11,7 +10,9 @@ 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.indic.PhoneService2; -import ru.nbch.credit_tracker.service.task.impl.KickPipeLineImpl; +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.KickPipeLineRestoreImpl; 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; @@ -20,11 +21,16 @@ import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @Configuration public class PipelineStarterConfig { @Bean - public BiFunction> pipelineStarterFactory(StaxXmlProcessor xml, - CreditTrackerSettings creditTrackerSettings, - UserPackageMapper userPackageMapper, - PhoneService2 phoneService2) { - return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, userPackageMapper, phoneService2, creditTrackerSettings); + public Function> pipelineStarterFactory(StaxXmlProcessor xml, + CreditTrackerSettings creditTrackerSettings, + UserPackageMapper userPackageMapper, + PhoneService2 phoneService2) { + return (setupParam) -> switch (setupParam.mode()) { + case DEFAULT -> + new KickPipeLineDefaultImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + case RESTORE -> + new KickPipeLineRestoreImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + }; } @Bean 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 1f81b2b..3c0f423 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -7,6 +7,8 @@ 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; import org.springframework.core.io.FileUrlResource; @@ -38,11 +40,16 @@ 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.PipelinePackageManager; +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; import ru.nbch.credit_tracker.utils.CTUtil; 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; @@ -56,18 +63,28 @@ public class SignalFacadeV3 { private final AuthorizationService authorizationService; private final PipelinePackageManager pipelinePackageManager; private final SignalRestService signalRestService; + private final Saver saver; + private final ExecutorService pipelineExecutorService; + private final ExecutorService saverExecutorService; + public SignalFacadeV3(MonitoringRequestValidator monitoringRequestValidator, CreditTrackerSettings config, UserPackageService userPackageService, AuthorizationService authorizationService, PipelinePackageManager pipelinePackageManager, - SignalRestService signalRestService) { + SignalRestService signalRestService, + Saver saver, + ExecutorService pipelineExecutorService, + ExecutorService saverExecutorService) { this.monitoringRequestValidator = monitoringRequestValidator; this.config = config; this.userPackageService = userPackageService; this.authorizationService = authorizationService; this.pipelinePackageManager = pipelinePackageManager; this.signalRestService = signalRestService; + this.saverExecutorService = saverExecutorService; + this.pipelineExecutorService = pipelineExecutorService; + this.saver = saver; } public void setupMonitoringRequest(MultipartFile file) { @@ -129,7 +146,17 @@ 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()); - pipelinePackageManager.buildAndRun(packageId, parsedFilePath); + Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT)); + CompletableFuture + .runAsync(() -> { + try { + saver.readAndSave(packageId, parsedFilePath); + } catch (Throwable t) { + log.error(ExceptionUtils.getStackTrace(t)); + pipeline.close(); + } + }, saverExecutorService) + .thenRun(() -> pipeline.start(pipelineExecutorService)); } catch (Exception e) { try { Files.delete(parsedFilePath); diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java index 9ccdaf1..18bcff8 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhoneFidMapMapper.java @@ -18,6 +18,7 @@ public interface PhoneFidMapMapper { void registerWithId(@Param("phoneFidMap") List phoneFidMap); List findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset); + List findNullFidPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset); void updatePhoneFidMaps(@Param("subjects") Collection subjects); void updateOne(@Param("subject") PhoneFidMap subject); @@ -28,4 +29,6 @@ public interface PhoneFidMapMapper { PhoneFidMap findPhoneFidMapById(@Param("id") Long packageId); boolean hasNonNullFids(@Param("packageId") Long packageId); + + Long deleteByPackageId(@Param("packageId") Long id); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java index 9ecf0e0..9b8b6c1 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -35,7 +35,7 @@ import ru.nbch.credit_tracker.service.indic.PhoneService; import ru.nbch.credit_tracker.service.indic.PhoneService2; import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; -import ru.nbch.credit_tracker.service.signal.UserPackageService; +import ru.nbch.credit_tracker.service.signal.UserPackageDao; import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; import ru.nbch.credit_tracker.utils.CTUtil; @@ -48,7 +48,7 @@ public class SignalService { private final CreditTrackerSettings config; private final HitsService hitsService; - private final UserPackageService userPackageService; + private final UserPackageDao userPackageService; private final PhoneFidMapService phoneFidMapService; private final PhoneService2 phoneService2; private final PhoneService phoneService; @@ -60,7 +60,7 @@ public class SignalService { public SignalService(CreditTrackerSettings config, HitsService hitsService, - UserPackageService userPackageService, + UserPackageDao userPackageService, PhoneFidMapService phoneFidMapService, PhoneService2 phoneService2, PhoneService phoneService, 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 25c38a3..02a8754 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java @@ -1,15 +1,27 @@ 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.aop.framework.AopContext; import org.springframework.context.annotation.EnableAspectJAutoProxy; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.UserPackage; 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.signal.SignalRestService; - -import java.util.List; +import ru.nbch.credit_tracker.service.task.Pipeline; +import ru.nbch.credit_tracker.service.task.PipelinePackageManager; +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; +import ru.nbch.credit_tracker.utils.CTUtil; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; /** * Сервис для завершения обработки пакетов находящихся в статусе PROCESSING @@ -23,25 +35,90 @@ public class UpdateProcessingPackagesService { private UpdateProcessingPackagesService self; private final SignalService signalService; + private final UserPackageMapper userPackageMapper; + private final PhoneFidMapMapper phoneFidMapMapper; private final SignalRestService signalRestService; private final MonitoringErrorProcesingService monitoringErrorProcesingService; + private final PipelinePackageManager pipelinePackageManager; + private final CreditTrackerSettings creditTrackerSettings; + private final Saver saver; + private final ExecutorService pipelineExecutorService; + private final ExecutorService saverExecutorService; public UpdateProcessingPackagesService(SignalService signalService, + UserPackageMapper userPackageMapper, + PhoneFidMapMapper phoneFidMapMapper, SignalRestService signalRestService, - MonitoringErrorProcesingService monitoringErrorProcesingService) { + MonitoringErrorProcesingService monitoringErrorProcesingService, + PipelinePackageManager pipelinePackageManager, + CreditTrackerSettings creditTrackerSettings, + Saver saver, + ExecutorService pipelineExecutorService, + ExecutorService saverExecutorService) { this.signalService = signalService; + this.userPackageMapper = userPackageMapper; + this.phoneFidMapMapper = phoneFidMapMapper; this.signalRestService = signalRestService; this.monitoringErrorProcesingService = monitoringErrorProcesingService; + this.pipelinePackageManager = pipelinePackageManager; + this.creditTrackerSettings = creditTrackerSettings; + this.saver = saver; + this.pipelineExecutorService = pipelineExecutorService; + this.saverExecutorService = saverExecutorService; } public void run() { - log.info("Start updating processing packages"); - List packages = signalService.findPackagesOnProcessing(); + acceptedRestore(); + processingRestore(); +// log.info("Start updating processing packages"); +// List packages = signalService.findPackagesOnProcessing(); +// for (UserPackage userPackage : packages) { +// log.info("Updating flags for Package {}", userPackage.getPackageName()); +// getCurrentProxy().process(userPackage.getId(), userPackage.getPackageName()); +// } +// log.info("Finish updating processing packages"); + } + + + private void acceptedRestore() { + List packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name()); + log.debug("Found {} packages in status ACCEPTED", packages.size()); for (UserPackage userPackage : packages) { - log.info("Updating flags for Package {}", userPackage.getPackageName()); - getCurrentProxy().process(userPackage.getId(), userPackage.getPackageName()); + try { + Long count = phoneFidMapMapper.deleteByPackageId(userPackage.getId()); + 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); + //todo ???запускает асинхронный процесс без ожидания, на старте мы должны восстановиться один раз и + // видимо дождаться выполнения, иначе может не хватить пула, памяти или чего еще + 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)); + } catch (Exception e) { + log.error("Error build default pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); + } + } + } + + private void processingRestore() { + List packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name()); + log.debug("Found {} packages in status PROCESSING", packages.size()); + for (UserPackage userPackage : packages) { + try { + Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE)); + pipeline.start(pipelineExecutorService); + } catch (Exception e) { + log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); + } } - log.info("Finish updating processing packages"); } @SetAndClearMdc(argIndex = 1) diff --git a/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java b/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java index 1681fe0..0805e96 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java @@ -7,13 +7,14 @@ import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.enums.Status; +import ru.nbch.credit_tracker.service.signal.UserPackageDao; @Slf4j @Service public class UserPackageService { - private final ru.nbch.credit_tracker.service.signal.UserPackageService userPackageService; + private final UserPackageDao userPackageService; - public UserPackageService(ru.nbch.credit_tracker.service.signal.UserPackageService userPackageService) { + public UserPackageService(UserPackageDao userPackageService) { this.userPackageService = userPackageService; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java index 8132099..4771996 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java @@ -91,7 +91,12 @@ public class PhoneService2 { } } - public List nextPage(Long packageId, Integer limit, Integer offset) { + public List pageByPackageId(Long packageId, Integer limit, Integer offset) { return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset); } + + public List pageByPackageIdWithNullFids(Long packageId, Integer limit, Integer offset) { + return phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset); + } + } diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/UserPackageService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/UserPackageDao.java similarity index 96% rename from src/main/java/ru/nbch/credit_tracker/service/signal/UserPackageService.java rename to src/main/java/ru/nbch/credit_tracker/service/signal/UserPackageDao.java index 8595b52..ab2998b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/UserPackageService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/UserPackageDao.java @@ -9,11 +9,11 @@ import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; @Service -public class UserPackageService { +public class UserPackageDao { private final UserPackageMapper mapper; - public UserPackageService(UserPackageMapper mapper) { + public UserPackageDao(UserPackageMapper mapper) { this.mapper = mapper; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java index d7aed44..a65bad6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java @@ -1,13 +1,9 @@ package ru.nbch.credit_tracker.service.task; -import java.nio.file.Path; import java.time.LocalDateTime; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; -import java.util.function.BiFunction; +import java.util.function.Function; import java.util.function.Supplier; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.InitializingBean; @@ -42,18 +38,10 @@ public class PipelinePackageManager implements InitializingBean { private final FidIdEnricher fidIdEnricherHandler; private final FidNameEnricher fidNameEnricher; private final Supplier> pipelineTerminatorFactory; - private final BiFunction> pipelineStarterFactory; + private final Function> pipelineStarterFactory; private final CreditTrackerSettings creditTrackerSettings; - private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> { - Thread t = new Thread(r); - t.setName("pipe-starter-%d".formatted(t.threadId())); - return t; - }); - - private final ExecutorService subjectSaverExecutor; private final Semaphore semaphore; private final PipelineBackoff pipelineBackoff; - private final Saver saver; public PipelinePackageManager(SignalRestService signalRestService, @Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, @@ -62,7 +50,7 @@ public class PipelinePackageManager implements InitializingBean { UserPackageMapper userPackageMapper, FidIdEnricher fidIdEnricherHandler, Supplier> pipelineTerminatorFactory, - BiFunction> pipelineStarterFactory, + Function> pipelineStarterFactory, FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher, CreditTrackerSettings creditTrackerSettings, PipelineBackoff pipelineBackoff, Saver saver) { @@ -79,19 +67,15 @@ public class PipelinePackageManager implements InitializingBean { this.creditTrackerSettings = creditTrackerSettings; this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); this.pipelineBackoff = pipelineBackoff; - this.saver = saver; - this.subjectSaverExecutor = Executors.newFixedThreadPool(creditTrackerSettings.getPipeline().parallelRequests(), r -> { - Thread t = new Thread(r); - t.setName("subject-saver-%d".formatted(t.threadId())); - return t; - }); } - public void buildAndRun(Long packageId, Path pkgPath) throws Exception { + public Pipeline buildAndRun(SetupParam setupParam) throws Exception { if (!semaphore.tryAcquire()) { throw new RuntimeException("Too many concurrent requests"); } log.info("throttler permits after acquire: {}", semaphore.availablePermits()); + + Long packageId = setupParam.packageId(); long startTime = System.currentTimeMillis(); try { Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject()) @@ -101,7 +85,7 @@ public class PipelinePackageManager implements InitializingBean { .stage(sendSignalQueueFactory.getObject()) .handler(fidIdEnricherHandler) .andThen() - .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath)) + .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) .pollFinishingQueue(pipelineTerminatorFactory.get()) .semaphore(semaphore) .backoff(pipelineBackoff) @@ -136,17 +120,7 @@ public class PipelinePackageManager implements InitializingBean { log.info("semaphore amount: {}", semaphore.availablePermits()); } }); - - CompletableFuture - .runAsync(() -> { - try { - saver.readAndSave(packageId, pkgPath); - } catch (Throwable t) { - log.error(ExceptionUtils.getStackTrace(t)); - pipeline.close(); - } - }, subjectSaverExecutor) - .thenRun(() -> pipeline.start(executorService)); + return pipeline; } catch (Exception e) { semaphore.release(); throw e; diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/SetupParam.java b/src/main/java/ru/nbch/credit_tracker/service/task/SetupParam.java new file mode 100644 index 0000000..fe2dc38 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/SetupParam.java @@ -0,0 +1,10 @@ +package ru.nbch.credit_tracker.service.task; + +import java.nio.file.Path; +import ru.nbch.credit_tracker.service.task.impl.PipelineMode; + +public record SetupParam( + Long packageId, + Path xmlReqPath, + PipelineMode mode) { +} 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 new file mode 100644 index 0000000..4939565 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLine.java @@ -0,0 +1,96 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.nio.file.Path; +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; +import ru.nbch.credit_tracker.entities.app.PhoneFidMap; +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.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.service.indic.PhoneService2; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; +import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; + +public abstract class KickPipeLine implements IPipeLineStarter { + private final Logger log = LoggerFactory.getLogger(getClass()); + protected final Long packageId; + private final Path pkgPath; + private final StaxXmlProcessor xmlProcessor; + protected final UserPackageMapper userPackageMapper; + protected final PhoneService2 phoneService2; + private final int BATCH_SIZE; + private boolean isInterrupted = false; + + + public KickPipeLine(StaxXmlProcessor xmlProcessor, + SetupParam setupParam, + UserPackageMapper userPackageMapper, + PhoneService2 phoneService2, + CreditTrackerSettings creditTrackerSettings) { + this.packageId = setupParam.packageId(); + this.pkgPath = setupParam.xmlReqPath(); + this.xmlProcessor = xmlProcessor; + this.userPackageMapper = userPackageMapper; + this.phoneService2 = phoneService2; + this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize(); + } + + private static class PipelineStarterException extends SignalClientException { + public PipelineStarterException() { + super(null, null, null); + } + } + + @Override + public void startFillingQueue(BlockingQueue firstQueue) { + try { + log.info("Start registering subjects. PackageId: {};", packageId); + int offset = 0; + int limit = BATCH_SIZE; + UserPackage userPackage = userPackageMapper.findPackage(packageId); + List page = nextPage(limit, offset); + while (!page.isEmpty()) { + log.debug("Get page {}", page.size()); + List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() + .mapId(phoneFidMap.getId()) + .identifier(phoneFidMap.getSubjectId()) + .phone(phoneFidMap.getPhone()) + .build()).toList(); + try { + firstQueue.put(new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles)); + } catch (InterruptedException e) { + isInterrupted = true; + log.error("Pipeline starter was interrupted!"); + throw new PipelineStarterException(); + } + offset += limit; + page = nextPage(limit, offset); + } + } catch (PipelineStarterException e) { + log.info("Pipeline starter shitting down..."); + } finally { + if (isInterrupted) { + //noinspection ReturnInsideFinallyBlock + return; + } + log.debug("Send poison pill"); + try { + SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); + firstQueue.put(poison); + } catch (InterruptedException e) { + log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); + } + } + } + + protected abstract PipelineMode mode(); + protected abstract List nextPage(Integer limit, Integer offset); +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java new file mode 100644 index 0000000..2d1c80c --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineDefaultImpl.java @@ -0,0 +1,31 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.util.List; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.PhoneFidMap; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.service.indic.PhoneService2; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; + +public class KickPipeLineDefaultImpl extends KickPipeLine { + + public KickPipeLineDefaultImpl(StaxXmlProcessor xmlProcessor, + SetupParam setupParam, + UserPackageMapper userPackageMapper, + PhoneService2 phoneService2, + CreditTrackerSettings creditTrackerSettings) { + super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + } + + @Override + protected PipelineMode mode() { + return PipelineMode.DEFAULT; + } + + @Override + protected List nextPage(Integer limit, Integer offset) { + return phoneService2.pageByPackageId(packageId, limit, offset); + } + +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java deleted file mode 100644 index 4b2f6e0..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineImpl.java +++ /dev/null @@ -1,158 +0,0 @@ -package ru.nbch.credit_tracker.service.task.impl; - -import java.io.BufferedInputStream; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.util.ArrayList; -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; -import ru.nbch.credit_tracker.entities.app.PhoneFidMap; -import ru.nbch.credit_tracker.entities.app.UserPackage; -import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; -import ru.nbch.credit_tracker.exceptions.SignalClientException; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.enums.PoisonType; -import ru.nbch.credit_tracker.model.SubjectProfile; -import ru.nbch.credit_tracker.model.impl.SubjectPayload; -import ru.nbch.credit_tracker.service.indic.PhoneService2; -import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; -import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import ru.nbch.credit_tracker.utils.error.ExceptionUtils; - -public class KickPipeLineImpl implements IPipeLineStarter { - private final Logger log = LoggerFactory.getLogger(getClass()); - private final Long packageId; - private final Path pkgPath; - private final StaxXmlProcessor xmlProcessor; - private final UserPackageMapper userPackageMapper; - private final PhoneService2 phoneService2; - private final int BATCH_SIZE; - private boolean isInterrupted = false; - - - public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, - Long packageId, - Path pkgPath, - UserPackageMapper userPackageMapper, - PhoneService2 phoneService2, - CreditTrackerSettings creditTrackerSettings) { - this.packageId = packageId; - this.pkgPath = pkgPath; - this.xmlProcessor = xmlProcessor; - this.userPackageMapper = userPackageMapper; - this.phoneService2 = phoneService2; - this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize(); - } - - private static class PipelineStarterException extends SignalClientException { - public PipelineStarterException() { - super(null, null, null); - } - } - - @Override - public void startFillingQueue(BlockingQueue firstQueue) { - try { - log.info("Start registering subjects. PackageId: {};", packageId); - int offset = 0; - int limit = BATCH_SIZE; - UserPackage userPackage = userPackageMapper.findPackage(packageId); - List page = phoneService2.nextPage(packageId, limit, offset); - while (!page.isEmpty()) { - log.debug("Get page {}", page.size()); - List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() - .mapId(phoneFidMap.getId()) - .identifier(phoneFidMap.getSubjectId()) - .phone(phoneFidMap.getPhone()) - .build()).toList(); - try { - firstQueue.put(new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles)); - } catch (InterruptedException e) { - isInterrupted = true; - log.error("Pipeline starter was interrupted!"); - throw new PipelineStarterException(); - } - offset += limit; - page = phoneService2.nextPage(packageId, limit, offset); - } - } catch (PipelineStarterException e) { - log.info("Pipeline starter shitting down..."); - } finally { - if (isInterrupted) { - //noinspection ReturnInsideFinallyBlock - return; - } - log.debug("Send poison pill"); - try { - SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); - firstQueue.put(poison); - } catch (InterruptedException e) { - log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); - } - } - } -// @Override - public void startFillingQueueXmnl(BlockingQueue firstQueue) { - log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath); - UserPackage userPackage = userPackageMapper.findPackage(packageId); - AtomicInteger counter = new AtomicInteger(0); - try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) { - xmlProcessor.read(in, MonitoringRequest.class, - request -> { - if (!request.getSubjects().getSubject().isEmpty()) { - - List subjectForSave = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder() - .subjectId(subject.getId()) - .packageId(packageId) - .phone(subject.getClientId()) - .build()).toList(); - List ids = phoneService2.registerSubjects(subjectForSave); - List subjectProfiles = new ArrayList<>(); - for (int i = 0; i < ids.size(); i++) { - PhoneFidMap phoneFidMap = subjectForSave.get(i); - subjectProfiles.add(SubjectProfile.builder() - .identifier(phoneFidMap.getSubjectId()) - .phone(phoneFidMap.getPhone()) - .mapId(ids.get(i)) - .build()); - } - - int count = counter.incrementAndGet(); - log.trace("Reading subject's {} batch. Total count: {}. PackageId: {}", count, request.getSubjects().getSubject().size(), packageId); - try { - firstQueue.put(new SubjectPayload(packageId,userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles)); - } catch (InterruptedException e) { - isInterrupted = true; - log.error("Pipeline starter was interrupted!"); - throw new PipelineStarterException(); - } - request.getSubjects().getSubject().clear(); - } - }, - request -> request.getSubjects().getSubject().isEmpty(), - BATCH_SIZE, "/MonitoringRequest/Subjects/Subject" - ); - } catch (IOException e) { - log.error("{}", ExceptionUtils.getStackTrace(e)); - } catch (PipelineStarterException e) { - log.info("Pipeline starter shitting down..."); - } finally { - if (isInterrupted) { - //noinspection ReturnInsideFinallyBlock - return; - } - log.debug("Send poison pill"); - try { - SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); - firstQueue.put(poison); - } catch (InterruptedException e) { - log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); - } - } - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java new file mode 100644 index 0000000..453c65f --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickPipeLineRestoreImpl.java @@ -0,0 +1,31 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.util.List; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.PhoneFidMap; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.service.indic.PhoneService2; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; + +public class KickPipeLineRestoreImpl extends KickPipeLine { + + + public KickPipeLineRestoreImpl(StaxXmlProcessor xmlProcessor, + SetupParam setupParam, + UserPackageMapper userPackageMapper, + PhoneService2 phoneService2, + CreditTrackerSettings creditTrackerSettings) { + super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings); + } + + @Override + protected PipelineMode mode() { + return PipelineMode.RESTORE; + } + + @Override + protected List nextPage(Integer limit, Integer offset) { + return phoneService2.pageByPackageIdWithNullFids(packageId, limit, offset); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java new file mode 100644 index 0000000..91403b6 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java @@ -0,0 +1,6 @@ +package ru.nbch.credit_tracker.service.task.impl; + +public enum PipelineMode { + DEFAULT, + RESTORE +} 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 57725ae..2dad394 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 @@ -92,7 +92,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator w.writeStartElement("Auth"); diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java index 31b35c2..db468e1 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java @@ -10,6 +10,7 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; +import ru.nbch.credit_tracker.entities.app.UserPackage; @Slf4j public class CTUtil { @@ -90,4 +91,8 @@ public class CTUtil { public static String normalizePhoneNumber(String phoneNumber) { return "9" + phoneNumber; } + + public static Path workingPathToReqByPackage(String root, UserPackage userPackage) { + return Path.of(root, "tmp", userPackage.getUserId().substring(0, 6) + "_" +userPackage.getPackageName() + ".xml"); + } } diff --git a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml index 3b3c523..275886c 100644 --- a/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhoneFidMapMapper.xml @@ -29,43 +29,41 @@ INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at) values - (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}, #{phoneFid.flagDebt5000}, #{phoneFid.flag90days}, #{phoneFid.calculatedAt}) + (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}, #{phoneFid.flagDebt5000}, + #{phoneFid.flag90days}, #{phoneFid.calculatedAt}) - - - - - - - - - - - - - + + + + + + + + + + + + + - + SELECT id, + package_id, + subject_id, + phone, + fid, + flag_90_12, + flag_debt5000, + calculated_at, + error_text + FROM signals_phone_fid_map + WHERE package_id = #{packageId} and fid is null + ORDER BY id + LIMIT #{limit} OFFSET #{offset} + + + select count(*) > 0 from signals_phone_fid_map - where package_id = #{packageId} AND fid is not null + where package_id = #{packageId} + AND fid is not null