DATACMNS-836 - Add reactive repository support.

We now expose reactive interfaces to facilitate reactive repository support in store-specific modules. Spring Data modules are free to implement their reactive support using either RxJava1 or Project Reactor (Reactive Streams). We expose a set of base interfaces:

* `ReactiveCrudRepository`
* `ReactivePagingAndSortingRepository`
* `RxJavaCrudRepository`
* `RxJavaPagingAndSortingRepository`

Reactive repositories provide a similar feature coverage to blocking repositories. Reactive paging support is limited to a `Mono<Page>`/`Single<Page>`. Data is fetched in a deferred way to provide a paging experience similar to blocking paging.

A store module can choose either Project Reactor or RxJava 1 to implement reactive repository support. Project Reactor and RxJava types are converted in both directions allowing repositories to be composed of Project Reactor and RxJava query methods. Reactive wrapper type conversion handles wrapper type conversion at repository level. Query/implementation method selection uses multi-pass candidate selection to invoke the most appropriate method (exact arguments, convertible wrappers, assignable arguments).

We also provide ReactiveWrappers and ReactiveWrapperConverters to expose metadata about reactive types and their value multiplicity.

Author: mp911de

pom.xml

@@ -105,6 +105,15 @@
 			<optional>true</optional>
 		</dependency>
 
+		<!--  RxJava -->
+
+		<dependency>
+			<groupId>io.reactivex</groupId>
+			<artifactId>rxjava</artifactId>
+			<version>${rxjava}</version>
+			<optional>true</optional>
+		</dependency>
+
 		<!--  Querydsl -->
 
 		<dependency>

src/main/java/org/springframework/data/repository/core/support/DefaultRepositoryInformation.java

