Support Kotlin Flow as possible return type in repository functions.

Original Pull Request #2387
Closes #2386

(cherry picked from commit 1fa6c9f3e5539b502ec3e8619929ae8cfbe050fd)

src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java

@@ -115,7 +115,7 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch
 
 		return entitiesPublisher //
 				.flatMapMany(entities -> Flux.fromIterable(entities) //
-						.concatMap(entity -> maybeCallBeforeConvert(entity, index)) //
+						.concatMap(entity -> maybeCallbackBeforeConvert(entity, index)) //
 				).collectList() //
 				.map(Entities::new) //
 				.flatMapMany(entities -> {
@@ -131,7 +131,7 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch
 								BulkResponseItem response = indexAndResponse.getT2();
 								updateIndexedObject(savedEntity, IndexedObjectInformation.of(response.id(), response.seqNo(),
 										response.primaryTerm(), response.version()));
-								return maybeCallAfterSave(savedEntity, index);
+								return maybeCallbackAfterSave(savedEntity, index);
 							});
 				});
 	}
@@ -329,8 +329,8 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch
 
 		Flux<ResponseBody<EntityAsMap>> searchResponses = Flux.usingWhen(Mono.fromSupplier(ScrollState::new), //
 				state -> Mono
-						.from(execute((ClientCallback<Publisher<ResponseBody<EntityAsMap>>>) client1 -> client1
+						.from(execute((ClientCallback<Publisher<ResponseBody<EntityAsMap>>>) client -> client.search(searchRequest,
-								.search(searchRequest, EntityAsMap.class))) //
+								EntityAsMap.class))) //
 						.expand(entityAsMapSearchResponse -> {
 
 							state.updateScrollId(entityAsMapSearchResponse.scrollId());
@@ -354,6 +354,10 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch
 
 	private Publisher<?> cleanupScroll(ScrollState state) {
 
+		if (state.getScrollIds().isEmpty()) {
+			return Mono.empty();
+		}
+
 		return execute((ClientCallback<Publisher<ClearScrollResponse>>) client -> client
 				.clearScroll(ClearScrollRequest.of(csr -> csr.scrollId(state.getScrollIds()))));
 	}

src/main/java/org/springframework/data/elasticsearch/client/erhlc/ReactiveElasticsearchTemplate.java

@@ -139,7 +139,7 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch
 
 		return entitiesPublisher //
 				.flatMapMany(entities -> Flux.fromIterable(entities) //
-						.concatMap(entity -> maybeCallBeforeConvert(entity, index)) //
+						.concatMap(entity -> maybeCallbackBeforeConvert(entity, index)) //
 				).collectList() //
 				.map(Entities::new) //
 				.flatMapMany(entities -> {
@@ -158,7 +158,7 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch
 								updateIndexedObject(savedEntity, IndexedObjectInformation.of(response.getId(), response.getSeqNo(),
 										response.getPrimaryTerm(), response.getVersion()));
 
-								return maybeCallAfterSave(savedEntity, index);
+								return maybeCallbackAfterSave(savedEntity, index);
 							});
 				});
 	}

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

@@ -311,7 +311,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
 		Assert.notNull(entity, "Entity must not be null!");
 		Assert.notNull(index, "index must not be null");
 
-		return maybeCallBeforeConvert(entity, index)
+		return maybeCallbackBeforeConvert(entity, index)
 				.flatMap(entityAfterBeforeConversionCallback -> doIndex(entityAfterBeforeConversionCallback, index)) //
 				.map(it -> {
 					T savedEntity = it.getT1();
@@ -321,7 +321,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
 							indexResponseMetaData.seqNo(), //
 							indexResponseMetaData.primaryTerm(), //
 							indexResponseMetaData.version()));
-				}).flatMap(saved -> maybeCallAfterSave(saved, index));
+				}).flatMap(saved -> maybeCallbackAfterSave(saved, index));
 	}
 
 	abstract protected <T> Mono<Tuple2<T, IndexResponseMetaData>> doIndex(T entity, IndexCoordinates index);
@@ -493,7 +493,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
 
 	// region callbacks
 
