DATAES-749 - Introduce SearchPage as return type for repository methods.

Original PR: spring-projects#397

Author: sothawo

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

@@ -21,9 +21,12 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
 import org.springframework.data.elasticsearch.core.aggregation.impl.AggregatedPageImpl;
+import org.springframework.data.elasticsearch.support.ReactiveSupport;
+import org.springframework.lang.Nullable;
 
 /**
  * Utility class with helper methods for working with {@link SearchHit}.
@@ -75,9 +78,12 @@ public static Object unwrapSearchHits(Object result) {
 			return unwrapSearchHits(searchHits.getSearchHits());
 		}
 
-		if (result instanceof Flux) {
-			Flux<?> flux = (Flux<?>) result;
-			return flux.map(SearchHitSupport::unwrapSearchHits);
+		if (ReactiveSupport.isReactorAvailable()) {
+
+			if (result instanceof Flux) {
+				Flux<?> flux = (Flux<?>) result;
+				return flux.map(SearchHitSupport::unwrapSearchHits);
+			}
 		}
 
 		return result;
@@ -94,4 +100,28 @@ public static <T> AggregatedPage<SearchHit<T>> page(SearchHits<T> searchHits, Pa
 		return new AggregatedPageImpl<>(searchHits.getSearchHits(), pageable, searchHits.getTotalHits(),
 				searchHits.getAggregations(), searchHits.getScrollId(), searchHits.getMaxScore());
 	}
+
+	public static <T> SearchPage<T> searchPageFor(SearchHits<T> searchHits, @Nullable Pageable pageable) {
+		return new SearchPageImpl<>(searchHits, (pageable != null) ? pageable : Pageable.unpaged());
+	}
+
+	/**
+	 * SearchPage implementation.
+	 * 
+	 * @param <T>
+	 */
+	static class SearchPageImpl<T> extends PageImpl<SearchHit<T>> implements SearchPage<T> {
+
+		private final SearchHits<T> searchHits;
+
+		public SearchPageImpl(SearchHits<T> searchHits, Pageable pageable) {
+			super(searchHits.getSearchHits(), pageable, searchHits.getTotalHits());
+			this.searchHits = searchHits;
+		}
+
+		@Override
+		public SearchHits<T> getSearchHits() {
+			return searchHits;
+		}
+	}
 }

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

@@ -0,0 +1,28 @@
+/*
+ * 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.core;
+
+import org.springframework.data.domain.Page;
+
+/**
+ * Page definition for repositories that need to return a paged SearchHits.
+ *
+ * @author Peter-Josef Meisch
+ * @since 4.0
+ */
+public interface SearchPage<T> extends Page<SearchHit<T>> {
+	SearchHits<T> getSearchHits();
+}

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

@@ -86,7 +86,11 @@ public Object execute(Object[] parameters) {
 		} else if (queryMethod.isPageQuery()) {
 			query.setPageable(accessor.getPageable());
 			SearchHits<?> searchHits = elasticsearchOperations.search(query, clazz, index);
-			result = SearchHitSupport.page(searchHits, query.getPageable());
+			if (queryMethod.isSearchPageMethod()) {
+				result = SearchHitSupport.searchPageFor(searchHits, query.getPageable());
+			} else {
+				result = SearchHitSupport.page(searchHits, query.getPageable());
+			}
 		} else if (queryMethod.isStreamQuery()) {
 			if (accessor.getPageable().isUnpaged()) {
 				query.setPageable(PageRequest.of(0, DEFAULT_STREAM_BATCH_SIZE));

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

@@ -25,6 +25,7 @@
 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.SearchPage;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
 import org.springframework.data.elasticsearch.core.query.HighlightQuery;
@@ -171,6 +172,25 @@ public boolean isSearchHitMethod() {
 		return false;
 	}
 
+	/**
+	 * checks if the return type is {@link SearchPage}.
+	 * 
+	 * @since 4.0
+	 */
+	public boolean isSearchPageMethod() {
+		return SearchPage.class.isAssignableFrom(methodReturnType());
+	}
+
+	/**
+	 * retusn the declared return type for this method.
+	 * 
+	 * @return the return type
+	 * @since 4.0
+	 */
+	public Class<?> methodReturnType() {
+		return method.getReturnType();
+	}
+
 	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/support/ReactiveSupport.java

@@ -0,0 +1,39 @@
+/*
+ * 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.support;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.springframework.data.repository.util.ClassUtils;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.0
+ */
+public final class ReactiveSupport {
+	private ReactiveSupport() {}
+
+	/**
+	 * @return true if project reactor is on the classpath
+	 */
+	public static boolean isReactorAvailable() {
+		AtomicBoolean available = new AtomicBoolean(false);
+		ClassUtils.ifPresent("reactor.core.publisher.Flux", null, aClass -> {
+			available.set(true);
+		});
+		return available.get();
+	}
+}

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

@@ -53,6 +53,7 @@
 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.SearchPage;
 import org.springframework.data.elasticsearch.core.geo.GeoBox;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
 import org.springframework.data.elasticsearch.core.query.GeoDistanceOrder;
@@ -1479,6 +1480,23 @@ void shouldUseGeoSortParameter() {
 		assertThat(searchHits.getSearchHit(2).getId()).isEqualTo("oslo");
 	}
 
+	@Test // DATAES-749
+	void shouldReturnSearchPage() {
+		List<SampleEntity> entities = createSampleEntities("abc", 20);
+		repository.saveAll(entities);
+
+		// when
+		SearchPage<SampleEntity> searchPage = repository.searchByMessage("Message", PageRequest.of(0, 10));
+
+		assertThat(searchPage).isNotNull();
+		SearchHits<SampleEntity> searchHits = searchPage.getSearchHits();
+		assertThat(searchHits).isNotNull();
+		assertThat((searchHits.getTotalHits())).isEqualTo(20);
+		assertThat(searchHits.getSearchHits()).hasSize(10);
+		Pageable nextPageable = searchPage.nextPageable();
+		assertThat((nextPageable.getPageNumber())).isEqualTo(1);
+	}
+
 	private List<SampleEntity> createSampleEntities(String type, int numberOfEntities) {
 
 		List<SampleEntity> entities = new ArrayList<>();
@@ -1627,6 +1645,8 @@ public interface SampleCustomMethodRepository extends ElasticsearchRepository<Sa
 		Stream<SearchHit<SampleEntity>> readByMessage(String message);
 
 		SearchHits<SampleEntity> searchBy(Sort sort);
+
+		SearchPage<SampleEntity> searchByMessage(String message, Pageable pageable);
 	}
 
 	/**