Added emptyBatchCheckFunction for StaxXmlProcessor to not process empty lists of data
This commit is contained in:
parent
af2cd72647
commit
3fe63a93ec
8 changed files with 77 additions and 14 deletions
|
|
@ -181,7 +181,10 @@ public class SignalFacade {
|
||||||
throw e;
|
throw e;
|
||||||
}
|
}
|
||||||
request.getSubjects().getSubject().clear();
|
request.getSubjects().getSubject().clear();
|
||||||
}, 40000, "/MonitoringRequest/Subjects/Subject");
|
},
|
||||||
|
null
|
||||||
|
,
|
||||||
|
40000, "/MonitoringRequest/Subjects/Subject");
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
throw new RuntimeException("Error while parsing and validating monitoring request", e);
|
throw new RuntimeException("Error while parsing and validating monitoring request", e);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -152,6 +152,8 @@ public class MonitoringService {
|
||||||
writer.accept(subjectsBatch);
|
writer.accept(subjectsBatch);
|
||||||
personsData.getPersons().clear(); // чистим список для экономии памяти
|
personsData.getPersons().clear(); // чистим список для экономии памяти
|
||||||
},
|
},
|
||||||
|
pesons -> pesons.getPersons().isEmpty()
|
||||||
|
,
|
||||||
10000,
|
10000,
|
||||||
"/Report/Persons/Person");
|
"/Report/Persons/Person");
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
|
|
@ -267,6 +269,8 @@ public class MonitoringService {
|
||||||
});
|
});
|
||||||
response.getSubjects().getSubjects().clear(); // чистим список для экономии памяти
|
response.getSubjects().getSubjects().clear(); // чистим список для экономии памяти
|
||||||
},
|
},
|
||||||
|
response -> response.getSubjects().getSubjects().isEmpty()
|
||||||
|
,
|
||||||
10000,
|
10000,
|
||||||
"/MonitoringSignalResponse/Subjects/Subject");
|
"/MonitoringSignalResponse/Subjects/Subject");
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
|
|
|
||||||
|
|
@ -91,7 +91,10 @@ public class SignalService {
|
||||||
log.trace("Registered subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
|
log.trace("Registered subject's batch. Total count: {}. PackageId: {}", request.getSubjects().getSubject().size(), packageId);
|
||||||
request.getSubjects().getSubject().clear();
|
request.getSubjects().getSubject().clear();
|
||||||
}
|
}
|
||||||
}, batchSize, "/MonitoringRequest/Subjects/Subject");
|
},
|
||||||
|
request -> request.getSubjects().getSubject().isEmpty()
|
||||||
|
,
|
||||||
|
batchSize, "/MonitoringRequest/Subjects/Subject");
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
throw new RuntimeException("Error while parsing monitoring request", e);
|
throw new RuntimeException("Error while parsing monitoring request", e);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -7,6 +7,7 @@ import java.util.ArrayDeque;
|
||||||
import java.util.Deque;
|
import java.util.Deque;
|
||||||
import java.util.Iterator;
|
import java.util.Iterator;
|
||||||
import java.util.function.Consumer;
|
import java.util.function.Consumer;
|
||||||
|
import java.util.function.Function;
|
||||||
|
|
||||||
public abstract class BaseStaxReader<T> implements StaxReader<T> {
|
public abstract class BaseStaxReader<T> implements StaxReader<T> {
|
||||||
private final Deque<String> stack = new ArrayDeque<>();
|
private final Deque<String> stack = new ArrayDeque<>();
|
||||||
|
|
@ -16,10 +17,10 @@ public abstract class BaseStaxReader<T> implements StaxReader<T> {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public T read(XMLStreamReader reader) throws XMLStreamException {
|
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<T> objProcessor, Integer batchSize, String endBatchElementPath) throws XMLStreamException {
|
public T read(XMLStreamReader reader, Consumer<T> objProcessor, Function<T, Boolean> emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath) throws XMLStreamException {
|
||||||
T obj = createObject();
|
T obj = createObject();
|
||||||
int objCount = 0;
|
int objCount = 0;
|
||||||
try {
|
try {
|
||||||
|
|
@ -47,7 +48,7 @@ public abstract class BaseStaxReader<T> implements StaxReader<T> {
|
||||||
if (endBatchElementPath.equals(lastElementPath)) {
|
if (endBatchElementPath.equals(lastElementPath)) {
|
||||||
objCount++;
|
objCount++;
|
||||||
if (objCount == batchSize) {
|
if (objCount == batchSize) {
|
||||||
objProcessor.accept(obj);
|
objProcessor.accept(obj); // обрабатываем пачку элементов
|
||||||
objCount = 0;
|
objCount = 0;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -59,7 +60,11 @@ public abstract class BaseStaxReader<T> implements StaxReader<T> {
|
||||||
if (breakExecution) break;
|
if (breakExecution) break;
|
||||||
}
|
}
|
||||||
if (objProcessor != null) {
|
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;
|
return obj;
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
|
|
|
||||||
|
|
@ -3,9 +3,10 @@ package ru.nbch.credit_tracker.service.xml.stax;
|
||||||
import javax.xml.stream.XMLStreamException;
|
import javax.xml.stream.XMLStreamException;
|
||||||
import javax.xml.stream.XMLStreamReader;
|
import javax.xml.stream.XMLStreamReader;
|
||||||
import java.util.function.Consumer;
|
import java.util.function.Consumer;
|
||||||
|
import java.util.function.Function;
|
||||||
|
|
||||||
public interface StaxReader<T> {
|
public interface StaxReader<T> {
|
||||||
Class<T> type();
|
Class<T> type();
|
||||||
T read(XMLStreamReader xmlStreamReader) throws XMLStreamException;
|
T read(XMLStreamReader xmlStreamReader) throws XMLStreamException;
|
||||||
T read(XMLStreamReader reader, Consumer<T> objProcessor, Integer batchSize, String endBatchElementPath) throws XMLStreamException;
|
T read(XMLStreamReader reader, Consumer<T> objProcessor, Function<T, Boolean> emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath) throws XMLStreamException;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -2,16 +2,20 @@ package ru.nbch.credit_tracker.service.xml.stax;
|
||||||
|
|
||||||
import java.io.InputStream;
|
import java.io.InputStream;
|
||||||
import java.util.function.Consumer;
|
import java.util.function.Consumer;
|
||||||
|
import java.util.function.Function;
|
||||||
|
|
||||||
public interface StaxXmlProcessor {
|
public interface StaxXmlProcessor {
|
||||||
<Z> Z read(String xml, Class<Z> clazz);
|
<Z> Z read(String xml, Class<Z> clazz);
|
||||||
|
|
||||||
<Z> Z read(byte[] xml, Class<Z> clazz);
|
<Z> Z read(byte[] xml, Class<Z> clazz);
|
||||||
|
|
||||||
<Z> Z read(InputStream inputStream, Class<Z> clazz);
|
<Z> Z read(InputStream inputStream, Class<Z> clazz);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param objProcessor: Консьюмер, который срабатывает, когда считается batchSize элементов endBatchElementPath
|
* @param objProcessor: Консьюмер, который срабатывает, когда считается batchSize элементов endBatchElementPath
|
||||||
* @param batchSize: количество считываемых элементов в пачке
|
* @param emptyBatchCheckFunction: Функция проверки на пустоту списка элементов. Пустые списки не будут обработаны консьюмером
|
||||||
* @param endBatchElementPath: localPath считываемых в пачку элементов. Например: "/Report/Subjects/Subject"
|
* @param batchSize: количество считываемых элементов в пачке
|
||||||
|
* @param endBatchElementPath: localPath считываемых в пачку элементов. Например: "/Report/Subjects/Subject"
|
||||||
* <Report>
|
* <Report>
|
||||||
* <Subjects>
|
* <Subjects>
|
||||||
* <Subject></Subject>
|
* <Subject></Subject>
|
||||||
|
|
@ -19,5 +23,5 @@ public interface StaxXmlProcessor {
|
||||||
* </Subjects>
|
* </Subjects>
|
||||||
* </Report>
|
* </Report>
|
||||||
*/
|
*/
|
||||||
<Z> Z read(InputStream inputStream, Class<Z> clazz, Consumer<Z> objProcessor, Integer batchSize, String endBatchElementPath);
|
<Z> Z read(InputStream inputStream, Class<Z> clazz, Consumer<Z> objProcessor, Function<Z, Boolean> emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath);
|
||||||
}
|
}
|
||||||
|
|
@ -16,6 +16,7 @@ import java.util.HashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.function.Consumer;
|
import java.util.function.Consumer;
|
||||||
|
import java.util.function.Function;
|
||||||
|
|
||||||
public class XmlProcessorStaxImpl implements StaxXmlProcessor {
|
public class XmlProcessorStaxImpl implements StaxXmlProcessor {
|
||||||
private final Logger log = LoggerFactory.getLogger(getClass());
|
private final Logger log = LoggerFactory.getLogger(getClass());
|
||||||
|
|
@ -96,13 +97,13 @@ public class XmlProcessorStaxImpl implements StaxXmlProcessor {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public <Z> Z read(InputStream inputStream, Class<Z> clazz, Consumer<Z> objProcessor, Integer batchSize, String endBatchElementPath) {
|
public <Z> Z read(InputStream inputStream, Class<Z> clazz, Consumer<Z> objProcessor, Function<Z, Boolean> emptyBatchCheckFunction, Integer batchSize, String endBatchElementPath) {
|
||||||
try {
|
try {
|
||||||
XMLStreamReader xmlStreamReader = xmlInputFactory.createXMLStreamReader(inputStream);
|
XMLStreamReader xmlStreamReader = xmlInputFactory.createXMLStreamReader(inputStream);
|
||||||
@SuppressWarnings("unchecked")
|
@SuppressWarnings("unchecked")
|
||||||
StaxReader<Z> staxProcessor = (StaxReader<Z>) staxReaders.get(clazz);
|
StaxReader<Z> staxProcessor = (StaxReader<Z>) staxReaders.get(clazz);
|
||||||
if (staxProcessor != null) {
|
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);
|
throw new IllegalStateException("no stax reader registered for " + clazz);
|
||||||
} catch (SignalClientException e) {
|
} catch (SignalClientException e) {
|
||||||
|
|
|
||||||
|
|
@ -89,10 +89,12 @@ public class StaxTest {
|
||||||
List<Person> firstPersonList = new ArrayList<>();
|
List<Person> firstPersonList = new ArrayList<>();
|
||||||
staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class,
|
staxXmlProcessor.read(getClass().getResourceAsStream("/monitoring_request/xml/nbki_monitoring_report.xml"), Persons.class,
|
||||||
personsData -> {
|
personsData -> {
|
||||||
Assertions.assertTrue(personsData.getPersons().size() == 1 || personsData.getPersons().isEmpty());
|
Assertions.assertEquals(1, personsData.getPersons().size());
|
||||||
firstPersonList.addAll(personsData.getPersons());
|
firstPersonList.addAll(personsData.getPersons());
|
||||||
personsData.getPersons().clear();
|
personsData.getPersons().clear();
|
||||||
},
|
},
|
||||||
|
obj -> obj.getPersons().isEmpty()
|
||||||
|
,
|
||||||
1,
|
1,
|
||||||
"/Report/Persons/Person");
|
"/Report/Persons/Person");
|
||||||
|
|
||||||
|
|
@ -102,9 +104,49 @@ public class StaxTest {
|
||||||
secondPersonList.addAll(personsData.getPersons());
|
secondPersonList.addAll(personsData.getPersons());
|
||||||
personsData.getPersons().clear();
|
personsData.getPersons().clear();
|
||||||
},
|
},
|
||||||
|
obj -> obj.getPersons().isEmpty(),
|
||||||
3,
|
3,
|
||||||
"/Report/Persons/Person");
|
"/Report/Persons/Person");
|
||||||
|
|
||||||
Assertions.assertEquals(firstPersonList.size(), secondPersonList.size());
|
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");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue