Introduce Scroll API.

See: #2151
Original Pull Request: #2787

src/main/java/org/springframework/data/domain/KeysetScrollPosition.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2023 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.domain;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * A {@link ScrollPosition} based on the last seen keyset. Keyset scrolling must be associated with a {@link Sort
+ * well-defined sort} to be able to extract the keyset when resuming scrolling within the sorted result set.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+public final class KeysetScrollPosition implements ScrollPosition {
+
+	private static final KeysetScrollPosition initial = new KeysetScrollPosition(Collections.emptyMap(),
+			Direction.Forward);
+
+	private final Map<String, Object> keys;
+
+	private final Direction direction;
+
+	private KeysetScrollPosition(Map<String, Object> keys, Direction direction) {
+		this.keys = keys;
+		this.direction = direction;
+	}
+
+	/**
+	 * Creates a new initial {@link KeysetScrollPosition} to start scrolling using keyset-queries.
+	 *
+	 * @return a new initial {@link KeysetScrollPosition} to start scrolling using keyset-queries.
+	 */
+	public static KeysetScrollPosition initial() {
+		return initial;
+	}
+
+	/**
+	 * Creates a new {@link KeysetScrollPosition} from a keyset.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @return a new {@link KeysetScrollPosition} for the given keyset.
+	 */
+	public static KeysetScrollPosition of(Map<String, ?> keys) {
+		return of(keys, Direction.Forward);
+	}
+
+	/**
+	 * Creates a new {@link KeysetScrollPosition} from a keyset and {@link Direction}.
+	 *
+	 * @param keys must not be {@literal null}.
+	 * @param direction must not be {@literal null}.
+	 * @return a new {@link KeysetScrollPosition} for the given keyset and {@link Direction}.
+	 */
+	public static KeysetScrollPosition of(Map<String, ?> keys, Direction direction) {
+
+		Assert.notNull(keys, "Keys must not be null");
+		Assert.notNull(direction, "Direction must not be null");
+
+		if (keys.isEmpty()) {
+			return initial();
+		}
+
+		return new KeysetScrollPosition(Collections.unmodifiableMap(new LinkedHashMap<>(keys)), direction);
+	}
+
+	@Override
+	public boolean isInitial() {
+		return keys.isEmpty();
+	}
+
+	/**
+	 * @return the keyset.
+	 */
+	public Map<String, Object> getKeys() {
+		return keys;
+	}
+
+	/**
+	 * @return the scroll direction.
+	 */
+	public Direction getDirection() {
+		return direction;
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o)
+			return true;
+		if (o == null || getClass() != o.getClass())
+			return false;
+		KeysetScrollPosition that = (KeysetScrollPosition) o;
+		return ObjectUtils.nullSafeEquals(keys, that.keys) && direction == that.direction;
+	}
+
+	@Override
+	public int hashCode() {
+
+		int result = 17;
+
+		result += 31 * ObjectUtils.nullSafeHashCode(keys);
+		result += 31 * ObjectUtils.nullSafeHashCode(direction);
+
+		return result;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("KeysetScrollPosition [%s, %s]", direction, keys);
+	}
+
+	/**
+	 * Keyset scrolling direction.
+	 */
+	enum Direction {
+
+		/**
+		 * Forward (default) direction to scroll from the beginning of the results to their end.
+		 */
+		Forward,
+
+		/**
+		 * Backward direction to scroll from the end of the results to their beginning.
+		 */
+		Backward;
+	}
+}

