DATAES-866 - Implement suggest query in reactive client.

Original PR: #483

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

@@ -96,6 +96,7 @@ import org.elasticsearch.rest.RestStatus;
 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 @@ import org.springframework.web.reactive.function.client.WebClient.RequestBodySpe
  * @author Henrique Amaral
  * @author Roman Puchkovskiy
  * @author Russell Parry
+ * @author Thomas Geese
  * @since 3.2
  * @see ClientConfiguration
  * @see ReactiveRestClients
@@ -890,6 +892,12 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
 		}
 	}
 
+	@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.BulkByScrollResponse;
 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 @@ import org.springframework.web.reactive.function.client.WebClient;
  * @author Mark Paluch
  * @author Peter-Josef Meisch
  * @author Henrique Amaral
+ * @author Thomas Geese
  * @since 3.2
  * @see ClientConfiguration
  * @see ReactiveRestClients
@@ -417,6 +419,27 @@ public interface ReactiveElasticsearchClient {
 	 */
 	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.get.GetResult;
 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 @@ import org.springframework.util.Assert;
  * @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 class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 		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 reactor.core.publisher.Mono;
 
 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 @@ import org.springframework.data.elasticsearch.core.query.StringQuery;
  *
  * @author Peter-Josef Meisch
  * @author Russell Parry
+ * @author Thomas Geese
  * @since 4.0
  */
 public interface ReactiveSearchOperations {
@@ -206,4 +209,22 @@ public interface ReactiveSearchOperations {
 	 * @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.rest.RestStatus;
 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 @@ import org.springframework.test.context.ContextConfiguration;
  * @author Peter-Josef Meisch
  * @author Henrique Amaral
  * @author Russell Parry
+ * @author Thomas Geese
  */
 @SpringIntegrationTest
 @ContextConfiguration(classes = { ElasticsearchRestTemplateConfiguration.class })
@@ -681,6 +684,32 @@ public class ReactiveElasticsearchClientTests {
 				.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);
 	}