@Test void scrollIterableShouldWorkWhenMoreThanSizeElement() { String id1 = "1"; client.index(new IndexRequest(INDEX_NAME.getValue()) .type(NodeMappingFactory.DEFAULT_MAPPING_NAME) .id(id1) .source(MESSAGE, "Sample message"), RequestOptions.DEFAULT) .block(); String id2 = "2"; client.index(new IndexRequest(INDEX_NAME.getValue()) .type(NodeMappingFactory.DEFAULT_MAPPING_NAME) .id(id2) .source(MESSAGE, "Sample message"), RequestOptions.DEFAULT) .block(); String id3 = "3"; client.index(new IndexRequest(INDEX_NAME.getValue()) .type(NodeMappingFactory.DEFAULT_MAPPING_NAME) .id(id3) .source(MESSAGE, "Sample message"), RequestOptions.DEFAULT) .block(); elasticSearch.awaitForElasticSearch(); WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2, id3)); SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue()) .scroll(TIMEOUT) .source(new SearchSourceBuilder() .query(QueryBuilders.matchAllQuery()) .size(SIZE)); assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block()) .extracting(SearchHit::getId) .containsOnly(id1, id2, id3); }
public Flux<SearchHit> searchHits() { return searchResponses() .concatMap(searchResponse -> Flux.just(searchResponse.getHits().getHits())); }
ScrolledSearch { public Flux<SearchHit> searchHits() { return searchResponses() .concatMap(searchResponse -> Flux.just(searchResponse.getHits().getHits())); } }
ScrolledSearch { public Flux<SearchHit> searchHits() { return searchResponses() .concatMap(searchResponse -> Flux.just(searchResponse.getHits().getHits())); } ScrolledSearch(ReactorElasticSearchClient client, SearchRequest searchRequest); }
ScrolledSearch { public Flux<SearchHit> searchHits() { return searchResponses() .concatMap(searchResponse -> Flux.just(searchResponse.getHits().getHits())); } ScrolledSearch(ReactorElasticSearchClient client, SearchRequest searchRequest); Flux<SearchHit> searchHits(); Flux<SearchResponse> searchResponses(); void close(AtomicReference<Optional<String>> scrollId); }
ScrolledSearch { public Flux<SearchHit> searchHits() { return searchResponses() .concatMap(searchResponse -> Flux.just(searchResponse.getHits().getHits())); } ScrolledSearch(ReactorElasticSearchClient client, SearchRequest searchRequest); Flux<SearchHit> searchHits(); Flux<SearchResponse> searchResponses(); void close(AtomicReference<Optional<String>> scrollId);  }