Как я могу дождаться, пока весь предмет будет израсходован ⇐ JAVA

[Anonymous]

Моя задача проста: мне нужно запросить скользящий индекс elasticsearch реактивным способом.
Поскольку @Document не поддерживает имя индекса с помощью Spring EL, например @Document(index = «indexName-#(new Date().format(yyyy-MM-dd))»)
Итак, я пытаюсь вызвать elasticsearch с помощью ReactiveElasticsearchTemplate, который поддерживает изменение имени индекса во время выполнения.
Но поскольку объем данных превышает 10 000, мне нужно использовать прокрутку, чтобы повторять запрос, пока мы не получим все данные.
Я закончил первый запрос и запрос прокрутки, и он может вернуть значение.
Но мне нужно объединить все результаты, а затем вернуться.
Как я могу это сделать ? На данный момент, когда потребитель все еще работает, пустой результат возвращается во внешний интерфейс. Как я могу попросить поток подождать, пока потребитель завершит поиск elasticsearch и вернет все данные? Спасибо.

Код: Выделить всё

public Flux getByTradeDateBetween(LocalDateTime from, LocalDateTime to)
throws Exception {
List result = new ArrayList();
List total = new ArrayList();
List currentSize = new ArrayList();
List scrollId = new ArrayList();

NativeSearchQueryBuilder sourceBuilder = new NativeSearchQueryBuilder();
sourceBuilder.withQuery(
QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery(TRADE_DATE).gte(from).lte(to)));
sourceBuilder.withPageable(PageRequest.of(0, SINGLE_QUERY_SIZE));
NativeSearchQuery query = sourceBuilder.build();
elasticsearchSupport
.scrollStart(query, ELKModel.class)
.map(ELKModelWrapper::valueFrom).
subscribe(
wrapper -> {
total.add(wrapper.getTotal());l
currentSize.add(wrapper.getCurrentSize());
result.addAll(wrapper.getResults());
scrollId.add(wrapper.getScrollId());
}).dispose();

while (currentSize.size() == 1 && total.size() == 1 && currentSize.get(0) < total.get(0)) {
elasticsearchSupport
.scrollContinue(scrollId.get(0), ELKModel.class)
.map(ELKModelWrapper::valueFrom)
.subscribe(
wrapper -> {
currentSize.add(0, currentSize.get(0) + wrapper.getCurrentSize());
result.addAll(wrapper.getResults());
scrollId.add(0, wrapper.getScrollId());
}).dispose();

}

return Flux.fromIterable(result);

Подробнее здесь: https://stackoverflow.com/questions/782 ... e-consumed