DATAES-717 - Enable Repositories to return a SearchHits instance instead of a list.

Original PR: #372

src/main/asciidoc/reference/elasticsearch-operations.adoc

@@ -133,12 +133,15 @@ Contains the following information:
 * Id
 * Score
 * Sort Values
-* the retrieved entity of type <T>
+* Highligth fields
+* The retrieved entity of type <T>
 
 .SearchHits<T>
 Contains the following information:
 
 * Number of total hits
+* Total hits relation
 * Maximum score
 * A list of `SearchHit<T>` objects
+* Returned aggregations
 

src/main/asciidoc/reference/elasticsearch-repository-queries.adoc

@@ -266,6 +266,16 @@ A list of supported keywords for Elasticsearch is shown below.
 
 |===
 
+== Method return types
+
+Repository methods can be defined to have the following return types for returning multiple Elements:
+
+* `List<T>`
+* `Stream<T>`
+* `SearchHits<T>`
+* `List<SearchHits<T>>`
+* `Stream<>>SearchHit<T>>`
+
 [[elasticsearch.query-methods.at-query]]
 == Using @Query Annotation
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -429,6 +429,10 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 
 	private Flux<SearchDocument> doFind(Query query, Class<?> clazz, IndexCoordinates index) {
 
+		if (query instanceof CriteriaQuery) {
+			converter.updateQuery((CriteriaQuery) query, clazz);
+		}
+
 		return Flux.defer(() -> {
 			SearchRequest request = requestFactory.searchRequest(query, clazz, index);
 
@@ -561,11 +565,10 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 
 	private QueryBuilder mappedQuery(Query query, ElasticsearchPersistentEntity<?> entity) {
 
-		// TODO: we need to actually map the fields to the according field names!
-
 		QueryBuilder elasticsearchQuery = null;
 
 		if (query instanceof CriteriaQuery) {
+			converter.updateQuery((CriteriaQuery) query, entity.getType());
 			elasticsearchQuery = new CriteriaQueryProcessor().createQueryFromCriteria(((CriteriaQuery) query).getCriteria());
 		} else if (query instanceof StringQuery) {
 			elasticsearchQuery = new WrapperQueryBuilder(((StringQuery) query).getSource());

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

@@ -35,11 +35,11 @@ import org.springframework.util.StringUtils;
 public class SearchHits<T> implements Streamable<SearchHit<T>> {
 
 	private final long totalHits;
+	private final TotalHitsRelation totalHitsRelation;
 	private final float maxScore;
 	private final String scrollId;
 	private final List<? extends SearchHit<T>> searchHits;
 	private final Aggregations aggregations;
-	private final TotalHitsRelation totalHitsRelation;
 
 	/**
 	 * @param totalHits the number of total hits for the search

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

@@ -62,7 +62,7 @@ abstract class AbstractReactiveElasticsearchRepositoryQuery implements Repositor
 
 		Object result = queryMethod.hasReactiveWrapperParameter() ? executeDeferred(parameters)
 				: execute(new ReactiveElasticsearchParametersParameterAccessor(queryMethod, parameters));
-		return SearchHitSupport.unwrapSearchHits(result);
+		return queryMethod.isNotSearchHitMethod() ? SearchHitSupport.unwrapSearchHits(result) : result;
 	}
 
 	private Object executeDeferred(Object[] parameters) {
@@ -122,7 +122,8 @@ abstract class AbstractReactiveElasticsearchRepositoryQuery implements Repositor
 			return (query, type, targetType, indexCoordinates) -> operations.search(query.setPageable(accessor.getPageable()),
 					type, targetType, indexCoordinates);
 		} else {
-			return (query, type, targetType, indexCoordinates) -> operations.search(query, type, targetType, indexCoordinates);
+			return (query, type, targetType, indexCoordinates) -> operations.search(query, type, targetType,
+					indexCoordinates);
 		}
 	}
 

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

@@ -86,7 +86,7 @@ public class ElasticsearchPartQuery extends AbstractElasticsearchRepositoryQuery
 			} else {
 				query.setPageable(accessor.getPageable());
 			}
-			result = StreamUtils.createStreamFromIterator(elasticsearchOperations.stream(query, clazz, index));
+			result = StreamUtils.createStreamFromIterator(elasticsearchOperations.searchForStream(query, clazz, index));
 		} else if (queryMethod.isCollectionQuery()) {
 
 			if (accessor.getPageable().isUnpaged()) {
@@ -97,13 +97,14 @@ public class ElasticsearchPartQuery extends AbstractElasticsearchRepositoryQuery
 			}
 
 			result = elasticsearchOperations.search(query, clazz, index);
+
 		} else if (tree.isCountProjection()) {
 			result = elasticsearchOperations.count(query, clazz, index);
 		} else {
 			result = elasticsearchOperations.searchOne(query, clazz, index);
 		}
 
-		return SearchHitSupport.unwrapSearchHits(result);
+		return queryMethod.isNotSearchHitMethod() ? SearchHitSupport.unwrapSearchHits(result) : result;
 	}
 
 	private Object countOrGetDocumentsForDelete(CriteriaQuery query, ParametersParameterAccessor accessor) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,9 +16,14 @@
 package org.springframework.data.elasticsearch.repository.query;
 
 import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.util.Collection;
+import java.util.stream.Stream;
 
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.data.elasticsearch.annotations.Query;
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.elasticsearch.core.SearchHits;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
 import org.springframework.data.mapping.context.MappingContext;
@@ -37,22 +42,25 @@ import org.springframework.util.ClassUtils;
  * @author Oliver Gierke
  * @author Mark Paluch
  * @author Christoph Strobl
+ * @author Peter-Josef Meisch
  */
 public class ElasticsearchQueryMethod extends QueryMethod {
 
+	private final Method method; // private in base class, but needed here as well
 	private final Query queryAnnotation;
 	private final MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext;
 	private @Nullable ElasticsearchEntityMetadata<?> metadata;
 
-	public ElasticsearchQueryMethod(Method method, RepositoryMetadata metadata, ProjectionFactory factory,
+	public ElasticsearchQueryMethod(Method method, RepositoryMetadata repositoryMetadata, ProjectionFactory factory,
 			MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> mappingContext) {
 
-		super(method, metadata, factory);
+		super(method, repositoryMetadata, factory);
 
 		Assert.notNull(mappingContext, "MappingContext must not be null!");
 
-		this.queryAnnotation = method.getAnnotation(Query.class);
+		this.method = method;
 		this.mappingContext = mappingContext;
+		this.queryAnnotation = method.getAnnotation(Query.class);
 	}
 
 	public boolean hasAnnotatedQuery() {
@@ -101,4 +109,52 @@ public class ElasticsearchQueryMethod extends QueryMethod {
 	protected MappingContext<? extends ElasticsearchPersistentEntity<?>, ElasticsearchPersistentProperty> getMappingContext() {
 		return mappingContext;
 	}
+
+	/**
+	 * checks whether the return type of the underlying method is a
+	 * {@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
+	 * @since 4.0
+	 */
+	public boolean isSearchHitMethod() {
+		Class<?> methodReturnType = method.getReturnType();
+
+		if (SearchHits.class.isAssignableFrom(methodReturnType)) {
+			return true;
+		}
+
+		try {
+			// dealing with Collection<SearchHit<T>>, getting to T
+			ParameterizedType methodGenericReturnType = ((ParameterizedType) method.getGenericReturnType());
+			if (isAllowedGenericType(methodGenericReturnType)) {
+				ParameterizedType collectionTypeArgument = (ParameterizedType) methodGenericReturnType
+						.getActualTypeArguments()[0];
+				if (SearchHit.class.isAssignableFrom((Class<?>) collectionTypeArgument.getRawType())) {
+					return true;
+				}
+			}
+		} catch (Exception ignored) {}
+
+		return false;
+	}
+
+	protected boolean isAllowedGenericType(ParameterizedType methodGenericReturnType) {
+		return Collection.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType())
+				|| Stream.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType());
+	}
+
+	/**
+	 * checks whether the return type of the underlying method is a
+	 * {@link org.springframework.data.elasticsearch.core.SearchHits} or a collection of
+	 * {@link org.springframework.data.elasticsearch.core.SearchHit}.
+	 *
+	 * @return true if the method has not a {@link org.springframework.data.elasticsearch.core.SearchHit}t related return
+	 *         type
+	 * @since 4.0
+	 */
+	public boolean isNotSearchHitMethod() {
+		return !isSearchHitMethod();
+	}
 }

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

@@ -89,8 +89,7 @@ public class ElasticsearchStringQuery extends AbstractElasticsearchRepositoryQue
 			result = elasticsearchOperations.searchOne(stringQuery, clazz, index);
 		}
 
-		return SearchHitSupport.unwrapSearchHits(result);
-
+		return queryMethod.isNotSearchHitMethod() ? SearchHitSupport.unwrapSearchHits(result) : result;
 	}
 
 	protected StringQuery createQuery(ParametersParameterAccessor parameterAccessor) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,7 +17,10 @@ package org.springframework.data.elasticsearch.repository.query;
 
 import static org.springframework.data.repository.util.ClassUtils.*;
 
+import reactor.core.publisher.Flux;
+
 import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
 
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.domain.Page;
@@ -115,4 +118,11 @@ public class ReactiveElasticsearchQueryMethod extends ElasticsearchQueryMethod {
 	public ElasticsearchParameters getParameters() {
 		return (ElasticsearchParameters) super.getParameters();
 	}
+
+	@Override
+	protected boolean isAllowedGenericType(ParameterizedType methodGenericReturnType) {
+		return super.isAllowedGenericType(methodGenericReturnType)
+				|| Flux.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType());
+	}
+
 }

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

@@ -115,4 +115,9 @@ public class ElasticsearchRepositoryFactory extends RepositoryFactorySupport {
 			return new ElasticsearchPartQuery(queryMethod, elasticsearchOperations);
 		}
 	}
+
+	@Override
+	protected RepositoryMetadata getRepositoryMetadata(Class<?> repositoryInterface) {
+		return new ElasticsearchRepositoryMetadata(repositoryInterface);
+	}
 }

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.repository.support;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.util.Collection;
+import java.util.stream.Stream;
+
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.repository.core.support.DefaultRepositoryMetadata;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.0
+ */
+public class ElasticsearchRepositoryMetadata extends DefaultRepositoryMetadata {
+
+	public ElasticsearchRepositoryMetadata(Class<?> repositoryInterface) {
+		super(repositoryInterface);
+	}
+
+	@Override
+	public Class<?> getReturnedDomainClass(Method method) {
+		Class<?> returnedDomainClass = super.getReturnedDomainClass(method);
+		if (SearchHit.class.isAssignableFrom(returnedDomainClass)) {
+			try {
+				// dealing with Collection<SearchHit<T>> or Flux<SearchHit<T>>, getting to T
+				ParameterizedType methodGenericReturnType = ((ParameterizedType) method.getGenericReturnType());
+				if (isAllowedGenericType(methodGenericReturnType)) {
+					ParameterizedType collectionTypeArgument = (ParameterizedType) methodGenericReturnType
+							.getActualTypeArguments()[0];
+					if (SearchHit.class.isAssignableFrom((Class<?>) collectionTypeArgument.getRawType())) {
+						returnedDomainClass = (Class<?>) collectionTypeArgument.getActualTypeArguments()[0];
+					}
+				}
+			} catch (Exception ignored) {}
+		}
+		return returnedDomainClass;
+	}
+
+	protected boolean isAllowedGenericType(ParameterizedType methodGenericReturnType) {
+		return Collection.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType())
+				|| Stream.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType());
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -120,6 +120,11 @@ public class ReactiveElasticsearchRepositoryFactory extends ReactiveRepositoryFa
 				entity.getIndexCoordinates(), entity.getVersionType());
 	}
 
