diff --git a/credit-tracker.properties b/credit-tracker.properties index ca9ec6c..982f795 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -69,4 +69,9 @@ ct.monitoring-interval=${MONITORING_INTERVAL:15} ct.update-flags-interval=${UPDATE_FLAGS_INTERVAL:1} ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30} -ct.encoding=windows-1251 \ No newline at end of file +ct.encoding=windows-1251 + +#Batch size and global pipeline timeout +ct.batchSize=1000 +#in minute +ct.timeout=30 \ No newline at end of file diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java index 9a59f0b..31c8291 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java @@ -33,4 +33,7 @@ public class CreditTrackerSettings { private Long daysToUpdateFlags; private String encoding; + + private Integer batchSize; + private Integer timeout; } 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 ab39072..c8d727e 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 @@ -20,8 +20,8 @@ import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; @Configuration public class PipelineStarterConfig { @Bean - public BiFunction> pipelineStarterFactory(StaxXmlProcessor xml) { - return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName); + public BiFunction> pipelineStarterFactory(StaxXmlProcessor xml, CreditTrackerSettings creditTrackerSettings) { + return (packageId, packageName) -> new KickPipeLineImpl(xml, packageId, packageName, creditTrackerSettings); } @Bean 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 625fced..e774c1b 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 @@ -9,6 +9,7 @@ import java.util.function.Supplier; import lombok.extern.slf4j.Slf4j; 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.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; @@ -34,6 +35,7 @@ public class PipelinePackageManager { private final FidIdEnricher fidIdEnricherHandler; private final Supplier> pipelineTerminatorFactory; private final BiFunction> 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())); @@ -49,7 +51,8 @@ public class PipelinePackageManager { SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler, Supplier> pipelineTerminatorFactory, BiFunction> pipelineStarterFactory, - FlagsEnricherV2 flagsEnricherV2) { + FlagsEnricherV2 flagsEnricherV2, + CreditTrackerSettings creditTrackerSettings) { this.signalRestService = signalRestService; this.saveSubjectQueueFactory = saveSubjectQueueFactory; this.idEnricherQueueFactory = idEnricherQueueFactory; @@ -59,6 +62,7 @@ public class PipelinePackageManager { this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.pipelineStarterFactory = pipelineStarterFactory; this.flagsEnricherV2 = flagsEnricherV2; + this.creditTrackerSettings = creditTrackerSettings; this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS); } @@ -86,7 +90,7 @@ public class PipelinePackageManager { pipeline.getTerminator() .doneFuture() - .orTimeout(5, TimeUnit.MINUTES) + .orTimeout(creditTrackerSettings.getTimeout(), TimeUnit.MINUTES) .whenComplete((v, e) -> { try { if (e != null) { 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 index 9a89517..bc9cace 100644 --- 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 @@ -10,6 +10,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; import ru.nbch.credit_tracker.model.SubjectProfile; import ru.nbch.credit_tracker.model.impl.SubjectPayload; @@ -23,13 +24,15 @@ public class KickPipeLineImpl implements IPipeLineStarter { private final String pkgPath; private final StaxXmlProcessor xmlProcessor; - private final static int BATCH_SIZE = 10_000; + private final int BATCH_SIZE; - public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, String pkgPath) { + public KickPipeLineImpl(StaxXmlProcessor xmlProcessor, Long packageId, String pkgPath, + CreditTrackerSettings creditTrackerSettings) { this.packageId = packageId; this.pkgPath = pkgPath; this.xmlProcessor = xmlProcessor; + this.BATCH_SIZE = creditTrackerSettings.getBatchSize(); } @Override