Add support for reactive CursorWindow query methods.

Author: mp911de

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java

@@ -128,7 +128,9 @@ default Optional<T> first() {
 		/**
 		 * Return a window from a {@link CursorRequest}.
 		 *
+		 * @param cursorRequest the cursor request.
 		 * @return a window from the resulting elements.
+		 * @since 4.1
 		 */
 		CursorWindow<T> window(CursorRequest cursorRequest);
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -855,8 +855,6 @@ public <T> List<T> find(Query query, Class<T> entityClass, String collectionName
 	@Override
 	public <T> CursorWindow<T> findWindow(CursorRequest cursorRequest, Query query, Class<T> entityType) {
 
-		Assert.notNull(cursorRequest, "CursorRequest must not be null");
-		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(entityType, "Entity type must not be null");
 
 		return findWindow(cursorRequest, query, entityType, getCollectionName(entityType));
@@ -867,7 +865,6 @@ public <T> CursorWindow<T> findWindow(CursorRequest cursorRequest, Query query,
 	// CursorRequest is similar to Pageable in the sense of being required to define a sort order
 	public <T> CursorWindow<T> findWindow(CursorRequest cursorRequest, Query query, Class<T> entityType,
 			String collectionName) {
-
 		return doFindWindow(cursorRequest, query, entityType, entityType, collectionName);
 	}
 
@@ -881,6 +878,7 @@ <T> CursorWindow<T> doFindWindow(CursorRequest cursorRequest, Query query, Class
 		Assert.notNull(targetClass, "Target type must not be null");
 
 		ReadDocumentCallback<T> callback = new ReadDocumentCallback<>(mongoConverter, targetClass, collectionName);
+
 		if (cursorRequest instanceof OffsetCursorRequest offset) {
 
 			int limit = offset.getSize() + 1;

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperation.java

@@ -18,6 +18,8 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.data.domain.CursorRequest;
+import org.springframework.data.domain.CursorWindow;
 import org.springframework.data.geo.GeoResult;
 import org.springframework.data.mongodb.core.query.CriteriaDefinition;
 import org.springframework.data.mongodb.core.query.NearQuery;
@@ -87,6 +89,15 @@ interface TerminatingFind<T> {
 		 */
 		Flux<T> all();
 
+		/**
+		 * Return a window from a {@link CursorRequest}.
+		 *
+		 * @param cursorRequest the cursor request.
+		 * @return a window from the resulting elements.
+		 * @since 4.1
+		 */
+		Mono<CursorWindow<T>> window(CursorRequest cursorRequest);
+
 		/**
 		 * Get all matching elements using a {@link com.mongodb.CursorType#TailableAwait tailable cursor}. The stream will
 		 * not be completed unless the {@link org.reactivestreams.Subscription} is

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveFindOperationSupport.java

@@ -20,6 +20,8 @@
 
 import org.bson.Document;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.domain.CursorRequest;
+import org.springframework.data.domain.CursorWindow;
 import org.springframework.data.mongodb.core.CollectionPreparerSupport.ReactiveCollectionPreparerDelegate;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
@@ -137,6 +139,11 @@ public Flux<T> all() {
 			return doFind(null);
 		}
 
+		@Override
+		public Mono<CursorWindow<T>> window(CursorRequest cursorRequest) {
+			return template.doFindWindow(cursorRequest, query, domainType, returnType, getCollectionName());
+		}
+
 		@Override
 		public Flux<T> tail() {
 			return doFind(template.new TailingQueryFindPublisherPreparer(query, domainType));

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -834,8 +834,6 @@ public <T> Flux<T> find(@Nullable Query query, Class<T> entityClass, String coll
 	@Override
 	public <T> Mono<CursorWindow<T>> findWindow(CursorRequest cursorRequest, Query query, Class<T> entityType) {
 
-		Assert.notNull(cursorRequest, "CursorRequest must not be null");
-		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(entityType, "Entity type must not be null");
 
 		return findWindow(cursorRequest, query, entityType, getCollectionName(entityType));
@@ -846,19 +844,25 @@ public <T> Mono<CursorWindow<T>> findWindow(CursorRequest cursorRequest, Query q
 	// CursorRequest is similar to Pageable in the sense of being required to define a sort order
 	public <T> Mono<CursorWindow<T>> findWindow(CursorRequest cursorRequest, Query query, Class<T> entityType,
 			String collectionName) {
+		return doFindWindow(cursorRequest, query, entityType, entityType, collectionName);
+	}
+
+	<T> Mono<CursorWindow<T>> doFindWindow(CursorRequest cursorRequest, Query query, Class<?> sourceClass,
+			Class<T> targetClass, String collectionName) {
 
 		Assert.notNull(cursorRequest, "CursorRequest must not be null");
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(collectionName, "CollectionName must not be null");
-		Assert.notNull(entityType, "Entity type must not be null");
+		Assert.notNull(sourceClass, "Entity type must not be null");
+		Assert.notNull(targetClass, "Target type must not be null");
 
 		if (cursorRequest instanceof OffsetCursorRequest offset) {
 
 			int limit = offset.getSize() + 1;
 
 			Mono<List<T>> result = doFind(collectionName, ReactiveCollectionPreparerDelegate.of(query),
-					query.getQueryObject(), query.getFieldsObject(), entityType, new QueryFindPublisherPreparer(query,
-							new Query().with(cursorRequest.getSort()).getSortObject(), limit, offset.getOffset(), entityType))
+					query.getQueryObject(), query.getFieldsObject(), targetClass, new QueryFindPublisherPreparer(query,
+							new Query().with(cursorRequest.getSort()).getSortObject(), limit, offset.getOffset(), sourceClass))
 									.collectList();
 
 			return result.map(it -> CursorUtils.createWindow(offset, it));
@@ -869,11 +873,11 @@ public <T> Mono<CursorWindow<T>> findWindow(CursorRequest cursorRequest, Query q
 			int limit = keyset.getSize() + 1;
 
 			KeySetCursorQuery keysetPaginationQuery = CursorUtils.createKeysetPaginationQuery(query, keyset,
-					operations.getIdPropertyName(entityType));
+					operations.getIdPropertyName(sourceClass));
 
 			Mono<List<T>> result = doFind(collectionName, ReactiveCollectionPreparerDelegate.of(query),
-					keysetPaginationQuery.query(), keysetPaginationQuery.fields(), entityType,
-					new QueryFindPublisherPreparer(query, keysetPaginationQuery.sort(), limit, 0, entityType)).collectList();
+					keysetPaginationQuery.query(), keysetPaginationQuery.fields(), targetClass,
+					new QueryFindPublisherPreparer(query, keysetPaginationQuery.sort(), limit, 0, sourceClass)).collectList();
 
 			return result.map(it -> CursorUtils.createWindow(keyset, it, operations));
 		}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractReactiveMongoQuery.java

@@ -203,6 +203,8 @@ private ReactiveMongoQueryExecution getExecutionToWrap(MongoParameterAccessor ac
 			return (q, t, c) -> operation.matching(q.with(accessor.getPageable())).tail();
 		} else if (method.isCollectionQuery()) {
 			return (q, t, c) -> operation.matching(q.with(accessor.getPageable())).all();
+		} else if (method.isCursorWindowQuery()) {
+			return (q, t, c) -> operation.matching(q).window(accessor.getCursorRequest());
 		} else if (isCountQuery()) {
 			return (q, t, c) -> operation.matching(q).count();
 		} else if (isExistsQuery()) {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ReactiveMongoRepositoryTests.java

@@ -34,17 +34,21 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.reactivestreams.Publisher;
-
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.domain.CursorRequest;
+import org.springframework.data.domain.CursorWindow;
+import org.springframework.data.domain.KeysetCursorRequest;
+import org.springframework.data.domain.OffsetCursorRequest;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
@@ -285,6 +289,39 @@ void shouldUseTailableCursorWithDtoProjection() {
 		cappedRepository.findDtoProjectionByKey("value").as(StepVerifier::create).expectNextCount(1).thenCancel().verify();
 	}
 
+	@Test // GH-4308
+	void appliesCursorWindowingCorrectly() {
+
+		CursorWindow<Person> cursorWindow = repository
+				.findByLastnameLike("*", KeysetCursorRequest.ofSize(2, Sort.by("firstname", "lastname"))).block();
+
+		assertThat(cursorWindow).hasSize(2);
+		assertThat(cursorWindow).containsSequence(alicia, boyd);
+		assertThat(cursorWindow.isFirst()).isTrue();
+		assertThat(cursorWindow.isLast()).isFalse();
+
+		CursorWindow<Person> nextCursorWindow = repository.findByLastnameLike("*", cursorWindow.nextCursorRequest())
+				.block();
+
+		assertThat(nextCursorWindow).hasSize(2);
+		assertThat(nextCursorWindow).containsSequence(carter, dave);
+		assertThat(nextCursorWindow.isFirst()).isFalse();
+		assertThat(nextCursorWindow.isLast()).isFalse();
+	}
+
+	@Test // GH-4308
+	void appliesCursorWindowingWithProjectionCorrectly() {
+
+		repository
+				.findCursorProjectionByLastnameLike("*",
+						OffsetCursorRequest.ofSize(2, Sort.by(Direction.ASC, "firstname", "lastname"))) //
+				.flatMapIterable(Function.identity()) //
+				.as(StepVerifier::create) //
+				.expectNext(new PersonSummaryDto(alicia.getFirstname(), alicia.getLastname())) //
+				.expectNextCount(1) //
+				.verifyComplete();
+	}
+
 	@Test // DATAMONGO-1444
 	@DirtiesState
 	void findsPeopleByLocationWithinCircle() {
@@ -712,6 +749,11 @@ interface ReactivePersonRepository
 		@Query("{ lastname: { $in: ?0 }, age: { $gt : ?1 } }")
 		Flux<Person> findStringQuery(Flux<String> lastname, Mono<Integer> age);
 
+		Mono<CursorWindow<Person>> findByLastnameLike(String lastname, CursorRequest cursorRequest);
+
+		Mono<CursorWindow<PersonSummaryDto>> findCursorProjectionByLastnameLike(String lastname,
+				OffsetCursorRequest cursorRequest);
+
 		Flux<Person> findByLocationWithin(Circle circle);
 
 		Flux<Person> findByLocationWithin(Circle circle, Pageable pageable);

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StubParameterAccessor.java

@@ -18,6 +18,7 @@
 import java.util.Arrays;
 import java.util.Iterator;
 
+import org.springframework.data.domain.CursorRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Range;
 import org.springframework.data.domain.Range.Bound;
@@ -72,6 +73,11 @@ public StubParameterAccessor(Object... values) {
 		}
 	}
 
+	@Override
+	public CursorRequest getCursorRequest() {
+		return null;
+	}
+
 	public Pageable getPageable() {
 		return null;
 	}