Add overload to ReactiveSortingRepository accepting a Limit parameter.

src/main/java/org/springframework/data/repository/reactive/ReactiveSortingRepository.java

@@ -18,18 +18,19 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.data.domain.Limit;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.repository.NoRepositoryBean;
 import org.springframework.data.repository.Repository;
 
 /**
- * Repository fragment to provide methods to retrieve entities using the sorting abstraction. In many cases it
- * should be combined with {@link ReactiveCrudRepository} or a similar repository interface in order to add CRUD
- * functionality.
+ * Repository fragment to provide methods to retrieve entities using the sorting abstraction. In many cases it should be
+ * combined with {@link ReactiveCrudRepository} or a similar repository interface in order to add CRUD functionality.
  *
  * @author Mark Paluch
  * @author Christoph Strobl
  * @author Jens Schauder
+ * @author YongHwan Kwon
  * @since 2.0
  * @see Sort
  * @see Mono
@@ -48,4 +49,39 @@ public interface ReactiveSortingRepository<T, ID> extends Repository<T, ID> {
 	 * @throws IllegalArgumentException in case the given {@link Sort} is {@literal null}.
 	 */
 	Flux<T> findAll(Sort sort);
+
+	/**
+	 * Returns all entities sorted by the given options with limit.
+	 *
+	 * @param sort the {@link Sort} specification to sort the results by, can be {@link Sort#unsorted()}, must not be
+	 *          {@literal null}.
+	 * @param limit the {@link Limit} to limit the results, must not be {@literal null} or negative.
+	 * @return all entities sorted by the given options.
+	 * @throws IllegalArgumentException in case the given {@link Sort} is {@literal null} or {@link Limit} is
+	 *           {@literal null} or negative.
+	 */
+	default Flux<T> findAll(Sort sort, Limit limit) {
+		if (limit == null) {
+			throw new IllegalArgumentException("Limit must not be null");
+		}
+
+		if (limit.isUnlimited()) {
+			return findAll(sort);
+		}
+
+		final int maxLimit = limit.max();
+		if (maxLimit < 0) {
+			throw new IllegalArgumentException("Limit value cannot be negative");
+		}
+
+		if (maxLimit == 0) {
+			throw new IllegalArgumentException("Limit value cannot be zero");
+		}
+
+		if (maxLimit == Integer.MAX_VALUE) {
+			return findAll(sort);
+		}
+
+		return findAll(sort).take(maxLimit);
+	}
 }

src/test/java/org/springframework/data/repository/reactive/ReactiveSortingRepositoryTest.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2014-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.repository.reactive;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import java.util.stream.IntStream;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Spy;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.data.domain.Limit;
+import org.springframework.data.domain.Sort;
+
+/**
+ * Unit tests for {@link ReactiveSortingRepository}.
+ *
+ * @author YongHwan Kwon
+ */
+@ExtendWith(MockitoExtension.class)
+class ReactiveSortingRepositoryTest {
+
+	@Spy ReactiveSortingRepository<Integer, Integer> repository;
+
+	@DisplayName("Entity fetching tests")
+	@Nested
+	class EntityFetchingTests {
+
+		@DisplayName("Return a single entity for a limit of one")
+		@Test
+		void shouldReturnOnlyOneEntityForLimitOfOne() {
+			when(repository.findAll(Sort.unsorted())).thenReturn(Flux.range(1, 20));
+
+			final Flux<Integer> results = repository.findAll(Sort.unsorted(), Limit.of(1));
+
+			StepVerifier.create(results)
+					.expectNext(1)
+					.expectComplete()
+					.verify();
+
+
+			verify(repository).findAll(Sort.unsorted());
+		}
+
+		@DisplayName("Return ten entities for a limit of ten")
+		@Test
+		void shouldReturnTenEntitiesForLimitOfTen() {
+			when(repository.findAll(Sort.unsorted())).thenReturn(Flux.range(1, 20));
+
+			final Flux<Integer> results = repository.findAll(Sort.unsorted(), Limit.of(10));
+
+			StepVerifier.create(results)
+					.expectNext(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+					.expectComplete()
+					.verify();
+
+			verify(repository).findAll(Sort.unsorted());
+		}
+
+		@DisplayName("Return all entities with unlimited")
+		@Test
+		void shouldReturnAllEntitiesForUnlimited() {
+			when(repository.findAll(Sort.unsorted())).thenReturn(Flux.range(1, 20));
+
+			final Flux<Integer> results = repository.findAll(Sort.unsorted(), Limit.unlimited());
+
+			StepVerifier.create(results)
+					.expectNextSequence(IntStream.rangeClosed(1, 20).boxed().toList())
+					.expectComplete()
+					.verify();
+
+			verify(repository).findAll(Sort.unsorted());
+		}
+
+		@DisplayName("Return all entities with maximum integer limit")
+		@Test
+		void shouldReturnAllEntitiesForLimitOfMaxValue() {
+			when(repository.findAll(Sort.unsorted())).thenReturn(Flux.range(1, 20));
+
+			final Flux<Integer> results = repository.findAll(Sort.unsorted(), Limit.unlimited());
+
+			StepVerifier.create(results)
+					.expectNextSequence(IntStream.rangeClosed(1, 20).boxed().toList())
+					.expectComplete()
+					.verify();
+
+			verify(repository).findAll(Sort.unsorted());
+		}
+
+		@DisplayName("Exception tests for invalid limits")
+		@Nested
+		class ExceptionTests {
+
+			@DisplayName("Throw IllegalArgumentException for null limit")
+			@Test
+			void shouldThrowIllegalArgumentExceptionWhenNullLimitIsGiven() {
+				assertThrows(IllegalArgumentException.class, () -> repository.findAll(Sort.unsorted(), null));
+			}
+
+			@DisplayName("Throw IllegalArgumentException for invalid limit values")
+			@ParameterizedTest
+			@ValueSource(ints = { -1, Integer.MIN_VALUE, 0 })
+			void shouldThrowIllegalArgumentExceptionForInvalidLimits(int limitValue) {
+				assertThrows(IllegalArgumentException.class, () -> repository.findAll(Sort.unsorted(), Limit.of(limitValue)));
+			}
+		}
+	}
+}