DATACMNS-1508 - Support for Kotlin Coroutines repositories.

We now support Kotlin Coroutines repositories backed by reactive repositories. Coroutine repositories can either invoke reactive query methods or implemented methods that are backed either by methods returning a reactive wrapper or that are native suspended functions.

Exclude Coroutines Continuation from bindable parameters and name discovery and do not unwrap results for suspended functions returning a reactive type.

interface CoroutinesRepository : CoroutineCrudRepository<User, String> {

    suspend fun findOne(id: String): User

    fun findByFirstname(firstname: String): Flow<User>
}

Original pull request: #415.

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

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2019 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.core.support;
+
+import kotlin.coroutines.Continuation;
+import kotlin.reflect.KFunction;
+import kotlinx.coroutines.reactive.AwaitKt;
+
+import java.lang.reflect.Method;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.KotlinDetector;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.data.util.KotlinReflectionUtils;
+import org.springframework.data.util.ReflectionUtils;
+import org.springframework.lang.Nullable;
+
+/**
+ * Metadata for a implementation {@link Method} invocation. This value object encapsulates whether the called and the
+ * backing method are regular methods or suspendable Kotlin coroutines methods. It also allows invocation of suspendable
+ * methods by backing the invocation using methods returning reactive types.
+ *
+ * @author Mark Paluch
+ * @since 2.3
+ */
+class ImplementationInvocationMetadata {
+
+	private final boolean suspendedDeclaredMethod;
+	private final boolean suspendedBaseClassMethod;
+	private final boolean reactiveBaseClassMethod;
+
+	ImplementationInvocationMetadata(Method declaredMethod, Method baseClassMethod) {
+
+		if (!KotlinDetector.isKotlinReflectPresent()) {
+			suspendedDeclaredMethod = false;
+			suspendedBaseClassMethod = false;
+			reactiveBaseClassMethod = false;
+			return;
+		}
+
+		KFunction<?> declaredFunction = ReflectionUtils.isKotlinClass(declaredMethod.getDeclaringClass())
+				? KotlinReflectionUtils.findKotlinFunction(declaredMethod)
+				: null;
+		KFunction<?> baseClassFunction = ReflectionUtils.isKotlinClass(baseClassMethod.getDeclaringClass())
+				? KotlinReflectionUtils.findKotlinFunction(baseClassMethod)
+				: null;
+
+		suspendedDeclaredMethod = declaredFunction != null && declaredFunction.isSuspend();
+		suspendedBaseClassMethod = baseClassFunction != null && baseClassFunction.isSuspend();
+		this.reactiveBaseClassMethod = !suspendedBaseClassMethod
+				&& ReactiveWrapperConverters.supports(baseClassMethod.getReturnType());
+	}
+
+	@Nullable
+	public Object invoke(Method methodToCall, Object instance, Object[] args) throws Throwable {
+
+		if (suspendedDeclaredMethod && !suspendedBaseClassMethod && reactiveBaseClassMethod) {
+			return invokeReactiveToSuspend(methodToCall, instance, args);
+		}
+
+		return methodToCall.invoke(instance, args);
+	}
+
+	@Nullable
+	@SuppressWarnings({ "unchecked", "ConstantConditions" })
+	private Object invokeReactiveToSuspend(Method methodToCall, Object instance, Object[] args)
+			throws ReflectiveOperationException {
+
+		Object[] invocationArguments = new Object[args.length - 1];
+		System.arraycopy(args, 0, invocationArguments, 0, invocationArguments.length);
+		Object result = methodToCall.invoke(instance, invocationArguments);
+
+		Publisher<?> publisher;
+		if (result instanceof Publisher) {
+			publisher = (Publisher<?>) result;
+		} else {
+			publisher = ReactiveWrapperConverters.toWrapper(result, Publisher.class);
+		}
+
+		return AwaitKt.awaitFirstOrNull(publisher, (Continuation) args[args.length - 1]);
+	}
+
+	boolean canInvoke(Method invokedMethod, Method backendMethod) {
+
+		if (suspendedDeclaredMethod == suspendedBaseClassMethod) {
+			return invokedMethod.getParameterCount() == backendMethod.getParameterCount();
+		}
+
+		if (suspendedDeclaredMethod && reactiveBaseClassMethod) {
+			return invokedMethod.getParameterCount() - 1 == backendMethod.getParameterCount();
+		}
+
+		return false;
+	}
+}

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

