DATAES-796 - Add method returning Mono<SearchPage<T>>.

Original PR: #540

Author: sothawo

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

@@ -24,7 +24,6 @@
 import io.netty.handler.timeout.WriteTimeoutHandler;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.Sinks;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.transport.ProxyProvider;
 
@@ -104,9 +103,7 @@
 import org.elasticsearch.search.aggregations.Aggregation;
 import org.elasticsearch.search.suggest.Suggest;
 import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
-
+import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
 import org.springframework.data.elasticsearch.client.ClientConfiguration;
 import org.springframework.data.elasticsearch.client.ClientLogger;
 import org.springframework.data.elasticsearch.client.ElasticsearchHost;
@@ -428,6 +425,11 @@ public Flux<SearchHit> search(HttpHeaders headers, SearchRequest searchRequest)
 				.flatMap(Flux::fromIterable);
 	}
 
+	@Override
+	public Mono<SearchResponse> searchForResponse(HttpHeaders headers, SearchRequest searchRequest) {
+		return sendRequest(searchRequest, requestCreator.search(), SearchResponse.class, headers).next();
+	}
+
 	@Override
 	public Flux<Suggest> suggest(HttpHeaders headers, SearchRequest searchRequest) {
 		return sendRequest(searchRequest, requestCreator.search(), SearchResponse.class, headers) //
@@ -468,21 +470,19 @@ public Flux<SearchHit> scroll(HttpHeaders headers, SearchRequest searchRequest)
 
 		return Flux.usingWhen(Mono.fromSupplier(ScrollState::new),
 
-				state -> {
-
-					return sendRequest(searchRequest, requestCreator.search(), SearchResponse.class, headers)
-							.expand(searchResponse -> {
+				state -> sendRequest(searchRequest, requestCreator.search(), SearchResponse.class, headers)
+						.expand(searchResponse -> {
 
-								state.updateScrollId(searchResponse.getScrollId());
-								if (isEmpty(searchResponse.getHits())) {
-									return Mono.empty();
-								}
+							state.updateScrollId(searchResponse.getScrollId());
+							if (isEmpty(searchResponse.getHits())) {
+								return Mono.empty();
+							}
 
-								return sendRequest(new SearchScrollRequest(searchResponse.getScrollId()).scroll(scrollTimeout),
-										requestCreator.scroll(), SearchResponse.class, headers);
+							return sendRequest(new SearchScrollRequest(searchResponse.getScrollId()).scroll(scrollTimeout),
+									requestCreator.scroll(), SearchResponse.class, headers);
 
-							});
-				}, state -> cleanupScroll(headers, state), //
+						}),
+				state -> cleanupScroll(headers, state), //
 				(state, ex) -> cleanupScroll(headers, state), //
 				state -> cleanupScroll(headers, state)) //
 				.filter(it -> !isEmpty(it.getHits())) //
@@ -776,6 +776,10 @@ private static <T> Mono<T> doDecode(ClientResponse response, Class<T> responseTy
 
 			Method fromXContent = ReflectionUtils.findMethod(responseType, "fromXContent", XContentParser.class);
 
+			if (fromXContent == null) {
+				return Mono.error(new UncategorizedElasticsearchException(
+						"No method named fromXContent found in " + responseType.getCanonicalName()));
+			}
 			return Mono.justOrEmpty(responseType
 					.cast(ReflectionUtils.invokeMethod(fromXContent, responseType, createParser(mediaType, content))));
 
@@ -925,34 +929,5 @@ public Collection<ElasticsearchHost> hosts() {
 		}
 	}
 
-	private static class SinkSubscriber implements Subscriber<SearchResponse> {
-
-		private final Sinks.Many<SearchResponse> inbound;
-
-		public SinkSubscriber(Sinks.Many<SearchResponse> inbound) {
-			this.inbound = inbound;
-		}
-
-		@Override
-		public void onSubscribe(Subscription s) {
-			s.request(Long.MAX_VALUE);
-		}
-
-		@Override
-		public void onNext(SearchResponse searchResponse) {
-			inbound.emitNext(searchResponse, Sinks.EmitFailureHandler.FAIL_FAST);
-		}
-
-		@Override
-		public void onError(Throwable t) {
-			inbound.emitError(t, Sinks.EmitFailureHandler.FAIL_FAST);
-		}
-
-		@Override
-		public void onComplete() {
-			inbound.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
-		}
-	}
-
 	// endregion
 }

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