src/main/java/org/springframework/data/domain/OffsetScrollPosition.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2023 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.domain;
+
+import java.util.function.IntFunction;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * A {@link ScrollPosition} based on the offsets within query results.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+public final class OffsetScrollPosition implements ScrollPosition {
+
+	private static final OffsetScrollPosition initial = new OffsetScrollPosition(0);
+
+	private final long offset;
+
+	private OffsetScrollPosition(long offset) {
+		this.offset = offset;
+	}
+
+	/**
+	 * Creates a new initial {@link OffsetScrollPosition} to start scrolling using offset/limit.
+	 *
+	 * @return a new initial {@link OffsetScrollPosition} to start scrolling using offset/limit.
+	 */
+	public static OffsetScrollPosition initial() {
+		return initial;
+	}
+
+	/**
+	 * Creates a new {@link OffsetScrollPosition} from an {@code offset}.
+	 *
+	 * @param offset
+	 * @return a new {@link OffsetScrollPosition} with the given {@code offset}.
+	 */
+	public static OffsetScrollPosition of(long offset) {
+
+		if (offset == 0) {
+			return initial();
+		}
+
+		return new OffsetScrollPosition(offset);
+	}
+
+	/**
+	 * Returns the {@link IntFunction position function} to calculate.
+	 *
+	 * @param startOffset the start offset to be used. Must not be negative.
+	 * @return the offset-based position function.
+	 */
+	public static IntFunction<OffsetScrollPosition> positionFunction(long startOffset) {
+
+		Assert.isTrue(startOffset >= 0, "Start offset must not be negative");
+
+		return startOffset == 0 ? OffsetPositionFunction.ZERO : new OffsetPositionFunction(startOffset);
+	}
+
+	@Override
+	public boolean isInitial() {
+		return offset == 0;
+	}
+
+	/**
+	 * @return the offset.
+	 */
+	public long getOffset() {
+		return offset;
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o)
+			return true;
+		if (o == null || getClass() != o.getClass())
+			return false;
+		OffsetScrollPosition that = (OffsetScrollPosition) o;
+		return offset == that.offset;
+	}
+
+	@Override
+	public int hashCode() {
+
+		int result = 17;
+
+		result += 31 * ObjectUtils.nullSafeHashCode(offset);
+
+		return result;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("OffsetScrollPosition [%s]", offset);
+	}
+
+	private record OffsetPositionFunction(long startOffset) implements IntFunction<OffsetScrollPosition> {
+
+		static final OffsetPositionFunction ZERO = new OffsetPositionFunction(0);
+
+		@Override
+		public OffsetScrollPosition apply(int offset) {
+
+			if (offset < 0) {
+				throw new IndexOutOfBoundsException(offset);
+			}
+
+			return of(startOffset + offset + 1);
+		}
+	}
+}

src/main/java/org/springframework/data/domain/Pageable.java

@@ -161,4 +161,20 @@ public interface Pageable {
 		return isUnpaged() ? Optional.empty() : Optional.of(this);
 	}
 
+	/**
+	 * Returns an {@link OffsetScrollPosition} from this pageable if the page request {@link #isPaged() is paged}.
+	 *
+	 * @return
+	 * @throws IllegalStateException if the request is {@link #isUnpaged()}
+	 * @since 3.1
+	 */
+	default OffsetScrollPosition toScrollPosition() {
+
+		if (isUnpaged()) {
+			throw new IllegalStateException("Cannot create OffsetScrollPosition from an unpaged instance");
+		}
+
+		return OffsetScrollPosition.of(getOffset());
+	}
+
 }