@@ -331,23 +331,22 @@ interface MethodLookups {
 
 			return Optional.of(candidate)//
 					.filter(it -> invokedMethod.getName().equals(it.getName()))//
-					.filter(it -> invokedMethod.getParameterCount() == it.getParameterCount())//
-					.filter(it -> parametersMatch(it, invokedMethod.getMethod(), predicate));
+					.filter(it -> parameterCountMatch(invokedMethod, it))//
+					.filter(it -> parametersMatch(invokedMethod.getMethod(), it, predicate));
 		}
 
 		/**
 		 * 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 baseClassMethod must not be {@literal null}.
 		 * @param declaredMethod must not be {@literal null}.
+		 * @param baseClassMethod must not be {@literal null}.
 		 * @param predicate must not be {@literal null}.
 		 * @return
 		 */
-		private static boolean parametersMatch(Method baseClassMethod, Method declaredMethod,
+		private static boolean parametersMatch(Method declaredMethod, Method baseClassMethod,
 				Predicate<ParameterOverrideCriteria> predicate) {
-
-			return methodParameters(baseClassMethod, declaredMethod).allMatch(predicate);
+			return methodParameters(declaredMethod, baseClassMethod).allMatch(predicate);
 		}
 
 		/**
@@ -378,13 +377,17 @@ interface MethodLookups {
 					&& parameterCriteria.getBaseType().isAssignableFrom(parameterCriteria.getDeclaredType());
 		}
 
-		private static Stream<ParameterOverrideCriteria> methodParameters(Method first, Method second) {
+		private static boolean parameterCountMatch(InvokedMethod invokedMethod, Method baseClassMethod) {
 
-			Assert.isTrue(first.getParameterCount() == second.getParameterCount(), "Method parameter count must be equal!");
+			ImplementationInvocationMetadata invocationMetadata = new ImplementationInvocationMetadata(
+					invokedMethod.getMethod(), baseClassMethod);
+			return invocationMetadata.canInvoke(invokedMethod.getMethod(), baseClassMethod);
+		}
 
-			return IntStream.range(0, first.getParameterCount()) //
-					.mapToObj(index -> ParameterOverrideCriteria.of(new MethodParameter(first, index),
-							new MethodParameter(second, index)));
+		private static Stream<ParameterOverrideCriteria> methodParameters(Method invokedMethod, Method baseClassMethod) {
+			return IntStream.range(0, baseClassMethod.getParameterCount()) //
+					.mapToObj(index -> ParameterOverrideCriteria.of(new MethodParameter(invokedMethod, index),
+							new MethodParameter(baseClassMethod, index)));
 		}
 
 		/**
@@ -395,8 +398,8 @@ interface MethodLookups {
 		@Value(staticConstructor = "of")
 		static class ParameterOverrideCriteria {
 
-			private final MethodParameter base;
 			private final MethodParameter declared;
+			private final MethodParameter base;
 
 			/**
 			 * @return base method parameter type.

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

@@ -28,6 +28,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -201,7 +202,7 @@ public class RepositoryComposition {
 
 		ReflectionUtils.makeAccessible(methodToCall);
 
-		return fragments.invoke(methodToCall, argumentConverter.apply(methodToCall, args));
+		return fragments.invoke(method, methodToCall, argumentConverter.apply(methodToCall, args));
 	}
 
 	/**
@@ -250,6 +251,7 @@ public class RepositoryComposition {
 		static final RepositoryFragments EMPTY = new RepositoryFragments(Collections.emptyList());
 
 		private final Map<Method, RepositoryFragment<?>> fragmentCache = new ConcurrentReferenceHashMap<>();
+		private final Map<Method, ImplementationInvocationMetadata> invocationMetadataCache = new ConcurrentHashMap<>();
 		private final List<RepositoryFragment<?>> fragments;
 
 		/**
@@ -354,21 +356,30 @@ public class RepositoryComposition {
 		/**
 		 * Invoke {@link Method} by resolving the
 		 *
-		 * @param method
+		 * @param invokedMethod invoked method as per invocation on the interface.
+		 * @param methodToCall backend method that is backing the call.
 		 * @param args
 		 * @return
 		 * @throws Throwable
 		 */
-		public Object invoke(Method method, Object[] args) throws Throwable {
+		@Nullable
+		public Object invoke(Method invokedMethod, Method methodToCall, Object[] args) throws Throwable {
 
-			RepositoryFragment<?> fragment = fragmentCache.computeIfAbsent(method, this::findImplementationFragment);
+			RepositoryFragment<?> fragment = fragmentCache.computeIfAbsent(methodToCall, this::findImplementationFragment);
 			Optional<?> optional = fragment.getImplementation();
 
 			if (!optional.isPresent()) {
-				throw new IllegalArgumentException(String.format("No implementation found for method %s", method));
+				throw new IllegalArgumentException(String.format("No implementation found for method %s", methodToCall));
 			}
 
-			return method.invoke(optional.get(), args);
+			ImplementationInvocationMetadata invocationMetadata = invocationMetadataCache.get(invokedMethod);
+
+			if (invocationMetadata == null) {
+				invocationMetadata = new ImplementationInvocationMetadata(invokedMethod, methodToCall);
+				invocationMetadataCache.put(invokedMethod, invocationMetadata);
+			}
+
+			return invocationMetadata.invoke(methodToCall, optional.get(), args);
 		}
 
 		private RepositoryFragment<?> findImplementationFragment(Method key) {

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

@@ -15,6 +15,9 @@
  */
 package org.springframework.data.repository.core.support;
 
+import kotlin.coroutines.Continuation;
+import kotlin.reflect.KFunction;
+import kotlinx.coroutines.reactive.AwaitKt;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NonNull;
@@ -35,6 +38,7 @@ import java.util.stream.Collectors;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
+import org.reactivestreams.Publisher;
 
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.interceptor.ExposeInvocationInterceptor;
@@ -43,6 +47,7 @@ import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.core.convert.support.GenericConversionService;
@@ -64,6 +69,7 @@ import org.springframework.data.repository.util.ClassUtils;
 import org.springframework.data.repository.util.QueryExecutionConverters;
 import org.springframework.data.repository.util.ReactiveWrapperConverters;
 import org.springframework.data.repository.util.ReactiveWrappers;
+import org.springframework.data.util.KotlinReflectionUtils;
 import org.springframework.data.util.Pair;
 import org.springframework.data.util.ReflectionUtils;
 import org.springframework.lang.Nullable;
@@ -85,35 +91,35 @@ import org.springframework.util.ConcurrentReferenceHashMap.ReferenceType;
 @Slf4j
 public abstract class RepositoryFactorySupport implements BeanClassLoaderAware, BeanFactoryAware {
 
-	private static final BiFunction<Method, Object[], Object[]> REACTIVE_ARGS_CONVERTER = (method, o) -> {
+	private static final BiFunction<Method, Object[], Object[]> REACTIVE_ARGS_CONVERTER = (method, args) -> {
 
 		if (ReactiveWrappers.isAvailable()) {
 
 			Class<?>[] parameterTypes = method.getParameterTypes();
 
-			Object[] converted = new Object[o.length];
-			for (int i = 0; i < parameterTypes.length; i++) {
+			Object[] converted = new Object[args.length];
+			for (int i = 0; i < args.length; i++) {
 
-				Class<?> parameterType = parameterTypes[i];
-				Object value = o[i];
+				Object value = args[i];
+				Object convertedArg = value;
 
-				if (value == null) {
-					continue;
+				Class<?> parameterType = parameterTypes.length > i ? parameterTypes[i] : null;
+
+				if (value != null && parameterType != null) {
+					if (!parameterType.isAssignableFrom(value.getClass())
+							&& ReactiveWrapperConverters.canConvert(value.getClass(), parameterType)) {
+
+						convertedArg = ReactiveWrapperConverters.toWrapper(value, parameterType);
+					}
 				}
 
-				if (!parameterType.isAssignableFrom(value.getClass())
-						&& ReactiveWrapperConverters.canConvert(value.getClass(), parameterType)) {
-
-					converted[i] = ReactiveWrapperConverters.toWrapper(value, parameterType);
-				} else {
-					converted[i] = value;
-				}
+				converted[i] = convertedArg;
 			}
 
 			return converted;
 		}
 
-		return o;
+		return args;
 	};
 
 	final static GenericConversionService CONVERSION_SERVICE = new DefaultConversionService();
@@ -534,6 +540,7 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 	public class QueryExecutorMethodInterceptor implements MethodInterceptor {
 
 		private final Map<Method, RepositoryQuery> queries;
+		private final Map<Method, QueryMethodInvoker> invocationMetadataCache = new ConcurrentReferenceHashMap<>();
 		private final QueryExecutionResultHandler resultHandler;
 
 		/**
@@ -615,7 +622,16 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 			Method method = invocation.getMethod();
 
 			if (hasQueryFor(method)) {
-				return queries.get(method).execute(invocation.getArguments());
+
+				QueryMethodInvoker invocationMetadata = invocationMetadataCache.get(method);
+
+				if (invocationMetadata == null) {
+					invocationMetadata = new QueryMethodInvoker(method);
+					invocationMetadataCache.put(method, invocationMetadata);
+				}
+
+				RepositoryQuery repositoryQuery = queries.get(method);
+				return invocationMetadata.invoke(repositoryQuery, invocation.getArguments());
 			}
 
 			return invocation.proceed();
@@ -711,4 +727,55 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 			this.compositionHash = composition.hashCode();
 		}
 	}
+
+	static class QueryMethodInvoker {
+
+		private final boolean suspendedDeclaredMethod;
+		private final Class<?> returnedType;
+		private final boolean returnsReactiveType;
+
+		QueryMethodInvoker(Method invokedMethod) {
+
+			this.suspendedDeclaredMethod = KotlinDetector.isKotlinReflectPresent() && isSuspendedMethod(invokedMethod);
+			this.returnedType = this.suspendedDeclaredMethod ? KotlinReflectionUtils.getReturnType(invokedMethod)
+					: invokedMethod.getReturnType();
+			this.returnsReactiveType = ReactiveWrappers.supports(returnedType);
+		}
+
+		private static boolean isSuspendedMethod(Method invokedMethod) {
+
+			KFunction<?> invokedFunction = ReflectionUtils.isKotlinClass(invokedMethod.getDeclaringClass())
+					? KotlinReflectionUtils.findKotlinFunction(invokedMethod)
+					: null;
+
+			return invokedFunction != null && invokedFunction.isSuspend();
+		}
+
+		@Nullable
+		public Object invoke(RepositoryQuery query, Object[] args) {
+			return suspendedDeclaredMethod ? invokeReactiveToSuspend(query, args) : query.execute(args);
+		}
+
+		@Nullable
+		@SuppressWarnings({ "unchecked", "ConstantConditions" })
+		private Object invokeReactiveToSuspend(RepositoryQuery query, Object[] args) {
+
+			Continuation<Object> continuation = (Continuation) args[args.length - 1];
+			args[args.length - 1] = null;
+			Object result = query.execute(args);
+
+			if (returnsReactiveType) {
+				return ReactiveWrapperConverters.toWrapper(result, returnedType);
+			}
+
+			Publisher<?> publisher;
+			if (result instanceof Publisher) {
+				publisher = (Publisher<?>) result;
+			} else {
+				publisher = ReactiveWrapperConverters.toWrapper(result, Publisher.class);
+			}
+
+			return AwaitKt.awaitFirstOrNull(publisher, continuation);
+		}
+	}
 }

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

@@ -18,7 +18,9 @@ package org.springframework.data.repository.query;
 import static java.lang.String.*;
 
 import java.lang.reflect.Method;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
@@ -31,6 +33,7 @@ import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.Lazy;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 
 /**
  * Class to abstract a single parameter of a query method. It is held in the context of a {@link Parameters} instance.
@@ -41,7 +44,7 @@ import org.springframework.util.Assert;
  */
 public class Parameter {
 
-	static final List<Class<?>> TYPES = Arrays.asList(Pageable.class, Sort.class);
+	static final List<Class<?>> TYPES;
 
 	private static final String NAMED_PARAMETER_TEMPLATE = ":%s";
 	private static final String POSITION_PARAMETER_TEMPLATE = "?%s";
@@ -51,6 +54,21 @@ public class Parameter {
 	private final boolean isDynamicProjectionParameter;
 	private final Lazy<Optional<String>> name;
 
+	static {
+
+		List<Class<?>> types = new ArrayList<>(Arrays.asList(Pageable.class, Sort.class));
+
+		try {
+			// consider Kotlin Coroutines Continuation a special parameter. That parameter is synthetic and should not get
+			// bound to any query.
+			types.add(ClassUtils.forName("kotlin.coroutines.Continuation", Parameter.class.getClassLoader()));
+		} catch (ClassNotFoundException e) {
+			// ignore
+		}
+
+		TYPES = Collections.unmodifiableList(types);
+	}
+
 	/**
 	 * Creates a new {@link Parameter} for the given {@link MethodParameter}.
 	 *
@@ -63,7 +81,7 @@ public class Parameter {
 		this.parameter = parameter;
 		this.parameterType = potentiallyUnwrapParameterType(parameter);
 		this.isDynamicProjectionParameter = isDynamicProjectionParameter(parameter);
-		this.name = Lazy.of(() -> {
+		this.name = TYPES.contains(parameter.getParameterType()) ? Lazy.of(Optional.empty()) : Lazy.of(() -> {
 			Param annotation = parameter.getParameterAnnotation(Param.class);
 			return Optional.ofNullable(annotation == null ? parameter.getParameterName() : annotation.value());
 		});

src/main/java/org/springframework/data/repository/reactive/package-info.java

@@ -1,5 +1,5 @@
 /**
- * Support for reactive repository.
+ * Support for reactive repositories.
  */
 @org.springframework.lang.NonNullApi
 package org.springframework.data.repository.reactive;

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

@@ -20,6 +20,8 @@ import static org.springframework.data.repository.util.ReactiveWrapperConverters
 import io.reactivex.BackpressureStrategy;
 import io.reactivex.Flowable;
 import io.reactivex.Maybe;
+import kotlinx.coroutines.flow.Flow;
+import kotlinx.coroutines.reactive.ReactiveFlowKt;
 import lombok.experimental.UtilityClass;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -34,10 +36,14 @@ import java.util.function.Function;
 import javax.annotation.Nonnull;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.ConditionalConverter;
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.converter.ConverterFactory;
 import org.springframework.core.convert.support.ConfigurableConversionService;
 import org.springframework.core.convert.support.GenericConversionService;
 import org.springframework.data.repository.util.ReactiveWrappers.ReactiveLibrary;
@@ -156,6 +162,12 @@ public class ReactiveWrapperConverters {
 				conversionService.addConverter(RxJava2ObservableToSingleConverter.INSTANCE);
 				conversionService.addConverter(RxJava2ObservableToMaybeConverter.INSTANCE);
 			}
+
+			if (ReactiveWrappers.isAvailable(ReactiveLibrary.KOTLIN_COROUTINES)) {
+				conversionService.addConverter(PublisherToFlowConverter.INSTANCE);
+			}
+
+			conversionService.addConverterFactory(ReactiveAdapterConverterFactory.INSTANCE);
 		}
 
 		return conversionService;
@@ -492,6 +504,27 @@ public class ReactiveWrapperConverters {
 		}
 	}
 
+	// -------------------------------------------------------------------------
+	// Coroutine converters
+	// -------------------------------------------------------------------------
+
+	/**
+	 * A {@link Converter} to convert a {@link Publisher} to {@link Flow}.
+	 *
+	 * @author Mark Paluch
+	 * @author 2.3
+	 */
+	private enum PublisherToFlowConverter implements Converter<Publisher<?>, Flow<?>> {
+
+		INSTANCE;
+
+		@Nonnull
+		@Override
+		public Flow<?> convert(Publisher<?> source) {
+			return ReactiveFlowKt.asFlow(source);
+		}
+	}
+
 	// -------------------------------------------------------------------------
 	// RxJava 1 converters
 	// -------------------------------------------------------------------------
@@ -1064,6 +1097,38 @@ public class ReactiveWrapperConverters {
 		}
 	}
 
+	/**
+	 * A {@link ConverterFactory} that adapts between reactive types using {@link ReactiveAdapterRegistry}.
+	 */
+	private enum ReactiveAdapterConverterFactory implements ConverterFactory<Object, Object>, ConditionalConverter {
+
+		INSTANCE;
+
+		@Override
+		public boolean matches(TypeDescriptor sourceType, TypeDescriptor targetType) {
+			return isSupported(sourceType) || isSupported(targetType);
+		}
+
+		private boolean isSupported(TypeDescriptor typeDescriptor) {
+			return REACTIVE_ADAPTER_REGISTRY != null
+					&& REACTIVE_ADAPTER_REGISTRY.getAdapter(typeDescriptor.getType()) != null;
+		}
+
+		@Override
+		@SuppressWarnings({ "ConstantConditions", "unchecked" })
+		public <T> Converter<Object, T> getConverter(Class<T> targetType) {
+			return source -> {
+
+				Publisher<?> publisher = source instanceof Publisher ? (Publisher<?>) source
+						: REACTIVE_ADAPTER_REGISTRY.getAdapter(Publisher.class, source).toPublisher(source);
+
+				ReactiveAdapter adapter = REACTIVE_ADAPTER_REGISTRY.getAdapter(targetType);
+
+				return (T) adapter.fromPublisher(publisher);
+			};
+		}
+	}
+
 	/**
 	 * Holder for delayed initialization of {@link ReactiveAdapterRegistry}.
 	 *

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.repository.util;
 
+import kotlin.Unit;
+import kotlinx.coroutines.flow.Flow;
 import lombok.experimental.UtilityClass;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -30,6 +32,7 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.core.ReactiveTypeDescriptor;
 import org.springframework.data.util.ProxyUtils;
 import org.springframework.data.util.ReflectionUtils;
@@ -71,6 +74,11 @@ public class ReactiveWrappers {
 	private static final boolean RXJAVA2_PRESENT = ClassUtils.isPresent("io.reactivex.Flowable",
 			ReactiveWrappers.class.getClassLoader());
 
+	private static final boolean KOTLIN_COROUTINES_PRESENT = ClassUtils.isPresent("kotlinx.coroutines.flow.Flow",
+			ReactiveWrappers.class.getClassLoader())
+			&& ClassUtils.isPresent("kotlinx.coroutines.reactive.ReactiveFlowKt", ReactiveWrappers.class.getClassLoader())
+			&& ClassUtils.isPresent("kotlinx.coroutines.reactor.ReactorFlowKt", ReactiveWrappers.class.getClassLoader());
+
 	private static final Collection<ReactiveTypeDescriptor> REACTIVE_WRAPPERS;
 
 	/**
@@ -79,7 +87,7 @@ public class ReactiveWrappers {
 	 * @author Mark Paluch
 	 */
 	static enum ReactiveLibrary {
-		PROJECT_REACTOR, RXJAVA1, RXJAVA2;
+		PROJECT_REACTOR, RXJAVA1, RXJAVA2, KOTLIN_COROUTINES;
 	}
 
 	static {
@@ -111,6 +119,13 @@ public class ReactiveWrappers {
 			reactiveWrappers.add(ReactiveTypeDescriptor.singleOptionalValue(Mono.class, Mono::empty));
 			reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Flux.class, Flux::empty));
 			reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Publisher.class, Flux::empty));
+
+			if (KOTLIN_COROUTINES_PRESENT) {
+
+				reactiveWrappers.add(ReactiveTypeDescriptor.multiValue(Flow.class, () -> {
+					return (Flow<Object>) (flowCollector, continuation) -> Unit.INSTANCE;
+				}));
+			}
 		}
 
 		REACTIVE_WRAPPERS = Collections.unmodifiableCollection(reactiveWrappers);
@@ -143,6 +158,8 @@ public class ReactiveWrappers {
 				return RXJAVA1_PRESENT;
 			case RXJAVA2:
 				return RXJAVA2_PRESENT;
+			case KOTLIN_COROUTINES:
+				return PROJECT_REACTOR_PRESENT && KOTLIN_COROUTINES_PRESENT;
 			default:
 				throw new IllegalArgumentException(String.format("Reactive library %s not supported", reactiveLibrary));
 		}

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

@@ -0,0 +1,171 @@
+/*
+ * Copyright 2019 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.util;
+
+import kotlin.jvm.JvmClassMappingKt;
+import kotlin.reflect.KCallable;
+import kotlin.reflect.KClass;
+import kotlin.reflect.KFunction;
+import kotlin.reflect.KMutableProperty;
+import kotlin.reflect.KProperty;
+import kotlin.reflect.KType;
+import kotlin.reflect.jvm.KTypesJvm;
+import kotlin.reflect.jvm.ReflectJvmMapping;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.lang.Nullable;
+
+/**
+ * Reflection utility methods specific to Kotlin reflection. Requires Kotlin classes to be present to avoid linkage
+ * errors.
+ *
+ * @author Mark Paluch
+ * @since 2.3
+ */
+public final class KotlinReflectionUtils {
+
+	private KotlinReflectionUtils() {}
+
+	/**
+	 * Returns a {@link KFunction} instance corresponding to the given Java {@link Method} instance, or {@code null} if
+	 * this method cannot be represented by a Kotlin function.
+	 *
+	 * @param method the method to look up.
+	 * @return the {@link KFunction} or {@code null} if the method cannot be looked up.
+	 */
+	@Nullable
+	public static KFunction<?> findKotlinFunction(Method method) {
+
+		KFunction<?> kotlinFunction = ReflectJvmMapping.getKotlinFunction(method);
+
+		if (kotlinFunction == null) {
+
+			// Fallback to own lookup because there's no public Kotlin API for that kind of lookup until
+			// https://youtrack.jetbrains.com/issue/KT-20768 gets resolved.
+			return findKFunction(method).orElse(null);
+		}
+
+		return kotlinFunction;
+	}
+
+	/**
+	 * Returns the {@link Class return type} of a Kotlin {@link Method}. Supports regular and suspended methods.
+	 *
+	 * @param method the method to inspect, typically any synthetic JVM {@link Method}.
+	 * @return return type of the method.
+	 */
+	public static Class<?> getReturnType(Method method) {
+
+		KFunction<?> kotlinFunction = KotlinReflectionUtils.findKotlinFunction(method);
+
+		if (kotlinFunction == null) {
+			throw new IllegalArgumentException(String.format("Cannot resolve %s to a KFunction!", method));
+		}
+
+		return JvmClassMappingKt.getJavaClass(KTypesJvm.getJvmErasure(kotlinFunction.getReturnType()));
+	}
+
+	/**
+	 * Returns {@literal} whether the given {@link MethodParameter} is nullable. Its declaring method can reference a
+	 * Kotlin function, property or interface property.
+	 *
+	 * @return {@literal true} if {@link MethodParameter} is nullable.
+	 * @since 2.0.1
+	 */
+	static boolean isNullable(MethodParameter parameter) {
+
+		Method method = parameter.getMethod();
+
+		if (method == null) {
+			throw new IllegalStateException(String.format("Cannot obtain method from parameter %s!", parameter));
+		}
+
+		KFunction<?> kotlinFunction = findKotlinFunction(method);
+
+		if (kotlinFunction == null) {
+			throw new IllegalArgumentException(String.format("Cannot resolve %s to a Kotlin function!", parameter));
+		}
+
+		// Special handling for Coroutines
+		if (kotlinFunction.isSuspend() && isLast(parameter)) {
+			return false;
+		}
+
+		// see https://github.com/spring-projects/spring-framework/issues/23991
+		if (kotlinFunction.getParameters().size() > parameter.getParameterIndex() + 1) {
+
+			KType type = parameter.getParameterIndex() == -1 //
+					? kotlinFunction.getReturnType() //
+					: kotlinFunction.getParameters().get(parameter.getParameterIndex() + 1).getType();
+
+			return type.isMarkedNullable();
+		}
+
+		return true;
+	}
+
+	private static boolean isLast(MethodParameter parameter) {
+		return parameter.getParameterIndex() == parameter.getMethod().getParameterCount() - 1;
+	}
+
+	/**
+	 * Lookup a {@link Method} to a {@link KFunction}.
+	 *
+	 * @param method the JVM {@link Method} to look up.
+	 * @return {@link Optional} wrapping a possibly existing {@link KFunction}.
+	 */
+	private static Optional<? extends KFunction<?>> findKFunction(Method method) {
+
+		KClass<?> kotlinClass = JvmClassMappingKt.getKotlinClass(method.getDeclaringClass());
+
+		return kotlinClass.getMembers() //
+				.stream() //
+				.flatMap(KotlinReflectionUtils::toKFunctionStream) //
+				.filter(it -> isSame(it, method)) //
+				.findFirst();
+	}
+
+	private static Stream<? extends KFunction<?>> toKFunctionStream(KCallable<?> it) {
+
+		if (it instanceof KMutableProperty<?>) {
+
+			KMutableProperty<?> property = (KMutableProperty<?>) it;
+			return Stream.of(property.getGetter(), property.getSetter());
+		}
+
+		if (it instanceof KProperty<?>) {
+
+			KProperty<?> property = (KProperty<?>) it;
+			return Stream.of(property.getGetter());
+		}
+
+		if (it instanceof KFunction<?>) {
+			return Stream.of((KFunction<?>) it);
+		}
+
+		return Stream.empty();
+	}
+
+	private static boolean isSame(KFunction<?> function, Method method) {
+
+		Method javaMethod = ReflectJvmMapping.getJavaMethod(function);
+		return javaMethod != null && javaMethod.equals(method);
+	}
+}

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

@@ -15,14 +15,6 @@
  */
 package org.springframework.data.util;
 
-import kotlin.jvm.JvmClassMappingKt;
-import kotlin.reflect.KCallable;
-import kotlin.reflect.KClass;
-import kotlin.reflect.KFunction;
-import kotlin.reflect.KMutableProperty;
-import kotlin.reflect.KProperty;
-import kotlin.reflect.KType;
-import kotlin.reflect.jvm.ReflectJvmMapping;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.experimental.UtilityClass;
@@ -449,86 +441,4 @@ public class ReflectionUtils {
 		throw new IllegalArgumentException(String.format("Primitive type %s not supported!", type));
 	}
 
-	/**
-	 * Reflection utility methods specific to Kotlin reflection.
-	 */
-	static class KotlinReflectionUtils {
-
-		/**
-		 * Returns {@literal} whether the given {@link MethodParameter} is nullable. Its declaring method can reference a
-		 * Kotlin function, property or interface property.
-		 *
-		 * @return {@literal true} if {@link MethodParameter} is nullable.
-		 * @since 2.0.1
-		 */
-		static boolean isNullable(MethodParameter parameter) {
-
-			Method method = parameter.getMethod();
-
-			if (method == null) {
-				throw new IllegalStateException(String.format("Cannot obtain method from parameter %s!", parameter));
-			}
-
-			KFunction<?> kotlinFunction = ReflectJvmMapping.getKotlinFunction(method);
-
-			if (kotlinFunction == null) {
-
-				// Fallback to own lookup because there's no public Kotlin API for that kind of lookup until
-				// https://youtrack.jetbrains.com/issue/KT-20768 gets resolved.
-				kotlinFunction = findKFunction(method)//
-						.orElseThrow(() -> new IllegalArgumentException(
-								String.format("Cannot resolve %s to a Kotlin function!", parameter)));
-			}
-
-			KType type = parameter.getParameterIndex() == -1 //
-					? kotlinFunction.getReturnType() //
-					: kotlinFunction.getParameters().get(parameter.getParameterIndex() + 1).getType();
-
-			return type.isMarkedNullable();
-		}
-
-		/**
-		 * Lookup a {@link Method} to a {@link KFunction}.
-		 *
-		 * @param method the JVM {@link Method} to look up.
-		 * @return {@link Optional} wrapping a possibly existing {@link KFunction}.
-		 */
-		private static Optional<? extends KFunction<?>> findKFunction(Method method) {
-
-			KClass<?> kotlinClass = JvmClassMappingKt.getKotlinClass(method.getDeclaringClass());
-
-			return kotlinClass.getMembers() //
-					.stream() //
-					.flatMap(KotlinReflectionUtils::toKFunctionStream) //
-					.filter(it -> isSame(it, method)) //
-					.findFirst();
-		}
-
-		private static Stream<? extends KFunction<?>> toKFunctionStream(KCallable<?> it) {
-
-			if (it instanceof KMutableProperty<?>) {
-
-				KMutableProperty<?> property = (KMutableProperty<?>) it;
-				return Stream.of(property.getGetter(), property.getSetter());
-			}
-
-			if (it instanceof KProperty<?>) {
-
-				KProperty<?> property = (KProperty<?>) it;
-				return Stream.of(property.getGetter());
-			}
-
-			if (it instanceof KFunction<?>) {
-				return Stream.of((KFunction<?>) it);
-			}
-
-			return Stream.empty();
-		}
-
-		private static boolean isSame(KFunction<?> function, Method method) {
-
-			Method javaMethod = ReflectJvmMapping.getJavaMethod(function);
-			return javaMethod != null && javaMethod.equals(method);
-		}
-	}
 }