@@ -427,6 +427,31 @@ default Flux<SearchHit> search(SearchRequest searchRequest) {
 	 */
 	Flux<SearchHit> search(HttpHeaders headers, SearchRequest searchRequest);
 
+	/**
+	 * Execute the given {@link SearchRequest} against the {@literal search} API returning the whole response in one Mono.
+	 *
+	 * @param searchRequest must not be {@literal null}.
+	 * @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html">Search API on
+	 *      elastic.co</a>
+	 * @return the {@link Mono} emitting the whole {@link SearchResponse}.
+	 * @since 4.1
+	 */
+	default Mono<SearchResponse> searchForResponse(SearchRequest searchRequest) {
+		return searchForResponse(HttpHeaders.EMPTY, searchRequest);
+	}
+
+	/**
+	 * Execute the given {@link SearchRequest} against the {@literal search} API returning the whole response in one Mono.
+	 *
+	 * @param headers Use {@link HttpHeaders} to provide eg. authentication data. Must not be {@literal null}.
+	 * @param searchRequest must not be {@literal null}.
+	 * @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html">Search API on
+	 *      elastic.co</a>
+	 * @return the {@link Mono} emitting the whole {@link SearchResponse}.
+	 * @since 4.1
+	 */
+	Mono<SearchResponse> searchForResponse(HttpHeaders headers, SearchRequest searchRequest);
+
 	/**
 	 * Execute the given {@link SearchRequest} against the {@literal search} API.
 	 *

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

@@ -62,6 +62,7 @@
 import org.springframework.data.elasticsearch.core.document.Document;
 import org.springframework.data.elasticsearch.core.document.DocumentAdapters;
 import org.springframework.data.elasticsearch.core.document.SearchDocument;
+import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
 import org.springframework.data.elasticsearch.core.event.ReactiveAfterConvertCallback;
 import org.springframework.data.elasticsearch.core.event.ReactiveAfterSaveCallback;
 import org.springframework.data.elasticsearch.core.event.ReactiveBeforeConvertCallback;
@@ -296,7 +297,7 @@ public <T> Flux<T> multiGet(Query query, Class<T> clazz, IndexCoordinates index)
 
 		MultiGetRequest request = requestFactory.multiGetRequest(query, clazz, index);
 		return Flux.from(execute(client -> client.multiGet(request))) //
-				.concatMap(result -> callback.doWith(DocumentAdapters.from(result)));
+				.concatMap(result -> callback.toEntity(DocumentAdapters.from(result)));
 	}
 
 	@Override
@@ -444,7 +445,7 @@ public <T> Mono<T> get(String id, Class<T> entityType, IndexCoordinates index) {
 
 		DocumentCallback<T> callback = new ReadDocumentCallback<>(converter, entityType, index);
 
-		return doGet(id, index).flatMap(it -> callback.doWith(DocumentAdapters.from(it)));
+		return doGet(id, index).flatMap(it -> callback.toEntity(DocumentAdapters.from(it)));
 	}
 
 	private Mono<GetResult> doGet(String id, IndexCoordinates index) {
@@ -656,14 +657,34 @@ protected <R extends WriteRequest<R>> R prepareWriteRequest(R request) {
 	@Override
 	public <T> Flux<SearchHit<T>> search(Query query, Class<?> entityType, Class<T> resultType, IndexCoordinates index) {
 		SearchDocumentCallback<T> callback = new ReadSearchDocumentCallback<>(resultType, index);
-		return doFind(query, entityType, index).concatMap(callback::doWith);
+		return doFind(query, entityType, index).concatMap(callback::toSearchHit);
 	}
 
 	@Override
 	public <T> Flux<SearchHit<T>> search(Query query, Class<?> entityType, Class<T> returnType) {
 		return search(query, entityType, returnType, getIndexCoordinatesFor(entityType));
 	}
 
+	@Override
+	public <T> Mono<SearchPage<T>> searchForPage(Query query, Class<?> entityType, Class<T> resultType) {
+		return searchForPage(query, entityType, resultType, getIndexCoordinatesFor(entityType));
+	}
+
+	@Override
+	public <T> Mono<SearchPage<T>> searchForPage(Query query, Class<?> entityType, Class<T> resultType,
+			IndexCoordinates index) {
+
+		SearchDocumentCallback<T> callback = new ReadSearchDocumentCallback<>(resultType, index);
+
+		return doFindForResponse(query, entityType, index) //
+				.flatMap(searchDocumentResponse -> Flux.fromIterable(searchDocumentResponse.getSearchDocuments()) //
+						.flatMap(callback::toEntity) //
+						.collectList() //
+						.map(entities -> SearchHitMapping.mappingFor(resultType, converter) //
+								.mapHits(searchDocumentResponse, entities))) //
+				.map(searchHits -> SearchHitSupport.searchPageFor(searchHits, query.getPageable()));
+	}
+
 	private Flux<SearchDocument> doFind(Query query, Class<?> clazz, IndexCoordinates index) {
 
 		return Flux.defer(() -> {
@@ -678,6 +699,15 @@ private Flux<SearchDocument> doFind(Query query, Class<?> clazz, IndexCoordinate
 		});
 	}
 
+	private Mono<SearchDocumentResponse> doFindForResponse(Query query, Class<?> clazz, IndexCoordinates index) {
+
+		return Mono.defer(() -> {
+			SearchRequest request = requestFactory.searchRequest(query, clazz, index);
+			request = prepareSearchRequest(request);
+			return doFindForResponse(request);
+		});
+	}
+
 	@Override
 	public Flux<Aggregation> aggregate(Query query, Class<?> entityType) {
 		return aggregate(query, entityType, getIndexCoordinatesFor(entityType));
@@ -748,6 +778,21 @@ protected Flux<SearchDocument> doFind(SearchRequest request) {
 				.onErrorResume(NoSuchIndexException.class, it -> Mono.empty());
 	}
 
+	/**
+	 * Customization hook on the actual execution result {@link Mono}. <br />
+	 *
+	 * @param request the already prepared {@link SearchRequest} ready to be executed.
+	 * @return a {@link Mono} emitting the result of the operation converted to s {@link SearchDocumentResponse}.
+	 */
+	protected Mono<SearchDocumentResponse> doFindForResponse(SearchRequest request) {
+
+		if (QUERY_LOGGER.isDebugEnabled()) {
+			QUERY_LOGGER.debug("Executing doFindForResponse: {}", request);
+		}
+
+		return Mono.from(execute(client1 -> client1.searchForResponse(request))).map(SearchDocumentResponse::from);
+	}
+
 	/**
 	 * Customization hook on the actual execution result {@link Publisher}. <br />
 	 *
@@ -935,7 +980,7 @@ protected <T> Mono<T> maybeCallAfterConvert(T entity, Document document, IndexCo
 	protected interface DocumentCallback<T> {
 
 		@NonNull
-		Mono<T> doWith(@Nullable Document document);
+		Mono<T> toEntity(@Nullable Document document);
 	}
 
 	protected class ReadDocumentCallback<T> implements DocumentCallback<T> {
@@ -953,7 +998,7 @@ public ReadDocumentCallback(EntityReader<? super T, Document> reader, Class<T> t
 		}
 
 		@NonNull
-		public Mono<T> doWith(@Nullable Document document) {
+		public Mono<T> toEntity(@Nullable Document document) {
 			if (document == null) {
 				return Mono.empty();
 			}
@@ -966,7 +1011,10 @@ public Mono<T> doWith(@Nullable Document document) {
 	protected interface SearchDocumentCallback<T> {
 
 		@NonNull
-		Mono<SearchHit<T>> doWith(@NonNull SearchDocument response);
+		Mono<T> toEntity(@NonNull SearchDocument response);
+
+		@NonNull
+		Mono<SearchHit<T>> toSearchHit(@NonNull SearchDocument response);
 	}
 
 	protected class ReadSearchDocumentCallback<T> implements SearchDocumentCallback<T> {
@@ -981,9 +1029,13 @@ public ReadSearchDocumentCallback(Class<T> type, IndexCoordinates index) {
 		}
 
 		@Override
-		public Mono<SearchHit<T>> doWith(SearchDocument response) {
-			return delegate.doWith(response)
-					.map(entity -> SearchHitMapping.mappingFor(type, converter).mapHit(response, entity));
+		public Mono<T> toEntity(SearchDocument response) {
+			return delegate.toEntity(response);
+		}
+
+		@Override
+		public Mono<SearchHit<T>> toSearchHit(SearchDocument response) {
+			return toEntity(response).map(entity -> SearchHitMapping.mappingFor(type, converter).mapHit(response, entity));
 		}
 	}