DATACMNS-836 - Add reactive repository interfaces.

Add ReactiveCrudRepository and ReactivePagingAndSortingRepository.

Author: mp911de

pom.xml

@@ -95,6 +95,15 @@
 			<optional>true</optional>
 		</dependency>
 
+		<!--  Project Reactor -->
+
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<version>${reactor}</version>
+			<optional>true</optional>
+		</dependency>
+
 		<!--  Querydsl -->
 
 		<dependency>

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

@@ -17,6 +17,8 @@
 
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -26,6 +28,7 @@
 import java.util.function.Function;
 import java.util.stream.Stream;
 
+import org.reactivestreams.Publisher;
 import org.springframework.core.CollectionFactory;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.converter.Converter;
@@ -38,7 +41,7 @@
 /**
  * A {@link ResultProcessor} to expose metadata about query result element projection and eventually post prcessing raw
  * query results into projections and data transfer objects.
- * 
+ *
  * @author Oliver Gierke
  * @author John Blum
  * @since 1.12
@@ -53,7 +56,7 @@ public class ResultProcessor {
 
 	/**
 	 * Creates a new {@link ResultProcessor} from the given {@link QueryMethod} and {@link ProjectionFactory}.
-	 * 
+	 *
 	 * @param method must not be {@literal null}.
 	 * @param factory must not be {@literal null}.
 	 */