-	protected <T> Mono<T> maybeCallBeforeConvert(T entity, IndexCoordinates index) {
+	protected <T> Mono<T> maybeCallbackBeforeConvert(T entity, IndexCoordinates index) {
 
 		if (null != entityCallbacks) {
 			return entityCallbacks.callback(ReactiveBeforeConvertCallback.class, entity, index);
@@ -502,7 +502,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
 		return Mono.just(entity);
 	}
 
-	protected <T> Mono<T> maybeCallAfterSave(T entity, IndexCoordinates index) {
+	protected <T> Mono<T> maybeCallbackAfterSave(T entity, IndexCoordinates index) {
 
 		if (null != entityCallbacks) {
 			return entityCallbacks.callback(ReactiveAfterSaveCallback.class, entity, index);
@@ -511,7 +511,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
 		return Mono.just(entity);
 	}
 
-	protected <T> Mono<T> maybeCallAfterConvert(T entity, Document document, IndexCoordinates index) {
+	protected <T> Mono<T> maybeCallbackAfterConvert(T entity, Document document, IndexCoordinates index) {
 
 		if (null != entityCallbacks) {
 			return entityCallbacks.callback(ReactiveAfterConvertCallback.class, entity, document, index);
@@ -528,8 +528,19 @@ abstract public class AbstractReactiveElasticsearchTemplate
 		return Mono.just(document);
 	}
 
+	/**
+	 * Callback to convert {@link Document} into an entity of type T
+	 *
+	 * @param <T> the entity type
+	 */
 	protected interface DocumentCallback<T> {
 
+		/**
+		 * Convert a document into an entity
+		 *
+		 * @param document the document to convert
+		 * @return a Mono of the entity
+		 */
 		@NonNull
 		Mono<T> toEntity(@Nullable Document document);
 	}
@@ -566,16 +577,30 @@ abstract public class AbstractReactiveElasticsearchTemplate
 								documentAfterLoad.hasVersion() ? documentAfterLoad.getVersion() : null); //
 						entity = updateIndexedObject(entity, indexedObjectInformation);
 
-						return maybeCallAfterConvert(entity, documentAfterLoad, index);
+						return maybeCallbackAfterConvert(entity, documentAfterLoad, index);
 					});
 		}
 	}
 
+	/**
+	 * Callback to convert a {@link SearchDocument} into different other classes
+	 * @param <T> the entity type
+	 */
 	protected interface SearchDocumentCallback<T> {
 
-		Mono<T> toEntity(SearchDocument response);
+		/**
+		 * converts a {@link SearchDocument} to an entity
+		 * @param searchDocument
+		 * @return the entity in a MOno
+		 */
+		Mono<T> toEntity(SearchDocument searchDocument);
 
-		Mono<SearchHit<T>> toSearchHit(SearchDocument response);
+		/**
+		 * converts a {@link SearchDocument} into a SearchHit
+		 * @param searchDocument
+		 * @return
+		 */
+		Mono<SearchHit<T>> toSearchHit(SearchDocument searchDocument);
 	}
 
 	protected class ReadSearchDocumentCallback<T> implements SearchDocumentCallback<T> {

src/main/java/org/springframework/data/elasticsearch/repository/query/AbstractReactiveElasticsearchRepositoryQuery.java

@@ -94,14 +94,14 @@ abstract class AbstractReactiveElasticsearchRepositoryQuery implements Repositor
 			query.addSourceFilter(sourceFilter);
 		}
 
-		Class<?> targetType = processor.getReturnedType().getTypeToRead();
 		String indexName = queryMethod.getEntityInformation().getIndexName();
 		IndexCoordinates index = IndexCoordinates.of(indexName);
 
 		ReactiveElasticsearchQueryExecution execution = getExecution(parameterAccessor,
 				new ResultProcessingConverter(processor));
 
-		return execution.execute(query, processor.getReturnedType().getDomainType(), targetType, index);
+		var returnedType = processor.getReturnedType();
+		return execution.execute(query, returnedType.getDomainType(), returnedType.getTypeToRead(), index);
 	}
 
 	private ReactiveElasticsearchQueryExecution getExecution(ElasticsearchParameterAccessor accessor,

src/main/java/org/springframework/data/elasticsearch/repository/query/ElasticsearchQueryMethod.java

@@ -184,7 +184,7 @@ public class ElasticsearchQueryMethod extends QueryMethod {
 	 * {@link org.springframework.data.elasticsearch.core.SearchHits} or a collection of
 	 * {@link org.springframework.data.elasticsearch.core.SearchHit}.
 	 *
-	 * @return true if the method has a {@link org.springframework.data.elasticsearch.core.SearchHit}t related return type
+	 * @return true if the method has a {@link org.springframework.data.elasticsearch.core.SearchHit} related return type
 	 * @since 4.0
 	 */
 	public boolean isSearchHitMethod() {
@@ -298,7 +298,8 @@ public class ElasticsearchQueryMethod extends QueryMethod {
 		return fetchSourceFilterBuilder.build();
 	}
 
-	private String[] mapParameters(String[] source, ParameterAccessor parameterAccessor, StringQueryUtil stringQueryUtil) {
+	private String[] mapParameters(String[] source, ParameterAccessor parameterAccessor,
+			StringQueryUtil stringQueryUtil) {
 
 		List<String> fieldNames = new ArrayList<>();
 

src/main/java/org/springframework/data/elasticsearch/repository/query/ReactiveElasticsearchQueryMethod.java

@@ -17,7 +17,6 @@ package org.springframework.data.elasticsearch.repository.query;
 
 import static org.springframework.data.repository.util.ClassUtils.*;
 
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.lang.reflect.Method;
@@ -36,8 +35,8 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.repository.core.RepositoryMetadata;
 import org.springframework.data.repository.util.ReactiveWrapperConverters;
-import org.springframework.data.repository.util.ReactiveWrappers;
 import org.springframework.data.util.Lazy;
+import org.springframework.data.util.ReactiveWrappers;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.ClassUtils;
 
@@ -79,12 +78,14 @@ public class ReactiveElasticsearchQueryMethod extends ElasticsearchQueryMethod {
 
 			if (hasParameterOfType(method, Sort.class)) {
 				throw new IllegalStateException(String.format("Method must not have Pageable *and* Sort parameter. "
-						+ "Use sorting capabilities on Pageble instead! Offending method: %s", method));
+						+ "Use sorting capabilities on Pageable instead! Offending method: %s", method));
 			}
 		}
 
-		this.isCollectionQuery = Lazy.of(() -> (!(isPageQuery() || isSliceQuery())
+		this.isCollectionQuery = Lazy.of(() -> {
-				&& ReactiveWrappers.isMultiValueType(metadata.getReturnType(method).getType()) || super.isCollectionQuery()));
+			return (!(isPageQuery() || isSliceQuery())
+					&& ReactiveWrappers.isMultiValueType(metadata.getReturnType(method).getType()) || super.isCollectionQuery());
+		});
 	}
 
 	@Override
@@ -150,7 +151,7 @@ public class ReactiveElasticsearchQueryMethod extends ElasticsearchQueryMethod {
 	@Override
 	protected boolean isAllowedGenericType(ParameterizedType methodGenericReturnType) {
 		return super.isAllowedGenericType(methodGenericReturnType)
-				|| Flux.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType());
+				|| ReactiveWrappers.supports((Class<?>) methodGenericReturnType.getRawType());
 	}
 
 }

src/main/java/org/springframework/data/elasticsearch/repository/support/ReactiveElasticsearchRepositoryMetadata.java

@@ -15,10 +15,10 @@
  */
 package org.springframework.data.elasticsearch.repository.support;
 
-import reactor.core.publisher.Flux;
-
 import java.lang.reflect.ParameterizedType;
 
+import org.springframework.data.util.ReactiveWrappers;
+
 /**
  * @author Peter-Josef Meisch
  * @since 4.0
@@ -32,6 +32,6 @@ public class ReactiveElasticsearchRepositoryMetadata extends ElasticsearchReposi
 	@Override
 	protected boolean isAllowedGenericType(ParameterizedType methodGenericReturnType) {
 		return super.isAllowedGenericType(methodGenericReturnType)
-				|| Flux.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType());
+				|| ReactiveWrappers.supports((Class<?>) methodGenericReturnType.getRawType());
 	}
 }