src/main/java/org/springframework/data/domain/Scroll.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2023 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.domain;
+
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.function.Function;
+import java.util.function.IntFunction;
+
+import org.springframework.data.util.Streamable;
+
+/**
+ * A scroll of data consumed from an underlying query result. A scroll is similar to {@link Slice} in the sense that it
+ * contains a subset of the actual query results for easier consumption of large result sets. The scroll is less
+ * opinionated about the actual data retrieval, whether the query has used index/offset, keyset-based pagination or
+ * cursor resume tokens.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ * @see ScrollPosition
+ */
+public interface Scroll<T> extends Streamable<T> {
+
+	/**
+	 * Construct a {@link Scroll}.
+	 *
+	 * @param items the list of data.
+	 * @param positionFunction the list of data.
+	 * @return the {@link Scroll}.
+	 * @param <T>
+	 */
+	static <T> Scroll<T> from(List<T> items, IntFunction<? extends ScrollPosition> positionFunction) {
+		return new ScrollImpl<>(items, positionFunction, false);
+	}
+
+	/**
+	 * Construct a {@link Scroll}.
+	 *
+	 * @param items the list of data.
+	 * @param positionFunction the list of data.
+	 * @param hasNext
+	 * @return the {@link Scroll}.
+	 * @param <T>
+	 */
+	static <T> Scroll<T> from(List<T> items, IntFunction<? extends ScrollPosition> positionFunction, boolean hasNext) {
+		return new ScrollImpl<>(items, positionFunction, hasNext);
+	}
+
+	/**
+	 * Returns the number of elements in this scroll.
+	 *
+	 * @return the number of elements in this scroll.
+	 */
+	int size();
+
+	/**
+	 * Returns {@code true} if this scroll contains no elements.
+	 *
+	 * @return {@code true} if this scroll contains no elements
+	 */
+	boolean isEmpty();
+
+	/**
+	 * Returns the scroll content as {@link List}.
+	 *
+	 * @return
+	 */
+	List<T> getContent();
+
+	/**
+	 * Returns whether the current scroll is the last one.
+	 *
+	 * @return
+	 */
+	default boolean isLast() {
+		return !hasNext();
+	}
+
+	/**
+	 * Returns if there is a next scroll.
+	 *
+	 * @return if there is a next scroll window.
+	 */
+	boolean hasNext();
+
+	/**
+	 * Returns the {@link ScrollPosition} at {@code index}.
+	 *
+	 * @param index
+	 * @return
+	 * @throws IndexOutOfBoundsException if the index is out of range ({@code index < 0 || index >= size()}).
+	 */
+	ScrollPosition positionAt(int index);
+
+	/**
+	 * Returns the {@link ScrollPosition} for {@code object}.
+	 *
+	 * @param object
+	 * @return
+	 * @throws NoSuchElementException if the object is not part of the result.
+	 */
+	default ScrollPosition positionAt(T object) {
+
+		int index = getContent().indexOf(object);
+
+		if (index == -1) {
+			throw new NoSuchElementException();
+		}
+
+		return positionAt(index);
+	}
+
+	// TODO: First and last seem to conflict with first/last scroll or first/last position of elements.
+	// these methods should rather express the position of the first element within this scroll and the scroll position
+	// to be used to get the next Scroll.
+	/**
+	 * Returns the first {@link ScrollPosition} or throw {@link NoSuchElementException} if the list is empty.
+	 *
+	 * @return the first {@link ScrollPosition}.
+	 * @throws NoSuchElementException if this result is empty.
+	 */
+	default ScrollPosition firstPosition() {
+
+		if (size() == 0) {
+			throw new NoSuchElementException();
+		}
+
+		return positionAt(0);
+	}
+
+	/**
+	 * Returns the last {@link ScrollPosition} or throw {@link NoSuchElementException} if the list is empty.
+	 *
+	 * @return the last {@link ScrollPosition}.
+	 * @throws NoSuchElementException if this result is empty.
+	 */
+	default ScrollPosition lastPosition() {
+
+		if (size() == 0) {
+			throw new NoSuchElementException();
+		}
+
+		return positionAt(size() - 1);
+	}
+
+	/**
+	 * Returns a new {@link Scroll} with the content of the current one mapped by the given {@code converter}.
+	 *
+	 * @param converter must not be {@literal null}.
+	 * @return a new {@link Scroll} with the content of the current one mapped by the given {@code converter}.
+	 */
+	<U> Scroll<U> map(Function<? super T, ? extends U> converter);
+
+}

src/main/java/org/springframework/data/domain/ScrollImpl.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2023 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.domain;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Function;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+
+import org.jetbrains.annotations.NotNull;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Default {@link Scroll} implementation.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+class ScrollImpl<T> implements Scroll<T> {
+
+	private final List<T> items;
+	private final IntFunction<? extends ScrollPosition> positionFunction;
+
+	private final boolean hasNext;
+
+	ScrollImpl(List<T> items, IntFunction<? extends ScrollPosition> positionFunction, boolean hasNext) {
+
+		Assert.notNull(items, "List of items must not be null");
+		Assert.notNull(positionFunction, "Position function must not be null");
+
+		this.items = items;
+		this.positionFunction = positionFunction;
+		this.hasNext = hasNext;
+	}
+
+	@Override
+	public int size() {
+		return items.size();
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return items.isEmpty();
+	}
+
+	@Override
+	public List<T> getContent() {
+		return items;
+	}
+
+	@Override
+	public boolean hasNext() {
+		return hasNext;
+	}
+
+	@Override
+	public ScrollPosition positionAt(int index) {
+
+		if (index < 0 || index >= size()) {
+			throw new IndexOutOfBoundsException(index);
+		}
+
+		return positionFunction.apply(index);
+	}
+
+	@Override
+	public <U> Scroll<U> map(Function<? super T, ? extends U> converter) {
+
+		Assert.notNull(converter, "Function must not be null");
+
+		return new ScrollImpl<>(stream().map(converter).collect(Collectors.toList()), positionFunction, hasNext);
+	}
+
+	@NotNull
+	@Override
+	public Iterator<T> iterator() {
+		return items.iterator();
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o)
+			return true;
+		if (o == null || getClass() != o.getClass())
+			return false;
+		ScrollImpl<?> that = (ScrollImpl<?>) o;
+		return ObjectUtils.nullSafeEquals(items, that.items)
+				&& ObjectUtils.nullSafeEquals(positionFunction, that.positionFunction)
+				&& ObjectUtils.nullSafeEquals(hasNext, that.hasNext);
+	}
+
+	@Override
+	public int hashCode() {
+		int result = ObjectUtils.nullSafeHashCode(items);
+		result = 31 * result + ObjectUtils.nullSafeHashCode(positionFunction);
+		result = 31 * result + ObjectUtils.nullSafeHashCode(hasNext);
+		return result;
+	}
+
+	@Override
+	public String toString() {
+		return "Scroll " + items;
+	}
+}

