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 14:23:12 +03:00
parent bf693c3fbc
commit aaf91195b1
3 changed files with 34 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -86,7 +86,7 @@ public class PipelinePackageManager {
Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject()) Pipeline pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
.handler(saveSubjectHandler) .handler(saveSubjectHandler)
.poison(SubjectPayload.POISON) .poison(SubjectPayload.POISON)
.handlerResult(flagsEnricherV2) // .handlerResult(flagsEnricherV2)
.andThen() .andThen()
.stage(sendSignalQueueFactory.getObject()) .stage(sendSignalQueueFactory.getObject())
.handler(fidIdEnricherHandler) .handler(fidIdEnricherHandler)

2
src/main/java/ru/nbch/credit_tracker/service/task/Stage.java
View file

@ -124,10 +124,10 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
void putQuiet(BlockingQueue<Z> q, Z b) { void putQuiet(BlockingQueue<Z> q, Z b) {
try { try {
q.put(b);
if (consumer != null) { if (consumer != null) {
consumer.accept(b); consumer.accept(b);
} }
q.put(b);
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
} }

79
src/main/java/ru/nbch/credit_tracker/service/task/impl/XmlMergeCoordinator.java
View file

@ -8,7 +8,7 @@ import java.nio.charset.Charset;
import java.nio.file.Files; import java.nio.file.Files;
import java.nio.file.Path; import java.nio.file.Path;
import java.nio.file.StandardOpenOption; import java.nio.file.StandardOpenOption;
import java.util.ArrayList; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
@ -39,15 +39,31 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
this.mergedDir = Path.of(config.getLoadDir() + File.separator + "tmp_folder" + File.separator + "merged" + File.separator); this.mergedDir = Path.of(config.getLoadDir() + File.separator + "tmp_folder" + File.separator + "merged" + File.separator);
} }
@Override @Override
public void setPollingQueue(BlockingQueue<PartFilePayload> mergerQueue) { public void run() {
this.inQueue = mergerQueue; try {
Files.createDirectories(mergedDir);
} catch (IOException e) {
throw new RuntimeException(e);
}
while (!Thread.currentThread().isInterrupted()) {
try {
PartFilePayload part = inQueue.poll(5, TimeUnit.SECONDS);
if (part == null) {
continue;
}
if (part.isPoison()) {
log.debug("Got poison");
done.complete(new SignalXmlFilePayload(out));
return;
}
mergeBatch(Collections.singletonList(part));
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error("Error: {}", ExceptionUtils.getStackTrace(e));
}
} }
@Override
public CompletableFuture<SignalXmlFilePayload> doneFuture() {
return done;
} }
private void mergeBatch(List<PartFilePayload> inputs) throws Exception { private void mergeBatch(List<PartFilePayload> inputs) throws Exception {
@ -71,7 +87,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
w.close(); w.close();
} }
for (PartFilePayload p : inputs) Files.deleteIfExists(p.getPartPath()); for (PartFilePayload p : inputs) Files.deleteIfExists(p.getPartPath());
inputs.clear(); // inputs.clear();
} }
/** /**
@ -121,47 +137,12 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
} }
@Override @Override
public void run() { public void setPollingQueue(BlockingQueue<PartFilePayload> mergerQueue) {
List<PartFilePayload> batch = new ArrayList<>(batchSize); this.inQueue = mergerQueue;
try {
Files.createDirectories(mergedDir);
} catch (IOException e) {
throw new RuntimeException(e);
} }
while (!Thread.currentThread().isInterrupted()) {
try {
PartFilePayload part = inQueue.poll(5, TimeUnit.SECONDS);
if (part == null) { @Override
continue; public CompletableFuture<SignalXmlFilePayload> doneFuture() {
} return done;
if (part.isPoison()) {
done.complete(new SignalXmlFilePayload(out));
return;
}
batch.add(part);
// добираем до batchSize, но не ждём бесконечно
while (batch.size() < batchSize) {
PartFilePayload p = inQueue.poll(1, TimeUnit.SECONDS);
if (p == null) {
break;
}
if (p.isPoison()) {
mergeBatch(batch);
done.complete(new SignalXmlFilePayload(out));
return;
}
batch.add(p);
}
if (!batch.isEmpty()) {
mergeBatch(batch);
}
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error("Error: {}", ExceptionUtils.getStackTrace(e));
batch.clear();
}
}
} }
} }