@@ -63,7 +66,7 @@ public class ResultProcessor {
 
 	/**
 	 * Creates a new {@link ResultProcessor} for the given {@link QueryMethod}, {@link ProjectionFactory} and type.
-	 * 
+	 *
 	 * @param method must not be {@literal null}.
 	 * @param factory must not be {@literal null}.
 	 * @param type must not be {@literal null}.
@@ -82,7 +85,7 @@ private ResultProcessor(QueryMethod method, ProjectionFactory factory, Class<?>
 
 	/**
 	 * Returns a new {@link ResultProcessor} with a new projection type obtained from the given {@link ParameterAccessor}.
-	 * 
+	 *
 	 * @param accessor can be {@literal null}.
 	 * @return
 	 */
@@ -99,7 +102,7 @@ public ResultProcessor withDynamicProjection(ParameterAccessor accessor) {
 
 	/**
 	 * Returns the {@link ReturnedType}.
-	 * 
+	 *
 	 * @return
 	 */
 	public ReturnedType getReturnedType() {
@@ -108,7 +111,7 @@ public ReturnedType getReturnedType() {
 
 	/**
 	 * Post-processes the given query result.
-	 * 
+	 *
 	 * @param source can be {@literal null}.
 	 * @return
 	 */
@@ -119,7 +122,7 @@ public <T> T processResult(Object source) {
 	/**
 	 * Post-processes the given query result using the given preparing {@link Converter} to potentially prepare collection
 	 * elements.
-	 * 
+	 *
 	 * @param source can be {@literal null}.
 	 * @param preparingConverter must not be {@literal null}.
 	 * @return
@@ -156,6 +159,18 @@ public <T> T processResult(Object source, Converter<Object, Object> preparingCon
 			return (T) new StreamQueryResultHandler(type, converter).handle(source);
 		}
 
+		if (source instanceof Mono) {
+
+			Mono<?> mono = (Mono<?>) source;
+			return (T) mono.map(o -> type.isInstance(o) ? o : converter.convert(o));
+		}
+
+		if (source instanceof Flux) {
+
+			Flux<?> flux = (Flux<?>) source;
+			return (T) flux.map(o -> type.isInstance(o) ? o : converter.convert(o));
+		}
+
 		return (T) converter.convert(source);
 	}
 
@@ -168,7 +183,7 @@ private static class ChainingConverter implements Converter<Object, Object> {
 		/**
 		 * Returns a new {@link ChainingConverter} that hands the elements resulting from the current conversion to the
 		 * given {@link Converter}.
-		 * 
+		 *
 		 * @param converter must not be {@literal null}.
 		 * @return
 		 */
@@ -187,7 +202,7 @@ public Object convert(Object source) {
 			});
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
 		 */
@@ -207,7 +222,7 @@ private static enum NoOpConverter implements Converter<Object, Object> {
 
 		INSTANCE;
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
 		 */
@@ -224,7 +239,7 @@ private static class ProjectingConverter implements Converter<Object, Object> {
 		private final @NonNull ProjectionFactory factory;
 		private final ConversionService conversionService = new DefaultConversionService();
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
 		 */

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

@@ -0,0 +1,164 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 java.io.Serializable;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.repository.NoRepositoryBean;
+import org.springframework.data.repository.Repository;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Interface for generic CRUD operations on a repository for a specific type.
+ *
+ * @author Mark Paluch
+ */
+@NoRepositoryBean
+public interface ReactiveCrudRepository<T, ID extends Serializable> extends Repository<T, ID> {
+
+	/**
+	 * Saves a given entity. Use the returned instance for further operations as the save operation might have changed the
+	 * entity instance completely.
+	 *
+	 * @param entity
+	 * @return the saved entity
+	 */
+	<S extends T> Mono<S> save(S entity);
+
+	/**
+	 * Saves all given entities.
+	 *
+	 * @param entities must not be {@literal null}.
+	 * @return the saved entities
+	 * @throws IllegalArgumentException in case the given entity is {@literal null}.
+	 */
+	<S extends T> Flux<S> save(Iterable<S> entities);
+
+	/**
+	 * Saves all given entities.
+	 *
+	 * @param entityStream must not be {@literal null}.
+	 * @return the saved entities
+	 * @throws IllegalArgumentException in case the given {@code Publisher} is {@literal null}.
+	 */
+	<S extends T> Flux<S> save(Publisher<S> entityStream);
+
+	/**
+	 * Retrieves an entity by its id.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return the entity with the given id or {@literal null} if none found
+	 * @throws IllegalArgumentException if {@code id} is {@literal null}
+	 */
+	Mono<T> findOne(ID id);
+
+	/**
+	 * Retrieves an entity by its id supplied by a {@link Mono}.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return the entity with the given id or {@literal null} if none found
+	 * @throws IllegalArgumentException if {@code id} is {@literal null}
+	 */
+	Mono<T> findOne(Mono<ID> id);
+
+	/**
+	 * Returns whether an entity with the given id exists.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return true if an entity with the given id exists, {@literal false} otherwise
+	 * @throws IllegalArgumentException if {@code id} is {@literal null}
+	 */
+	Mono<Boolean> exists(ID id);
+
+	/**
+	 * Returns whether an entity with the given id exists supplied by a {@link Mono}.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @return true if an entity with the given id exists, {@literal false} otherwise
+	 * @throws IllegalArgumentException if {@code id} is {@literal null}
+	 */
+	Mono<Boolean> exists(Mono<ID> id);
+
+	/**
+	 * Returns all instances of the type.
+	 *
+	 * @return all entities
+	 */
+	Flux<T> findAll();
+
+	/**
+	 * Returns all instances of the type with the given IDs.
+	 *
+	 * @param ids
+	 * @return
+	 */
+	Flux<T> findAll(Iterable<ID> ids);
+
+	/**
+	 * Returns all instances of the type with the given IDs.
+	 *
+	 * @param idStream
+	 * @return
+	 */
+	Flux<T> findAll(Publisher<ID> idStream);
+
+	/**
+	 * Returns the number of entities available.
+	 *
+	 * @return the number of entities
+	 */
+	Mono<Long> count();
+
+	/**
+	 * Deletes the entity with the given id.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @throws IllegalArgumentException in case the given {@code id} is {@literal null}
+	 */
+	Mono<Void> delete(ID id);
+
+	/**
+	 * Deletes a given entity.
+	 *
+	 * @param entity
+	 * @throws IllegalArgumentException in case the given entity is {@literal null}.
+	 */
+	Mono<Void> delete(T entity);
+
+	/**
+	 * Deletes the given entities.
+	 *
+	 * @param entities
+	 * @throws IllegalArgumentException in case the given {@link Iterable} is {@literal null}.
+	 */
+	Mono<Void> delete(Iterable<? extends T> entities);
+
+	/**
+	 * Deletes the given entities.
+	 *
+	 * @param entityStream
+	 * @throws IllegalArgumentException in case the given {@link Publisher} is {@literal null}.
+	 */
+	Mono<Void> delete(Publisher<? extends T> entityStream);
+
+	/**
+	 * Deletes all entities managed by the repository.
+	 */
+	Mono<Void> deleteAll();
+}

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

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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 java.io.Serializable;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.repository.NoRepositoryBean;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Extension of {@link ReactiveCrudRepository} to provide additional methods to retrieve entities using the pagination
+ * and sorting abstraction.
+ *
+ * @author Mark Paluch
+ * @see Sort
+ * @see Pageable
+ * @see org.springframework.data.domain.reactive.ReactivePage
+ */
+@NoRepositoryBean
+public interface ReactivePagingAndSortingRepository<T, ID extends Serializable> extends ReactiveCrudRepository<T, ID> {
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findAll()
+	 */
+	@Override
+	Flux<T> findAll();
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findAll(java.lang.Iterable)
+	 */
+	@Override
+	Flux<T> findAll(Iterable<ID> ids);
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findAll(org.reactivestreams.Publisher)
+	 */
+	@Override
+	Flux<T> findAll(Publisher<ID> idStream);
+
+	/**
+	 * Returns all entities sorted by the given options.
+	 *
+	 * @param sort
+	 * @return all entities sorted by the given options
+	 */
+	Flux<T> findAll(Sort sort);
+
+	/**
+	 * Returns a {@link Page} of entities meeting the paging restriction provided in the {@code Pageable} object.
+	 *
+	 * @param pageable
+	 * @return a page of entities
+	 */
+	Mono<Page<T>> findAll(Pageable pageable);
+}