diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java index 6ed3264..026aae4 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -181,7 +181,10 @@ public class SignalFacade { throw e; } request.getSubjects().getSubject().clear(); - }, 40000, "/MonitoringRequest/Subjects/Subject"); + }, + null + , + 40000, "/MonitoringRequest/Subjects/Subject"); } catch (IOException e) { throw new RuntimeException("Error while parsing and validating monitoring request", e); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java index f9fa4e3..763f4b2 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java @@ -152,6 +152,8 @@ public class MonitoringService { writer.accept(subjectsBatch); personsData.getPersons().clear(); // чистим список для экономии памяти }, + pesons -> pesons.getPersons().isEmpty() + , 10000, "/Report/Persons/Person"); } catch (IOException e) { @@ -267,6 +269,8 @@ public class MonitoringService { }); response.getSubjects().getSubjects().clear(); // чистим список для экономии памяти }, + response -> response.getSubjects().getSubjects().isEmpty() + , 10000, "/MonitoringSignalResponse/Subjects/Subject"); } catch (IOException e) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java index daa2356..745c53b 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -91,7 +91,10 @@ public class SignalService { log.trace("Registered subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId); request.getSubjects().getSubject().clear(); } - }, batchSize, "/MonitoringRequest/Subjects/Subject"); + }, + request -> request.getSubjects().getSubject().isEmpty() + , + batchSize, "/MonitoringRequest/Subjects/Subject"); } catch (IOException e) { throw new RuntimeException("Error while parsing monitoring request", e); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/BaseStaxReader.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/BaseStaxReader.java index 9365668..c00d6b2 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/BaseStaxReader.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/BaseStaxReader.java @@ -7,6 +7,7 @@ import java.util.ArrayDeque; import java.util.Deque; import java.util.Iterator; import java.util.function.Consumer; +import java.util.function.Function; public abstract class BaseStaxReader implements StaxReader { private final Deque stack = new ArrayDeque<>(); @@ -16,10 +17,10 @@ public abstract class BaseStaxReader implements StaxReader { @Override public T read(XMLStreamReader reader) throws XMLStreamException { - return read(reader, null, null, null); + return read(reader, null, null, null, null); } - public T read(XMLStreamReader reader, Consumer objProcessor, Integer batchSize, String endBatchElementPath) throws XMLStreamException { + public T read(XMLStreamReader reader, Consumer objProcessor, Function emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath) throws XMLStreamException { T obj = createObject(); int objCount = 0; try { @@ -47,7 +48,7 @@ public abstract class BaseStaxReader implements StaxReader { if (endBatchElementPath.equals(lastElementPath)) { objCount++; if (objCount == batchSize) { - objProcessor.accept(obj); + objProcessor.accept(obj); // обрабатываем пачку элементов objCount = 0; } } @@ -59,7 +60,11 @@ public abstract class BaseStaxReader implements StaxReader { if (breakExecution) break; } if (objProcessor != null) { - objProcessor.accept(obj); + if (emptyBatchCheckFunction == null) { + objProcessor.accept(obj); // Дополнительная обработка пачки если: batchSize меньше количества элементов в xml или дообрабатываем "хвост" меньший чем batchSize + } else if (!emptyBatchCheckFunction.apply(obj)) { // Проверка на пустоту пачки, чтобы не обрабатывать пустой список + objProcessor.accept(obj); + } } return obj; } catch (Exception e) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxReader.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxReader.java index 74b944d..34edb26 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxReader.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxReader.java @@ -3,9 +3,10 @@ package ru.nbch.credit_tracker.service.xml.stax; import javax.xml.stream.XMLStreamException; import javax.xml.stream.XMLStreamReader; import java.util.function.Consumer; +import java.util.function.Function; public interface StaxReader { Class type(); T read(XMLStreamReader xmlStreamReader) throws XMLStreamException; - T read(XMLStreamReader reader, Consumer objProcessor, Integer batchSize, String endBatchElementPath) throws XMLStreamException; + T read(XMLStreamReader reader, Consumer objProcessor, Function emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath) throws XMLStreamException; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlProcessor.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlProcessor.java index a340ec7..665efe4 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlProcessor.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/StaxXmlProcessor.java @@ -2,16 +2,20 @@ package ru.nbch.credit_tracker.service.xml.stax; import java.io.InputStream; import java.util.function.Consumer; +import java.util.function.Function; public interface StaxXmlProcessor { Z read(String xml, Class clazz); + Z read(byte[] xml, Class clazz); + Z read(InputStream inputStream, Class clazz); /** - * @param objProcessor: Консьюмер, который срабатывает, когда считается batchSize элементов endBatchElementPath - * @param batchSize: количество считываемых элементов в пачке - * @param endBatchElementPath: localPath считываемых в пачку элементов. Например: "/Report/Subjects/Subject" + * @param objProcessor: Консьюмер, который срабатывает, когда считается batchSize элементов endBatchElementPath + * @param emptyBatchCheckFunction: Функция проверки на пустоту списка элементов. Пустые списки не будут обработаны консьюмером + * @param batchSize: количество считываемых элементов в пачке + * @param endBatchElementPath: localPath считываемых в пачку элементов. Например: "/Report/Subjects/Subject" * * * @@ -19,5 +23,5 @@ public interface StaxXmlProcessor { * * */ - Z read(InputStream inputStream, Class clazz, Consumer objProcessor, Integer batchSize, String endBatchElementPath); + Z read(InputStream inputStream, Class clazz, Consumer objProcessor, Function emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath); } \ No newline at end of file diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/XmlProcessorStaxImpl.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/XmlProcessorStaxImpl.java index d9f1c68..3088443 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/XmlProcessorStaxImpl.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/XmlProcessorStaxImpl.java @@ -16,6 +16,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.function.Consumer; +import java.util.function.Function; public class XmlProcessorStaxImpl implements StaxXmlProcessor { private final Logger log = LoggerFactory.getLogger(getClass()); @@ -96,13 +97,13 @@ public class XmlProcessorStaxImpl implements StaxXmlProcessor { } @Override - public Z read(InputStream inputStream, Class clazz, Consumer objProcessor, Integer batchSize, String endBatchElementPath) { + public Z read(InputStream inputStream, Class clazz, Consumer objProcessor, Function emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath) { try { XMLStreamReader xmlStreamReader = xmlInputFactory.createXMLStreamReader(inputStream); @SuppressWarnings("unchecked") StaxReader staxProcessor = (StaxReader) staxReaders.get(clazz); if (staxProcessor != null) { - return staxProcessor.read(xmlStreamReader, objProcessor, batchSize, endBatchElementPath); + return staxProcessor.read(xmlStreamReader, objProcessor, emptyBatchCheckFunction, batchSize, endBatchElementPath); } throw new IllegalStateException("no stax reader registered for " + clazz); } catch (SignalClientException e) { diff --git a/src/test/java/ru/nbch/credit_tracker/stax/StaxTest.java b/src/test/java/ru/nbch/credit_tracker/stax/StaxTest.java index 6a27d08..eb9e3cc 100644 --- a/src/test/java/ru/nbch/credit_tracker/stax/StaxTest.java +++ b/src/test/java/ru/nbch/credit_tracker/stax/StaxTest.java @@ -89,10 +89,12 @@ public class StaxTest { List firstPersonList = new ArrayList<>(); staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class, personsData -> { - Assertions.assertTrue(personsData.getPersons().size() == 1 || personsData.getPersons().isEmpty()); + Assertions.assertEquals(1, personsData.getPersons().size()); firstPersonList.addAll(personsData.getPersons()); personsData.getPersons().clear(); }, + obj -> obj.getPersons().isEmpty() + , 1, "/Report/Persons/Person"); @@ -102,9 +104,49 @@ public class StaxTest { secondPersonList.addAll(personsData.getPersons()); personsData.getPersons().clear(); }, + obj -> obj.getPersons().isEmpty(), 3, "/Report/Persons/Person"); Assertions.assertEquals(firstPersonList.size(), secondPersonList.size()); } + + @Test + void testParseNbkiMonitoringReportWithBounds() { + staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class, + personsData -> { + Assertions.assertEquals(5, personsData.getPersons().size()); + personsData.getPersons().clear(); + }, + obj -> obj.getPersons().isEmpty(), + 5, + "/Report/Persons/Person"); + + staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class, + personsData -> { + Assertions.assertFalse(personsData.getPersons().isEmpty()); + personsData.getPersons().clear(); + }, + obj -> obj.getPersons().isEmpty(), + 6, + "/Report/Persons/Person"); + + staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class, + personsData -> { + Assertions.assertEquals(1, personsData.getPersons().size()); + personsData.getPersons().clear(); + }, + obj -> obj.getPersons().isEmpty(), + 1, + "/Report/Persons/Person"); + + staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class, + personsData -> { + Assertions.assertTrue(personsData.getPersons().size() == 4 || personsData.getPersons().size() == 1); + personsData.getPersons().clear(); + }, + obj -> obj.getPersons().isEmpty(), + 4, + "/Report/Persons/Person"); + } }