DATAES-866 - Implement suggest query in reactive client.

src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java

@@ -96,6 +96,7 @@
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.SearchHits;
 import org.elasticsearch.search.aggregations.Aggregation;
+import org.elasticsearch.search.suggest.Suggest;
 import org.reactivestreams.Publisher;
 
 import org.springframework.data.elasticsearch.client.ClientConfiguration;
@@ -132,6 +133,7 @@
  * @author Henrique Amaral
  * @author Roman Puchkovskiy
  * @author Russell Parry
+ * @author Thomas Geese
  * @since 3.2
  * @see ClientConfiguration
  * @see ReactiveRestClients
@@ -886,6 +888,12 @@ private static ElasticsearchException getElasticsearchException(String content,
 		}
 	}
 
+	@Override
+	public Flux<Suggest> suggest(HttpHeaders headers, SearchRequest searchRequest) {
+		return sendRequest(searchRequest, requestCreator.search(), SearchResponse.class, headers) //
+				.map(SearchResponse::getSuggest);
+	}
+
 	private static void buildExceptionMessages(StringBuilder sb, Throwable t) {
 
 		sb.append(t.getMessage());

src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java

@@ -52,6 +52,7 @@
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.aggregations.Aggregation;
+import org.elasticsearch.search.suggest.Suggest;
 import org.springframework.data.elasticsearch.client.ClientConfiguration;
 import org.springframework.data.elasticsearch.client.ElasticsearchHost;
 import org.springframework.http.HttpHeaders;
@@ -67,6 +68,7 @@
  * @author Mark Paluch
  * @author Peter-Josef Meisch
  * @author Henrique Amaral
+ * @author Thomas Geese
  * @since 3.2
  * @see ClientConfiguration
  * @see ReactiveRestClients
@@ -417,6 +419,27 @@ default Flux<SearchHit> search(SearchRequest searchRequest) {
 	 */
 	Flux<SearchHit> search(HttpHeaders headers, SearchRequest searchRequest);
 
+	/**
+	 * Execute the given {@link SearchRequest} against the {@literal search} API.
+	 *
+	 * @param searchRequest must not be {@literal null}.
+	 * @return the {@link Flux} emitting {@link Suggest suggestions} one by one.
+	 * @since 4.1
+	 */
+	default Flux<Suggest> suggest(SearchRequest searchRequest) {
+		return suggest(HttpHeaders.EMPTY, searchRequest);
+	}
+
+	/**
+	 * Execute the given {@link SearchRequest} against the {@literal search} API.
+	 *
+	 * @param headers Use {@link HttpHeaders} to provide eg. authentication data. Must not be {@literal null}.
+	 * @param searchRequest must not be {@literal null}.
+	 * @return the {@link Flux} emitting {@link Suggest suggestions} one by one.
+	 * @since 4.1
+	 */
+	Flux<Suggest> suggest(HttpHeaders headers, SearchRequest searchRequest);
+
 	/**
 	 * Execute the given {@link SearchRequest} with aggregations against the {@literal search} API.
 	 *

src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java

@@ -41,6 +41,8 @@
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 import org.elasticsearch.search.aggregations.Aggregation;
+import org.elasticsearch.search.suggest.Suggest;
+import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -90,6 +92,7 @@
  * @author Aleksei Arsenev
  * @author Roman Puchkovskiy
  * @author Russell Parry
+ * @author Thomas Geese
  * @since 3.2
  */
 public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOperations, ApplicationContextAware {
@@ -640,6 +643,23 @@ public Flux<Aggregation> aggregate(Query query, Class<?> entityType, IndexCoordi
 		return doAggregate(query, entityType, index);
 	}
 
+	@Override
+	public Flux<Suggest> suggest(SuggestBuilder suggestion, Class<?> entityType) {
+		return suggest(suggestion, getIndexCoordinatesFor(entityType));
+	}
+
+	@Override
+	public Flux<Suggest> suggest(SuggestBuilder suggestion, IndexCoordinates index) {
+		return doSuggest(suggestion, index);
+	}
+
+	private Flux<Suggest> doSuggest(SuggestBuilder suggestion, IndexCoordinates index) {
+		return Flux.defer(() -> {
+			SearchRequest request = requestFactory.searchRequest(suggestion, index);
+			return Flux.from(execute(client -> client.suggest(request)));
+		});
+	}
+
 	private Flux<Aggregation> doAggregate(Query query, Class<?> entityType, IndexCoordinates index) {
 		return Flux.defer(() -> {
 			SearchRequest request = requestFactory.searchRequest(query, entityType, index);

src/main/java/org/springframework/data/elasticsearch/core/ReactiveSearchOperations.java

@@ -20,6 +20,8 @@
 
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.aggregations.Aggregation;
+import org.elasticsearch.search.suggest.Suggest;
+import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.Query;
@@ -32,6 +34,7 @@
  *
  * @author Peter-Josef Meisch
  * @author Russell Parry
+ * @author Thomas Geese
  * @since 4.0
  */
 public interface ReactiveSearchOperations {
@@ -206,4 +209,22 @@ default <T> Flux<SearchHit<T>> search(Query query, Class<T> entityType, IndexCoo
 	 * @since 4.0
 	 */
 	Flux<Aggregation> aggregate(Query query, Class<?> entityType, IndexCoordinates index);
+
+	/**
+	 * Does a suggest query
+	 *
+	 * @param suggestion the query
+	 * @param entityType must not be {@literal null}.
+	 * @return the suggest response
+	 */
+	Flux<Suggest> suggest(SuggestBuilder suggestion, Class<?> entityType);
+
+	/**
+	 * Does a suggest query
+	 *
+	 * @param suggestion the query
+	 * @param index the index to run the query against
+	 * @return the suggest response
+	 */
+	Flux<Suggest> suggest(SuggestBuilder suggestion, IndexCoordinates index);
 }

src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientTests.java

@@ -53,6 +53,8 @@
 import org.elasticsearch.search.aggregations.AggregationBuilders;
 import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.suggest.SuggestBuilder;
+import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -71,6 +73,7 @@
  * @author Peter-Josef Meisch
  * @author Henrique Amaral
  * @author Russell Parry
+ * @author Thomas Geese
  */
 @SpringIntegrationTest
 @ContextConfiguration(classes = { ElasticsearchRestTemplateConfiguration.class })
@@ -681,6 +684,32 @@ public void aggregateReturnsAggregationResults() throws IOException {
 				.expectNextMatches(aggregation -> aggregation.getType().equals(StringTerms.NAME)).verifyComplete();
 	}
 
+	@Test // DATAES-866
+	public void suggestReturnsSuggestionResults() throws IOException {
+		syncClient.indices().create(new CreateIndexRequest(INDEX_I), RequestOptions.DEFAULT);
+		Map<String, Object> jsonMap = Collections.singletonMap("properties",
+				Collections.singletonMap("firstname", Collections.singletonMap("type", "completion")));
+		syncClient.indices().putMapping(new PutMappingRequest(INDEX_I).source(jsonMap), RequestOptions.DEFAULT);
+
+		addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		SuggestBuilder suggestBuilder = new SuggestBuilder().addSuggestion(
+				"firstname",
+				new CompletionSuggestionBuilder("firstname").prefix("ch")
+		);
+
+		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().query(QueryBuilders.matchAllQuery());
+		searchSourceBuilder.suggest(suggestBuilder);
+
+		SearchRequest request = new SearchRequest(INDEX_I) //
+				.source(searchSourceBuilder);
+
+		client
+				.suggest(request).as(StepVerifier::create).expectNextMatches(suggestions -> suggestions
+						.getSuggestion("firstname").getEntries().get(0).getOptions().get(0).getText().string().equals("chade"))
+				.verifyComplete();
+	}
+
 	private AddToIndexOfType addSourceDocument() {
 		return add(DOC_SOURCE);
 	}