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

flag calc

Browse source
This commit is contained in:
etreschenkov 2025-11-11 16:47:58 +03:00
parent 53a2ab77a9
commit 395613ae01
2 changed files with 6 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -23,9 +23,4 @@ public class StageConfig {
return new StageQueueFactory<>(10000);
}
@Bean("flagQueueFactory")
public StageQueueFactory<SubjectPayload> flagQueueFactory() {
return new StageQueueFactory<>(100);
}
}

27
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java
View file

@ -8,7 +8,6 @@ import java.util.function.BiFunction;
import java.util.function.Supplier;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
import ru.nbch.credit_tracker.config.task.StageQueueFactory;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
@ -16,62 +15,48 @@ import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
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.FlagsEnricher;
import ru.nbch.credit_tracker.service.task.impl.FlagsEnricherV2;
import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler;
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;
import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
@Component
@Slf4j
public class PipelinePackageManager {
private final ApplicationContext applicationContext;
private final StaxXmlProcessor xmlProcessor;
private final SignalRestService signalRestService;
private final FlagsEnricherV2 flagsEnricherV2;
private final StageQueueFactory<SubjectPayload> saveSubjectQueueFactory;
private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
private final StageQueueFactory<SubjectPayload> flagQueueFactory;
private final SaveSubjectHandler saveSubjectHandler;
private final FidIdEnricher fidIdEnricherHandler;
private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
private final Supplier<FlagsEnricher> flagEnricherFactory;
private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
private final FlagsEnricherV2 flagsEnricherV2;
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 Semaphore semaphore;
private final static int BATCH_SIZE = 10_000;
private final static int MAX_PARALLEL_REQUESTS = 3;
public PipelinePackageManager(ApplicationContext applicationContext,
StaxXmlProcessor xmlProcessor,
SignalRestService signalRestService,
public PipelinePackageManager(SignalRestService signalRestService,
@Qualifier("saveSubjectQueueFactory") StageQueueFactory<SubjectPayload> saveSubjectQueueFactory,
@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
@Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
@Qualifier("flagQueueFactory") StageQueueFactory<SubjectPayload> flagQueueFactory,
SaveSubjectHandler saveSubjectHandler, FidIdEnricher fidIdEnricherHandler,
Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
Supplier<FlagsEnricher> flagEnricherFactory,
BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory, FlagsEnricherV2 flagsEnricherV2) {
this.applicationContext = applicationContext;
this.xmlProcessor = xmlProcessor;
BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
FlagsEnricherV2 flagsEnricherV2) {
this.signalRestService = signalRestService;
this.saveSubjectQueueFactory = saveSubjectQueueFactory;
this.idEnricherQueueFactory = idEnricherQueueFactory;
this.sendSignalQueueFactory = sendSignalQueueFactory;
this.flagQueueFactory = flagQueueFactory;
this.saveSubjectHandler = saveSubjectHandler;
this.fidIdEnricherHandler = fidIdEnricherHandler;
this.pipelineTerminatorFactory = pipelineTerminatorFactory;
this.flagEnricherFactory = flagEnricherFactory;
this.pipelineStarterFactory = pipelineStarterFactory;
this.flagsEnricherV2 = flagsEnricherV2;
this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS);
@ -86,7 +71,7 @@ public class PipelinePackageManager {
Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
.handler(saveSubjectHandler)
.poison(SubjectPayload.POISON)
// .handlerResult(flagsEnricherV2)
.handlerResult(flagsEnricherV2)
.andThen()
.stage(sendSignalQueueFactory.getObject())
.handler(fidIdEnricherHandler)
@ -118,7 +103,7 @@ public class PipelinePackageManager {
pipeline.getStages().forEach(Stage::close);
semaphore.release();
}
});
});
} catch (Exception e) {
semaphore.release();
throw e;