@@ -336,7 +336,7 @@ public Set<Class<?>> getAlternativeDomainTypes() {
 
 	/**
 	 * Checks the given method's parameters to match the ones of the given base class method. Matches generic arguments
-	 * agains the ones bound in the given repository interface.
+	 * against the ones bound in the given repository interface.
 	 * 
 	 * @param method
 	 * @param baseClassMethod
@@ -381,7 +381,7 @@ private boolean parametersMatch(Method method, Method baseClassMethod) {
 	 * @param parameterType must not be {@literal null}.
 	 * @return
 	 */
-	private boolean matchesGenericType(TypeVariable<?> variable, ResolvableType parameterType) {
+	protected boolean matchesGenericType(TypeVariable<?> variable, ResolvableType parameterType) {
 
 		GenericDeclaration declaration = variable.getGenericDeclaration();
 

src/main/java/org/springframework/data/repository/core/support/QueryExecutionResultHandler.java

@@ -28,6 +28,7 @@
  * Simple domain service to convert query results into a dedicated type.
  *
  * @author Oliver Gierke
+ * @author Mark Paluch
  */
 class QueryExecutionResultHandler {
 
@@ -50,25 +51,36 @@ public QueryExecutionResultHandler() {
 	 * Post-processes the given result of a query invocation to the given type.
 	 * 
 	 * @param result can be {@literal null}.
-	 * @param returnTypeDesciptor can be {@literal null}, if so, no conversion is performed.
+	 * @param returnTypeDescriptor can be {@literal null}, if so, no conversion is performed.
 	 * @return
 	 */
-	public Object postProcessInvocationResult(Object result, TypeDescriptor returnTypeDesciptor) {
+	public Object postProcessInvocationResult(Object result, TypeDescriptor returnTypeDescriptor) {
 
-		if (returnTypeDesciptor == null) {
+		if (returnTypeDescriptor == null) {
 			return result;
 		}
 
-		Class<?> expectedReturnType = returnTypeDesciptor.getType();
+		Class<?> expectedReturnType = returnTypeDescriptor.getType();
 
 		if (result != null && expectedReturnType.isInstance(result)) {
 			return result;
 		}
 
-		if (QueryExecutionConverters.supports(expectedReturnType)
-				&& conversionService.canConvert(WRAPPER_TYPE, returnTypeDesciptor)
-				&& !conversionService.canBypassConvert(WRAPPER_TYPE, TypeDescriptor.valueOf(expectedReturnType))) {
-			return conversionService.convert(new NullableWrapper(result), expectedReturnType);
+		if (QueryExecutionConverters.supports(expectedReturnType)) {
+
+			TypeDescriptor targetType = TypeDescriptor.valueOf(expectedReturnType);
+
+			if(conversionService.canConvert(WRAPPER_TYPE, returnTypeDescriptor)
+				&& !conversionService.canBypassConvert(WRAPPER_TYPE, targetType)) {
+
+				return conversionService.convert(new NullableWrapper(result), expectedReturnType);
+			}
+
+			if(result != null && conversionService.canConvert(TypeDescriptor.valueOf(result.getClass()), returnTypeDescriptor)
+				&& !conversionService.canBypassConvert(TypeDescriptor.valueOf(result.getClass()), targetType)) {
+
+				return conversionService.convert(result, expectedReturnType);
+			}
 		}
 
 		if (result != null) {
@@ -82,4 +94,5 @@ public Object postProcessInvocationResult(Object result, TypeDescriptor returnTy
 
 		return null;
 	}
+
 }

src/main/java/org/springframework/data/repository/core/support/ReactiveRepositoryInformation.java

@@ -0,0 +1,259 @@
+/*
+ * 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.core.support;
+
+import static org.springframework.core.GenericTypeResolver.*;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
+import java.util.function.BiPredicate;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.data.repository.core.RepositoryInformation;
+import org.springframework.data.repository.core.RepositoryMetadata;
+import org.springframework.data.repository.util.QueryExecutionConverters;
+import org.springframework.util.Assert;
+
+/**
+ * This {@link RepositoryInformation} uses a {@link ConversionService} to check whether method arguments can be
+ * converted for invocation of implementation methods.
+ * 
+ * @author Mark Paluch
+ */
+public class ReactiveRepositoryInformation extends DefaultRepositoryInformation {
+
+	private final ConversionService conversionService;
+
+	/**
+	 * Creates a new {@link ReactiveRepositoryInformation} for the given repository interface and repository base class
+	 * using a {@link ConversionService}.
+	 * 
+	 * @param metadata must not be {@literal null}.
+	 * @param repositoryBaseClass must not be {@literal null}.
+	 * @param customImplementationClass
+	 * @param conversionService must not be {@literal null}.
+	 */
+	public ReactiveRepositoryInformation(RepositoryMetadata metadata, Class<?> repositoryBaseClass,
+			Class<?> customImplementationClass, ConversionService conversionService) {
+
+		super(metadata, repositoryBaseClass, customImplementationClass);
+
+		Assert.notNull(conversionService, "Conversion service must not be null!");
+
+		this.conversionService = conversionService;
+	}
+
+	/**
+	 * Returns the given target class' method if the given method (declared in the repository interface) was also declared
+	 * at the target class. Returns the given method if the given base class does not declare the method given. Takes
+	 * generics into account.
+	 *
+	 * @param method must not be {@literal null}
+	 * @param baseClass
+	 * @return
+	 */
+	Method getTargetClassMethod(Method method, Class<?> baseClass) {
+
+		if (baseClass == null) {
+			return method;
+		}
+
+		boolean wantsWrappers = wantsMethodUsingReactiveWrapperParameters(method);
+
+		if (wantsWrappers) {
+			Method candidate = getMethodCandidate(method, baseClass, new ExactWrapperMatch(method));
+
+			if (candidate != null) {
+				return candidate;
+			}
+
+			candidate = getMethodCandidate(method, baseClass, new WrapperConversionMatch(method, conversionService));
+
+			if (candidate != null) {
+				return candidate;
+			}
+		}
+
+		Method candidate = getMethodCandidate(method, baseClass,
+				new MatchParameterOrComponentType(method, getRepositoryInterface()));
+
+		if (candidate != null) {
+			return candidate;
+		}
+
+		return method;
+	}
+
+	private boolean wantsMethodUsingReactiveWrapperParameters(Method method) {
+
+		boolean wantsWrappers = false;
+
+		for (Class<?> parameterType : method.getParameterTypes()) {
+			if (isNonunwrappingWrapper(parameterType)) {
+				wantsWrappers = true;
+				break;
+			}
+		}
+
+		return wantsWrappers;
+	}
+
+	private Method getMethodCandidate(Method method, Class<?> baseClass, BiPredicate<Class<?>, Integer> predicate) {
+
+		for (Method baseClassMethod : baseClass.getMethods()) {
+
+			// Wrong name
+			if (!method.getName().equals(baseClassMethod.getName())) {
+				continue;
+			}
+
+			// Wrong number of arguments
+			if (!(method.getParameterTypes().length == baseClassMethod.getParameterTypes().length)) {
+				continue;
+			}
+
+			// Check whether all parameters match
+			if (!parametersMatch(method, baseClassMethod, predicate)) {
+				continue;
+			}
+
+			return baseClassMethod;
+		}
+
+		return null;
+	}
+
+	/**
+	 * Checks the given method's parameters to match the ones of the given base class method. Matches generic arguments
+	 * against the ones bound in the given repository interface.
+	 *
+	 * @param method
+	 * @param baseClassMethod
+	 * @param predicate
+	 * @return
+	 */
+	private boolean parametersMatch(Method method, Method baseClassMethod, BiPredicate<Class<?>, Integer> predicate) {
+
+		Type[] genericTypes = baseClassMethod.getGenericParameterTypes();
+		Class<?>[] types = baseClassMethod.getParameterTypes();
+
+		for (int i = 0; i < genericTypes.length; i++) {
+			if (!predicate.test(types[i], i)) {
+				return false;
+			}
+		}
+
+		return true;
+	}
+
+	/**
+	 * Checks whether the type is a wrapper without unwrapping support. Reactive wrappers don't like to be unwrapped.
+	 *
+	 * @param parameterType
+	 * @return
+	 */
+	static boolean isNonunwrappingWrapper(Class<?> parameterType) {
+		return QueryExecutionConverters.supports(parameterType)
+				&& !QueryExecutionConverters.supportsUnwrapping(parameterType);
+	}
+
+	static class WrapperConversionMatch implements BiPredicate<Class<?>, Integer> {
+
+		final Method declaredMethod;
+		final Class<?>[] declaredParameterTypes;
+		final ConversionService conversionService;
+
+		public WrapperConversionMatch(Method declaredMethod, ConversionService conversionService) {
+
+			this.declaredMethod = declaredMethod;
+			this.declaredParameterTypes = declaredMethod.getParameterTypes();
+			this.conversionService = conversionService;
+		}
+
+		@Override
+		public boolean test(Class<?> candidateParameterType, Integer index) {
+
+			// TODO: should check for component type
+			if (isNonunwrappingWrapper(candidateParameterType) && isNonunwrappingWrapper(declaredParameterTypes[index])) {
+
+				if (conversionService.canConvert(declaredParameterTypes[index], candidateParameterType)) {
+					return true;
+				}
+			}
+
+			return false;
+		}
+
+	}
+
+	static class ExactWrapperMatch implements BiPredicate<Class<?>, Integer> {
+
+		final Method declaredMethod;
+		final Class<?>[] declaredParameterTypes;
+
+		public ExactWrapperMatch(Method declaredMethod) {
+
+			this.declaredMethod = declaredMethod;
+			this.declaredParameterTypes = declaredMethod.getParameterTypes();
+		}
+
+		@Override
+		public boolean test(Class<?> candidateParameterType, Integer index) {
+
+			// TODO: should check for component type
+			if (isNonunwrappingWrapper(candidateParameterType) && isNonunwrappingWrapper(declaredParameterTypes[index])) {
+
+				if (declaredParameterTypes[index].isAssignableFrom(candidateParameterType)) {
+					return true;
+				}
+			}
+
+			return false;
+		}
+
+	}
+
+	static class MatchParameterOrComponentType implements BiPredicate<Class<?>, Integer> {
+
+		final Method declaredMethod;
+		final Class<?>[] declaredParameterTypes;
+		final Class<?> repositoryInterface;
+
+		public MatchParameterOrComponentType(Method declaredMethod, Class<?> repositoryInterface) {
+
+			this.declaredMethod = declaredMethod;
+			this.declaredParameterTypes = declaredMethod.getParameterTypes();
+			this.repositoryInterface = repositoryInterface;
+		}
+
+		@Override
+		public boolean test(Class<?> candidateParameterType, Integer index) {
+
+			MethodParameter parameter = new MethodParameter(declaredMethod, index);
+			Class<?> parameterType = resolveParameterType(parameter, repositoryInterface);
+
+			if (!candidateParameterType.isAssignableFrom(parameterType)
+					|| !candidateParameterType.equals(declaredParameterTypes[index])) {
+				return false;
+			}
+
+			return true;
+		}
+
+	}
+
+}