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

Моя задача проста: мне нужно запросить скользящий индекс elasticsearch реактивным способом.

Поскольку @Document не поддерживает имя индекса с помощью Spring EL, например @Document(index = "indexName-#(new Date().format(yyyy-MM-dd))")

Итак, я пытаюсь вызвать elasticsearch с помощью ReactiveElasticsearchTemplate, который поддерживает изменение имени индекса во время выполнения.

Но поскольку объем данных превышает 10 000, мне нужно использовать прокрутку, чтобы повторять запрос, пока мы не получим все данные.

Я завершил первый запрос и запрос прокрутки, и он может вернуть значение.

Но мне нужно объединить все результаты и затем вернуться.

Как мне это сделать? На данный момент, когда потребитель все еще работает, пустой результат возвращается во внешний интерфейс. Как я могу попросить поток подождать, пока потребитель завершит поиск elasticsearch и вернет все данные? Спасибо.

public Flux<ELKModel> getByTradeDateBetween(LocalDateTime from, LocalDateTime to)
  throws Exception {
List<ELKModel> result = new ArrayList<ELKModel>();
List<Long> total = new ArrayList<>();
List<Long> currentSize = new ArrayList<>();
List<String> 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);

}