flag calc

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

@@ -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());
+                    merge(batch);
-                    mergeBatch(batch);
-//                     удаляем исходники после успешного мерджа
-                    for (PartFilePayload p : batch) Files.deleteIfExists(p.getPartPath());
-                    batch.clear();
                 }
             } 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();
+    }
 }