+	@Override
+	protected RepositoryMetadata getRepositoryMetadata(Class<?> repositoryInterface) {
+		return new ReactiveElasticsearchRepositoryMetadata(repositoryInterface);
+	}
+
 	/**
 	 * @author Christoph Strobl
 	 */

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.repository.support;
+
+import reactor.core.publisher.Flux;
+
+import java.lang.reflect.ParameterizedType;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.0
+ */
+public class ReactiveElasticsearchRepositoryMetadata extends ElasticsearchRepositoryMetadata {
+
+	public ReactiveElasticsearchRepositoryMetadata(Class<?> repositoryInterface) {
+		super(repositoryInterface);
+	}
+
+	@Override
+	protected boolean isAllowedGenericType(ParameterizedType methodGenericReturnType) {
+		return super.isAllowedGenericType(methodGenericReturnType)
+				|| Flux.class.isAssignableFrom((Class<?>) methodGenericReturnType.getRawType());
+	}
+}

src/test/java/org/springframework/data/elasticsearch/repositories/custommethod/CustomMethodRepositoryBaseTests.java

@@ -30,6 +30,7 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import org.junit.jupiter.api.AfterEach;
@@ -48,6 +49,8 @@ import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.annotations.Query;
 import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
 import org.springframework.data.elasticsearch.core.IndexOperations;
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.elasticsearch.core.SearchHits;
 import org.springframework.data.elasticsearch.core.geo.GeoBox;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
 import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