src/main/java/org/springframework/data/domain/ScrollPosition.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2023 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.domain;
+
+/**
+ * Interface to specify a position within a total query result. Scroll positions are used to start scrolling from the
+ * beginning of a query result or to resume scrolling from a given position within the query result.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+public interface ScrollPosition {
+
+	/**
+	 * Returns whether the current scroll position is the initial one.
+	 *
+	 * @return
+	 */
+	boolean isInitial();
+}

src/main/java/org/springframework/data/repository/query/FluentQuery.java

@@ -26,6 +26,8 @@ import java.util.stream.Stream;
 
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Scroll;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.lang.Nullable;
 
@@ -44,10 +46,23 @@ public interface FluentQuery<T> {
 	 * @param sort the {@link Sort} specification to sort the results by, may be {@link Sort#unsorted()}, must not be
 	 *          {@literal null}.
 	 * @return a new instance of {@link FluentQuery}.
-	 * @throws IllegalArgumentException if resultType is {@code null}.
+	 * @throws IllegalArgumentException if {@code sort} is {@code null}.
 	 */
 	FluentQuery<T> sortBy(Sort sort);
 
+	/**
+	 * Define the query limit.
+	 *
+	 * @param limit the limit to apply to the query to limit results. Must not be negative.
+	 * @return a new instance of {@link FluentQuery}.
+	 * @throws IllegalArgumentException if {@code limit} is less than zero.
+	 * @throws UnsupportedOperationException if not supported by the underlying implementation.
+	 * @since 3.1
+	 */
+	default FluentQuery<T> limit(int limit) {
+		throw new UnsupportedOperationException("Limit not supported");
+	}
+
 	/**
 	 * Define the target type the result should be mapped to. Skip this step if you are only interested in the original
 	 * domain type.
@@ -55,7 +70,7 @@ public interface FluentQuery<T> {
 	 * @param resultType must not be {@code null}.
 	 * @param <R> result type.
 	 * @return a new instance of {@link FluentQuery}.
-	 * @throws IllegalArgumentException if resultType is {@code null}.
+	 * @throws IllegalArgumentException if {@code resultType} is {@code null}.
 	 */
 	<R> FluentQuery<R> as(Class<R> resultType);
 
@@ -64,7 +79,7 @@ public interface FluentQuery<T> {
 	 *
 	 * @param properties must not be {@code null}.
 	 * @return a new instance of {@link FluentQuery}.
-	 * @throws IllegalArgumentException if fields is {@code null}.
+	 * @throws IllegalArgumentException if {@code properties} is {@code null}.
 	 */
 	default FluentQuery<T> project(String... properties) {
 		return project(Arrays.asList(properties));
@@ -75,7 +90,7 @@ public interface FluentQuery<T> {
 	 *
 	 * @param properties must not be {@code null}.
 	 * @return a new instance of {@link FluentQuery}.
-	 * @throws IllegalArgumentException if fields is {@code null}.
+	 * @throws IllegalArgumentException if {@code properties} is {@code null}.
 	 */
 	FluentQuery<T> project(Collection<String> properties);
 
@@ -90,6 +105,11 @@ public interface FluentQuery<T> {
 		@Override
 		FetchableFluentQuery<T> sortBy(Sort sort);
 
+		@Override
+		default FetchableFluentQuery<T> limit(int limit) {
+			throw new UnsupportedOperationException("Limit not supported");
+		}
+
 		@Override
 		<R> FetchableFluentQuery<R> as(Class<R> resultType);
 
@@ -144,12 +164,27 @@ public interface FluentQuery<T> {
 		 */
 		List<T> all();
 
+		/**
+		 * Get all matching elements as {@link Scroll} to start result scrolling or resume scrolling at
+		 * {@code scrollPosition}.
+		 *
+		 * @param scrollPosition must not be {@literal null}.
+		 * @return
+		 * @throws IllegalArgumentException if {@code scrollPosition} is {@literal null}.
+		 * @throws UnsupportedOperationException if not supported by the underlying implementation.
+		 * @since 3.1
+		 */
+		default Scroll<T> scroll(ScrollPosition scrollPosition) {
+			throw new UnsupportedOperationException("Scrolling not supported");
+		}
+
 		/**
 		 * Get a page of matching elements for {@link Pageable}.
 		 *
 		 * @param pageable the pageable to request a paged result, can be {@link Pageable#unpaged()}, must not be
 		 *          {@literal null}. The given {@link Pageable} will override any previously specified {@link Sort sort} if
-		 *          the {@link Sort} object is not {@link Sort#isUnsorted()}.
+		 *          the {@link Sort} object is not {@link Sort#isUnsorted()}. Any potentially specified {@link #limit(int)}
+		 *          will be overridden by {@link Pageable#getPageSize()}.
 		 * @return
 		 */
 		Page<T> page(Pageable pageable);
@@ -187,6 +222,11 @@ public interface FluentQuery<T> {
 		@Override
 		ReactiveFluentQuery<T> sortBy(Sort sort);
 
+		@Override
+		default ReactiveFluentQuery<T> limit(int limit) {
+			throw new UnsupportedOperationException("Limit not supported");
+		}
+
 		@Override
 		<R> ReactiveFluentQuery<R> as(Class<R> resultType);
 
@@ -220,12 +260,27 @@ public interface FluentQuery<T> {
 		 */
 		Flux<T> all();
 
+		/**
+		 * Get all matching elements as {@link Scroll} to start result scrolling or resume scrolling at
+		 * {@code scrollPosition}.
+		 *
+		 * @param scrollPosition must not be {@literal null}.
+		 * @return
+		 * @throws IllegalArgumentException if {@code scrollPosition} is {@literal null}.
+		 * @throws UnsupportedOperationException if not supported by the underlying implementation.
+		 * @since 3.1
+		 */
+		default Mono<Scroll<T>> scroll(ScrollPosition scrollPosition) {
+			throw new UnsupportedOperationException("Scrolling not supported");
+		}
+
 		/**
 		 * Get a page of matching elements for {@link Pageable}.
 		 *
 		 * @param pageable the pageable to request a paged result, can be {@link Pageable#unpaged()}, must not be
 		 *          {@literal null}. The given {@link Pageable} will override any previously specified {@link Sort sort} if
-		 *          the {@link Sort} object is not {@link Sort#isUnsorted()}.
+		 *          the {@link Sort} object is not {@link Sort#isUnsorted()}. Any potentially specified {@link #limit(int)}
+		 *          will be overridden by {@link Pageable#getPageSize()}.
 		 * @return
 		 */
 		Mono<Page<T>> page(Pageable pageable);

src/main/java/org/springframework/data/repository/query/Parameter.java

@@ -26,6 +26,7 @@ import java.util.Optional;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.repository.util.ClassUtils;
 import org.springframework.data.repository.util.QueryExecutionConverters;
@@ -57,7 +58,7 @@ public class Parameter {
 
 	static {
 
-		List<Class<?>> types = new ArrayList<>(Arrays.asList(Pageable.class, Sort.class));
+		List<Class<?>> types = new ArrayList<>(Arrays.asList(ScrollPosition.class, Pageable.class, Sort.class));
 
 		// consider Kotlin Coroutines Continuation a special parameter. That parameter is synthetic and should not get
 		// bound to any query.
@@ -192,6 +193,16 @@ public class Parameter {
 		return format("%s:%s", isNamedParameter() ? getName() : "#" + getIndex(), getType().getName());
 	}
 
+	/**
+	 * Returns whether the {@link Parameter} is a {@link ScrollPosition} parameter.
+	 *
+	 * @return
+	 * @since 3.1
+	 */
+	boolean isScrollPosition() {
+		return ScrollPosition.class.isAssignableFrom(getType());
+	}
+
 	/**
 	 * Returns whether the {@link Parameter} is a {@link Pageable} parameter.
 	 *

src/main/java/org/springframework/data/repository/query/ParameterAccessor.java

@@ -18,6 +18,7 @@ package org.springframework.data.repository.query;
 import java.util.Iterator;
 
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.lang.Nullable;
 
@@ -29,6 +30,14 @@ import org.springframework.lang.Nullable;
  */
 public interface ParameterAccessor extends Iterable<Object> {
 
+	/**
+	 * Returns the {@link ScrollPosition} of the parameters, if available. Returns {@code null} otherwise.
+	 *
+	 * @return
+	 */
+	@Nullable
+	ScrollPosition getScrollPosition();
+
 	/**
 	 * Returns the {@link Pageable} of the parameters, if available. Returns {@link Pageable#unpaged()} otherwise.
 	 *

src/main/java/org/springframework/data/repository/query/Parameters.java

@@ -28,6 +28,7 @@ import org.springframework.core.DefaultParameterNameDiscoverer;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ParameterNameDiscoverer;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.util.Lazy;
 import org.springframework.data.util.Streamable;
@@ -42,7 +43,7 @@ import org.springframework.util.Assert;
  */
 public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter> implements Streamable<T> {
 
-	public static final List<Class<?>> TYPES = Arrays.asList(Pageable.class, Sort.class);
+	public static final List<Class<?>> TYPES = Arrays.asList(ScrollPosition.class, Pageable.class, Sort.class);
 
 	private static final String PARAM_ON_SPECIAL = format("You must not use @%s on a parameter typed %s or %s",
 			Param.class.getSimpleName(), Pageable.class.getSimpleName(), Sort.class.getSimpleName());
@@ -52,6 +53,7 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 
 	private static final ParameterNameDiscoverer PARAMETER_NAME_DISCOVERER = new DefaultParameterNameDiscoverer();
 
+	private final int scrollPositionIndex;
 	private final int pageableIndex;
 	private final int sortIndex;
 	private final List<T> parameters;
@@ -91,6 +93,7 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 		this.parameters = new ArrayList<>(parameterCount);
 		this.dynamicProjectionIndex = -1;
 
+		int scrollPositionIndex = -1;
 		int pageableIndex = -1;
 		int sortIndex = -1;
 
@@ -111,6 +114,10 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 				this.dynamicProjectionIndex = parameter.getIndex();
 			}
 
+			if (ScrollPosition.class.isAssignableFrom(parameter.getType())) {
+				scrollPositionIndex = i;
+			}
+
 			if (Pageable.class.isAssignableFrom(parameter.getType())) {
 				pageableIndex = i;
 			}
@@ -122,6 +129,7 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 			parameters.add(parameter);
 		}
 
+		this.scrollPositionIndex = scrollPositionIndex;
 		this.pageableIndex = pageableIndex;
 		this.sortIndex = sortIndex;
 		this.bindable = Lazy.of(this::getBindable);
@@ -138,6 +146,7 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 
 		this.parameters = new ArrayList<>(originals.size());
 
+		int scrollPositionIndexTemp = -1;
 		int pageableIndexTemp = -1;
 		int sortIndexTemp = -1;
 		int dynamicProjectionTemp = -1;
@@ -147,11 +156,13 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 			T original = originals.get(i);
 			this.parameters.add(original);
 
+			scrollPositionIndexTemp = original.isScrollPosition() ? i : -1;
 			pageableIndexTemp = original.isPageable() ? i : -1;
 			sortIndexTemp = original.isSort() ? i : -1;
 			dynamicProjectionTemp = original.isDynamicProjectionParameter() ? i : -1;
 		}
 
+		this.scrollPositionIndex = scrollPositionIndexTemp;
 		this.pageableIndex = pageableIndexTemp;
 		this.sortIndex = sortIndexTemp;
 		this.dynamicProjectionIndex = dynamicProjectionTemp;
@@ -185,6 +196,27 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 		return (T) new Parameter(parameter);
 	}
 
+	/**
+	 * Returns whether the method the {@link Parameters} was created for contains a {@link ScrollPosition} argument.
+	 *
+	 * @return
+	 * @since 3.1
+	 */
+	public boolean hasScrollPositionParameter() {
+		return scrollPositionIndex != -1;
+	}
+
+	/**
+	 * Returns the index of the {@link ScrollPosition} {@link Method} parameter if available. Will return {@literal -1} if
+	 * there is no {@link ScrollPosition} argument in the {@link Method}'s parameter list.
+	 *
+	 * @return the scrollPositionIndex
+	 * @since 3.1
+	 */
+	public int getScrollPositionIndex() {
+		return scrollPositionIndex;
+	}
+
 	/**
 	 * Returns whether the method the {@link Parameters} was created for contains a {@link Pageable} argument.
 	 *
@@ -195,8 +227,8 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 	}
 
 	/**
-	 * Returns the index of the {@link Pageable} {@link Method} parameter if available. Will return {@literal -1} if
-	 * there is no {@link Pageable} argument in the {@link Method}'s parameter list.
+	 * Returns the index of the {@link Pageable} {@link Method} parameter if available. Will return {@literal -1} if there
+	 * is no {@link Pageable} argument in the {@link Method}'s parameter list.
 	 *
 	 * @return the pageableIndex
 	 */
@@ -205,8 +237,8 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 	}
 
 	/**
-	 * Returns the index of the {@link Sort} {@link Method} parameter if available. Will return {@literal -1} if there
-	 * is no {@link Sort} argument in the {@link Method}'s parameter list.
+	 * Returns the index of the {@link Sort} {@link Method} parameter if available. Will return {@literal -1} if there is
+	 * no {@link Sort} argument in the {@link Method}'s parameter list.
 	 *
 	 * @return
 	 */
@@ -288,7 +320,7 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 	 * @return
 	 */
 	public boolean hasSpecialParameter() {
-		return hasSortParameter() || hasPageableParameter();
+		return hasScrollPositionParameter() || hasSortParameter() || hasPageableParameter();
 	}
 
 	/**
@@ -315,8 +347,8 @@ public abstract class Parameters<S extends Parameters<S, T>, T extends Parameter
 
 	/**
 	 * Returns a bindable parameter with the given index. So for a method with a signature of
-	 * {@code (Pageable pageable, String name)} a call to {@code #getBindableParameter(0)} will return the
-	 * {@link String} parameter.
+	 * {@code (Pageable pageable, String name)} a call to {@code #getBindableParameter(0)} will return the {@link String}
+	 * parameter.
 	 *
 	 * @param bindableIndex
 	 * @return

src/main/java/org/springframework/data/repository/query/ParametersParameterAccessor.java

@@ -18,6 +18,7 @@ package org.springframework.data.repository.query;
 import java.util.Iterator;
 
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.repository.util.QueryExecutionConverters;
 import org.springframework.data.repository.util.ReactiveWrapperConverters;
@@ -91,6 +92,22 @@ public class ParametersParameterAccessor implements ParameterAccessor {
 		return this.values;
 	}
 
+	@Override
+	public ScrollPosition getScrollPosition() {
+
+		if (!parameters.hasScrollPositionParameter()) {
+
+			Pageable pageable = getPageable();
+			if (pageable.isPaged()) {
+				return pageable.toScrollPosition();
+			}
+
+			return null;
+		}
+
+		return (ScrollPosition) values[parameters.getScrollPositionIndex()];
+	}
+
 	@Override
 	public Pageable getPageable() {
 

src/main/java/org/springframework/data/repository/query/QueryMethod.java

@@ -18,11 +18,14 @@ package org.springframework.data.repository.query;
 import static org.springframework.data.repository.util.ClassUtils.*;
 
 import java.lang.reflect.Method;
+import java.util.Collections;
 import java.util.Set;
 import java.util.stream.Stream;
 
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Scroll;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.projection.ProjectionFactory;
@@ -31,6 +34,7 @@ import org.springframework.data.repository.core.RepositoryMetadata;
 import org.springframework.data.repository.util.QueryExecutionConverters;
 import org.springframework.data.repository.util.ReactiveWrapperConverters;
 import org.springframework.data.util.Lazy;
+import org.springframework.data.util.ReactiveWrappers;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
 
@@ -92,6 +96,10 @@ public class QueryMethod {
 			}
 		}
 
+		if (hasParameterOfType(method, ScrollPosition.class)) {
+			assertReturnTypeAssignable(method, Collections.singleton(Scroll.class));
+		}
+
 		Assert.notNull(this.parameters,
 				() -> String.format("Parameters extracted from method '%s' must not be null", method.getName()));
 
@@ -100,6 +108,12 @@ public class QueryMethod {
 					String.format("Paging query needs to have a Pageable parameter; Offending method: %s", method));
 		}
 
+		if (isScrollQuery()) {
+
+			Assert.isTrue(this.parameters.hasScrollPositionParameter() || this.parameters.hasPageableParameter(),
+					String.format("Scroll query needs to have a ScrollPosition parameter; Offending method: %s", method));
+		}
+
 		this.domainClass = Lazy.of(() -> {
 
 			Class<?> repositoryDomainClass = metadata.getDomainType();
@@ -188,6 +202,16 @@ public class QueryMethod {
 		return isCollectionQuery.get();
 	}
 
+	/**
+	 * Returns whether the query method will return a {@link Scroll}.
+	 *
+	 * @return
+	 * @since 3.1
+	 */
+	public boolean isScrollQuery() {
+		return org.springframework.util.ClassUtils.isAssignable(Scroll.class, unwrappedReturnType);
+	}
+
 	/**
 	 * Returns whether the query method will return a {@link Slice}.
 	 *
@@ -268,7 +292,7 @@ public class QueryMethod {
 
 	private boolean calculateIsCollectionQuery() {
 
-		if (isPageQuery() || isSliceQuery()) {
+		if (isPageQuery() || isSliceQuery() || isScrollQuery()) {
 			return false;
 		}
 
@@ -314,6 +338,10 @@ public class QueryMethod {
 		// TODO: to resolve generics fully we'd need the actual repository interface here
 		TypeInformation<?> returnType = TypeInformation.fromReturnTypeOf(method);
 
+		returnType = ReactiveWrappers.isSingleValueType(returnType.getType()) //
+				? returnType.getRequiredComponentType() //
+				: returnType;
+
 		returnType = QueryExecutionConverters.isSingleValue(returnType.getType()) //
 				? returnType.getRequiredComponentType() //
 				: returnType;
@@ -324,6 +352,6 @@ public class QueryMethod {
 			}
 		}
 
-		throw new IllegalStateException("Method has to have one of the following return types" + types);
+		throw new IllegalStateException("Method has to have one of the following return types " + types);
 	}
 }

src/main/java/org/springframework/data/repository/query/ResultProcessor.java

@@ -26,6 +26,7 @@ import org.springframework.core.CollectionFactory;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.data.domain.Scroll;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.repository.util.ReactiveWrapperConverters;
@@ -143,6 +144,10 @@ public class ResultProcessor {
 
 		ChainingConverter converter = ChainingConverter.of(type.getReturnedType(), preparingConverter).and(this.converter);
 
+		if (source instanceof Scroll<?> && method.isScrollQuery()) {
+			return (T) ((Scroll<?>) source).map(converter::convert);
+		}
+
 		if (source instanceof Slice && (method.isPageQuery() || method.isSliceQuery())) {
 			return (T) ((Slice<?>) source).map(converter::convert);
 		}
@@ -163,7 +168,7 @@ public class ResultProcessor {
 		}
 
 		if (ReactiveWrapperConverters.supports(source.getClass())) {
-			return (T) ReactiveWrapperConverters.map(source, converter::convert);
+			return (T) ReactiveWrapperConverters.map(source, it -> processResult(it, preparingConverter));
 		}
 
 		return (T) converter.convert(source);

src/main/java/org/springframework/data/repository/util/QueryExecutionConverters.java

@@ -36,6 +36,7 @@ import org.springframework.core.convert.converter.GenericConverter;
 import org.springframework.core.convert.support.ConfigurableConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Scroll;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.util.CustomCollections;
@@ -97,6 +98,7 @@ public abstract class QueryExecutionConverters {
 		ALLOWED_PAGEABLE_TYPES.add(Slice.class);
 		ALLOWED_PAGEABLE_TYPES.add(Page.class);
 		ALLOWED_PAGEABLE_TYPES.add(List.class);
+		ALLOWED_PAGEABLE_TYPES.add(Scroll.class);
 
 		WRAPPER_TYPES.add(NullableWrapperToCompletableFutureConverter.getWrapperType());
 

src/main/java/org/springframework/data/util/CustomCollections.java

@@ -223,7 +223,7 @@ public class CustomCollections {
 		/**
 		 * Returns whether the current's raw type is one of the given ones.
 		 *
-		 * @param candidates must not be {@literal null}.
+		 * @param type must not be {@literal null}.
 		 * @return
 		 */
 		public boolean has(Class<?> type) {