From 0f04ab3e42cea94f476db435e4ab917266a5c643 Mon Sep 17 00:00:00 2001
From: etreschenkov <ETreschenkov@mfd.ru>
Date: Tue, 11 Nov 2025 13:01:00 +0300
Subject: [PATCH] flag calc

---
 .../task/impl/XmlMergeCoordinator.java        | 20 +++++++++++--------
 1 file changed, 12 insertions(+), 8 deletions(-)

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 f035f0e..232d6c0 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
@@ -35,7 +35,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
     private Path out;
     private BlockingQueue<PartFilePayload> inQueue;
 
-    public XmlMergeCoordinator(CreditTrackerSettings config)  {
+    public XmlMergeCoordinator(CreditTrackerSettings config) {
         this.mergedDir = Path.of(config.getLoadDir() + File.separator + "tmp_folder" + File.separator + "merged" + File.separator);
     }
 
@@ -138,24 +138,20 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
                 }
                 batch.add(part);
                 // добираем до batchSize, но не ждём бесконечно
-                //todo подозрительное действие, прочекать не будет ли мешать poison
                 while (batch.size() < batchSize) {
-                    PartFilePayload p = inQueue.poll(1000, TimeUnit.MILLISECONDS);
+                    PartFilePayload p = inQueue.poll(1, TimeUnit.SECONDS);
                     if (p == null) {
                         break;
                     }
                     if (p.isPoison()) {
+                        merge(batch);
                         done.complete(new SignalXmlFilePayload(out));
                         return;
                     }
                     batch.add(p);
                 }
                 if (!batch.isEmpty()) {
-                    log.debug("Starting merge batch; size: {}", batch.size());
-                    mergeBatch(batch);
-//                     удаляем исходники после успешного мерджа
-                    for (PartFilePayload p : batch) Files.deleteIfExists(p.getPartPath());
-                    batch.clear();
+                    merge(batch);
                 }
             } catch (InterruptedException ie) {
                 Thread.currentThread().interrupt();
@@ -165,4 +161,12 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
             }
         }
     }
+
+    private void merge(List<PartFilePayload> inputs) throws Exception {
+        log.debug("Starting merge batch; size: {}", inputs.size());
+        mergeBatch(inputs);
+        //удаляем исходники после успешного мерджа
+        for (PartFilePayload p : inputs) Files.deleteIfExists(p.getPartPath());
+        inputs.clear();
+    }
 }