@@ -379,7 +382,7 @@ public abstract class CustomMethodRepositoryBaseTests {
 		List<SampleEntity> list = repository.findByKeywordIn(keywords);
 
 		// then
-        assertThat(list).hasSize(1);
+		assertThat(list).hasSize(1);
 		assertThat(list.get(0).getId()).isEqualTo(documentId1);
 	}
 
@@ -1366,6 +1369,53 @@ public abstract class CustomMethodRepositoryBaseTests {
 		stream.forEach(o -> assertThat(o).isInstanceOf(SampleEntity.class));
 	}
 
+	@Test // DATAES-717
+	void shouldReturnSearchHits() {
+		List<SampleEntity> entities = createSampleEntities("abc", 20);
+		repository.saveAll(entities);
+
+		// when
+		SearchHits<SampleEntity> searchHits = repository.queryByType("abc");
+
+		assertThat(searchHits.getTotalHits()).isEqualTo(20);
+	}
+
+	@Test // DATAES-717
+	void shouldReturnSearchHitList() {
+		List<SampleEntity> entities = createSampleEntities("abc", 20);
+		repository.saveAll(entities);
+
+		// when
+		List<SearchHit<SampleEntity>> searchHitList = repository.queryByMessage("Message");
+
+		assertThat(searchHitList).hasSize(20);
+	}
+
+	@Test // DATAES-717
+	void shouldReturnSearchHitStream() {
+		List<SampleEntity> entities = createSampleEntities("abc", 20);
+		repository.saveAll(entities);
+
+		// when
+		Stream<SearchHit<SampleEntity>> searchHitStream = repository.readByMessage("Message");
+
+		List<SearchHit<SampleEntity>> searchHitList = searchHitStream //
+				.peek(searchHit -> assertThat(searchHit.getContent().getType()).isEqualTo("abc")) //
+				.collect(Collectors.toList());
+		assertThat(searchHitList).hasSize(20);
+	}
+
+	@Test // DATAES-717
+	void shouldReturnSearchHitsForStringQuery() {
+		List<SampleEntity> entities = createSampleEntities("abc", 20);
+		repository.saveAll(entities);
+
+		// when
+		SearchHits<SampleEntity> searchHits = repository.queryByString("abc");
+
+		assertThat(searchHits.getTotalHits()).isEqualTo(20);
+	}
+
 	private List<SampleEntity> createSampleEntities(String type, int numberOfEntities) {
 
 		List<SampleEntity> entities = new ArrayList<>();
@@ -1386,8 +1436,7 @@ public abstract class CustomMethodRepositoryBaseTests {
 	@NoArgsConstructor
 	@AllArgsConstructor
 	@Builder
-	@Document(indexName = "test-index-sample-repositories-custo-method", replicas = 0,
-			refreshInterval = "-1")
+	@Document(indexName = "test-index-sample-repositories-custom-method", replicas = 0, refreshInterval = "-1")
 	static class SampleEntity {
 
 		@Id private String id;
@@ -1502,6 +1551,15 @@ public abstract class CustomMethodRepositoryBaseTests {
 		long countByLocationNear(Point point, Distance distance);
 
 		long countByLocationNear(GeoPoint point, String distance);
+
+		SearchHits<SampleEntity> queryByType(String type);
+
+		@Query("{\"bool\": {\"must\": [{\"term\": {\"type\": \"?0\"}}]}}")
+		SearchHits<SampleEntity> queryByString(String type);
+
+		List<SearchHit<SampleEntity>> queryByMessage(String type);
+
+		Stream<SearchHit<SampleEntity>> readByMessage(String type);
 	}
 
 	/**

src/test/java/org/springframework/data/elasticsearch/repository/support/SimpleReactiveElasticsearchRepositoryTests.java

@@ -23,7 +23,6 @@ import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.junit.jupiter.api.AfterEach;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -43,6 +42,7 @@ import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
@@ -62,6 +62,7 @@ import org.springframework.data.elasticsearch.annotations.Query;
 import org.springframework.data.elasticsearch.annotations.Score;
 import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
 import org.springframework.data.elasticsearch.config.AbstractReactiveElasticsearchConfiguration;
+import org.springframework.data.elasticsearch.core.SearchHit;
 import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchRestTemplateConfiguration;
 import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
 import org.springframework.data.elasticsearch.repository.config.EnableReactiveElasticsearchRepositories;
@@ -181,7 +182,8 @@ public class SimpleReactiveElasticsearchRepositoryTests {
 
 		repository.findAllById(Arrays.asList("id-one", "id-two")) //
 				.as(StepVerifier::create)//
-				.expectNextCount(2) //
+				.expectNextMatches(entity -> entity.getId().equals("id-one") || entity.getId().equals("id-two")) //
+				.expectNextMatches(entity -> entity.getId().equals("id-one") || entity.getId().equals("id-two")) //
 				.verifyComplete();
 	}
 
@@ -197,6 +199,34 @@ public class SimpleReactiveElasticsearchRepositoryTests {
 				.verifyComplete();
 	}
 
+	@Test // DATAES-717
+	void shouldReturnFluxOfSearchHit() throws IOException {
+
+		bulkIndex(SampleEntity.builder().id("id-one").message("message").build(), //
+				SampleEntity.builder().id("id-two").message("message").build(), //
+				SampleEntity.builder().id("id-three").message("message").build());
+
+		repository.queryByMessageWithString("message") //
+				.as(StepVerifier::create) //
+				.expectNextMatches(searchHit -> SearchHit.class.isAssignableFrom(searchHit.getClass()))//
+				.expectNextCount(2) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-717
+	void shouldReturnFluxOfSearchHitForStringQuery() throws IOException {
+
+		bulkIndex(SampleEntity.builder().id("id-one").message("message").build(), //
+				SampleEntity.builder().id("id-two").message("message").build(), //
+				SampleEntity.builder().id("id-three").message("message").build());
+
+		repository.queryAllByMessage("message") //
+				.as(StepVerifier::create) //
+				.expectNextMatches(searchHit -> SearchHit.class.isAssignableFrom(searchHit.getClass()))//
+				.expectNextCount(2) //
+				.verifyComplete();
+	}
+
 	@Test // DATAES-519
 	public void countShouldReturnZeroWhenIndexDoesNotExist() {
 		repository.count().as(StepVerifier::create).expectNext(0L).verifyComplete();
@@ -505,6 +535,11 @@ public class SimpleReactiveElasticsearchRepositoryTests {
 
 		Flux<SampleEntity> findAllByMessage(Publisher<String> message);
 
+		Flux<SearchHit<SampleEntity>> queryAllByMessage(String message);
+
+		@Query("{\"bool\": {\"must\": [{\"term\": {\"message\": \"?0\"}}]}}")
+		Flux<SearchHit<SampleEntity>> queryByMessageWithString(String message);
+
 		@Query("{ \"bool\" : { \"must\" : { \"term\" : { \"message\" : \"?0\" } } } }")
 		Flux<SampleEntity> findAllViaAnnotatedQueryByMessageLike(String message);