DATACMNS-1763 - Allow registration of QueryMethod invocation listeners.

We now allow for registering RepositoryMethodInvocationListeners to notify listeners upon repository method invocations (query methods and repository fragments). Listeners are notified upon method completion reporting the repository interface, the invoked methods, duration, outcome and arguments.

Original Pull Request: #455

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

@@ -1,112 +0,0 @@
-/*
- * Copyright 2019-2020 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 suspended
- * 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 = KotlinDetector.isKotlinType(declaredMethod.getDeclaringClass())
-				? KotlinReflectionUtils.findKotlinFunction(declaredMethod)
-				: null;
-		KFunction<?> baseClassFunction = KotlinDetector.isKotlinType(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 {
-
-		return shouldAdaptReactiveToSuspended() ? invokeReactiveToSuspend(methodToCall, instance, args)
-				: methodToCall.invoke(instance, args);
-
-	}
-
-	private boolean shouldAdaptReactiveToSuspended() {
-		return suspendedDeclaredMethod && !suspendedBaseClassMethod && reactiveBaseClassMethod;
-	}
-
-	@Nullable
-	@SuppressWarnings({ "unchecked", "ConstantConditions" })
-	private Object invokeReactiveToSuspend(Method methodToCall, Object instance, Object[] args)
-			throws ReflectiveOperationException {
-
-		/*
-		 * Kotlin suspended functions are invoked with a synthetic Continuation parameter that keeps track of the Coroutine context.
-		 * We're invoking a method without Continuation as we expect the method to return any sort of reactive type,
-		 * therefore we need to strip the Continuation parameter.
-		 */
-		Object[] invocationArguments = new Object[args.length - 1];
-		System.arraycopy(args, 0, invocationArguments, 0, invocationArguments.length);
-		Object result = methodToCall.invoke(instance, invocationArguments);
-
-		Publisher<?> publisher = result instanceof Publisher ? (Publisher<?>) result
-				: 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

@@ -379,10 +379,7 @@ interface MethodLookups {
 		}
 
 		private static boolean parameterCountMatch(InvokedMethod invokedMethod, Method baseClassMethod) {
-
-			ImplementationInvocationMetadata invocationMetadata = new ImplementationInvocationMetadata(
-					invokedMethod.getMethod(), baseClassMethod);
-			return invocationMetadata.canInvoke(invokedMethod.getMethod(), baseClassMethod);
+			return RepositoryMethodInvoker.canInvoke(invokedMethod.getMethod(), baseClassMethod);
 		}
 
 		private static Stream<ParameterOverrideCriteria> methodParameters(Method invokedMethod, Method baseClassMethod) {

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

@@ -15,9 +15,6 @@
  */
 package org.springframework.data.repository.core.support;
 
-import kotlin.coroutines.Continuation;
-import kotlinx.coroutines.reactive.AwaitKt;
-
 import java.lang.reflect.Method;
 import java.util.Collections;
 import java.util.List;
@@ -26,9 +23,7 @@ import java.util.Optional;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
-import org.reactivestreams.Publisher;
 
-import org.springframework.core.KotlinDetector;
 import org.springframework.core.ResolvableType;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.repository.core.NamedQueries;
@@ -37,9 +32,6 @@ import org.springframework.data.repository.query.QueryLookupStrategy;
 import org.springframework.data.repository.query.QueryMethod;
 import org.springframework.data.repository.query.RepositoryQuery;
 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.lang.Nullable;
 import org.springframework.util.ConcurrentReferenceHashMap;
@@ -54,11 +46,13 @@ import org.springframework.util.ConcurrentReferenceHashMap;
  */
 class QueryExecutorMethodInterceptor implements MethodInterceptor {
 
+	private final RepositoryInformation repositoryInformation;
 	private final Map<Method, RepositoryQuery> queries;
-	private final Map<Method, QueryMethodInvoker> invocationMetadataCache = new ConcurrentReferenceHashMap<>();
+	private final Map<Method, RepositoryMethodInvoker> invocationMetadataCache = new ConcurrentReferenceHashMap<>();
 	private final QueryExecutionResultHandler resultHandler;
 	private final NamedQueries namedQueries;
 	private final List<QueryCreationListener<?>> queryPostProcessors;
+	private final List<RepositoryMethodInvocationListener> methodInvocationListeners;
 
 	/**
 	 * Creates a new {@link QueryExecutorMethodInterceptor}. Builds a model of {@link QueryMethod}s to be invoked on
@@ -66,10 +60,13 @@ class QueryExecutorMethodInterceptor implements MethodInterceptor {
 	 */
 	public QueryExecutorMethodInterceptor(RepositoryInformation repositoryInformation,
 			ProjectionFactory projectionFactory, Optional<QueryLookupStrategy> queryLookupStrategy, NamedQueries namedQueries,
-			List<QueryCreationListener<?>> queryPostProcessors) {
+			List<QueryCreationListener<?>> queryPostProcessors,
+			List<RepositoryMethodInvocationListener> methodInvocationListeners) {
 
+		this.repositoryInformation = repositoryInformation;
 		this.namedQueries = namedQueries;
 		this.queryPostProcessors = queryPostProcessors;
+		this.methodInvocationListeners = methodInvocationListeners;
 
 		this.resultHandler = new QueryExecutionResultHandler(RepositoryFactorySupport.CONVERSION_SERVICE);
 
@@ -141,20 +138,27 @@ class QueryExecutorMethodInterceptor implements MethodInterceptor {
 
 		if (hasQueryFor(method)) {
 
-			QueryMethodInvoker invocationMetadata = invocationMetadataCache.get(method);
+			RepositoryInvocationListener invocationListener = getInvocationListener();
+			RepositoryMethodInvoker invocationMetadata = invocationMetadataCache.get(method);
 
 			if (invocationMetadata == null) {
-				invocationMetadata = new QueryMethodInvoker(method);
+				invocationMetadata = RepositoryMethodInvoker.forRepositoryQuery(method, queries.get(method));
 				invocationMetadataCache.put(method, invocationMetadata);
 			}
 
-			RepositoryQuery repositoryQuery = queries.get(method);
-			return invocationMetadata.invoke(repositoryQuery, invocation.getArguments());
+			return invocationMetadata.invoke(invocationListener, invocation.getArguments());
 		}
 
 		return invocation.proceed();
 	}
 
+	private RepositoryInvocationListener getInvocationListener() {
+		return methodInvocationListeners.isEmpty() ? RepositoryInvocationListener.NoOpRepositoryInvocationListener.INSTANCE
+				: new RepositoryInvocationListener.RepositoryInvocationMulticaster(
+						repositoryInformation.getRepositoryInterface(), methodInvocationListeners);
+
+	}
+
 	/**
 	 * Returns whether we know of a query to execute for the given {@link Method};
 	 *
@@ -165,57 +169,4 @@ class QueryExecutorMethodInterceptor implements MethodInterceptor {
 		return queries.containsKey(method);
 	}
 
-	/**
-	 * Invoker for Query Methods. Considers
-	 */
-	static class QueryMethodInvoker {
-
-		private final boolean suspendedDeclaredMethod;
-		private final Class<?> returnedType;
-		private final boolean returnsReactiveType;
-
-		QueryMethodInvoker(Method invokedMethod) {
-
-			if (KotlinDetector.isKotlinReflectPresent()) {
-
-				this.suspendedDeclaredMethod = KotlinReflectionUtils.isSuspend(invokedMethod);
-				this.returnedType = this.suspendedDeclaredMethod ? KotlinReflectionUtils.getReturnType(invokedMethod)
-						: invokedMethod.getReturnType();
-			} else {
-
-				this.suspendedDeclaredMethod = false;
-				this.returnedType = invokedMethod.getReturnType();
-			}
-
-			this.returnsReactiveType = ReactiveWrappers.supports(returnedType);
-		}
-
-		@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) {
-
-			/*
-			* Kotlin suspended functions are invoked with a synthetic Continuation parameter that keeps track of the Coroutine context.
-			* We're invoking a method without Continuation as we expect the method to return any sort of reactive type,
-			* therefore we need to strip the Continuation parameter.
-			*/
-			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 = result instanceof Publisher ? (Publisher<?>) result
-					: ReactiveWrapperConverters.toWrapper(result, Publisher.class);
-
-			return AwaitKt.awaitFirstOrNull(publisher, continuation);
-		}
-	}
 }

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

@@ -194,6 +194,18 @@ public class RepositoryComposition {
 	 * @throws Throwable
 	 */
 	public Object invoke(Method method, Object... args) throws Throwable {
+		return invoke(RepositoryInvocationListener.NoOpRepositoryInvocationListener.INSTANCE, method, args);
+	}
+
+	/**
+	 * Invoke a method on the repository by routing the invocation to the appropriate {@link RepositoryFragment}.
+	 *
+	 * @param method
+	 * @param args
+	 * @return
+	 * @throws Throwable
+	 */
+	Object invoke(RepositoryInvocationListener listener, Method method, Object[] args) throws Throwable {
 
 		Method methodToCall = getMethod(method);
 
@@ -203,7 +215,7 @@ public class RepositoryComposition {
 
 		ReflectionUtils.makeAccessible(methodToCall);
 
-		return fragments.invoke(method, methodToCall, argumentConverter.apply(methodToCall, args));
+		return fragments.invoke(listener, method, methodToCall, argumentConverter.apply(methodToCall, args));
 	}
 
 	/**
@@ -290,7 +302,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 Map<Method, RepositoryMethodInvoker> invocationMetadataCache = new ConcurrentHashMap<>();
 		private final List<RepositoryFragment<?>> fragments;
 
 		private RepositoryFragments(List<RepositoryFragment<?>> fragments) {
@@ -397,8 +409,10 @@ public class RepositoryComposition {
 		}
 
 		/**
-		 * Invoke {@link Method} by resolving the
+		 * Invoke {@link Method} by resolving the fragment that implements a suitable method.
 		 *
+		 * @param repositoryInformation
+		 * @param listener
 		 * @param invokedMethod invoked method as per invocation on the interface.
 		 * @param methodToCall backend method that is backing the call.
 		 * @param args
@@ -407,6 +421,25 @@ public class RepositoryComposition {
 		 */
 		@Nullable
 		public Object invoke(Method invokedMethod, Method methodToCall, Object[] args) throws Throwable {
+			return invoke(RepositoryInvocationListener.NoOpRepositoryInvocationListener.INSTANCE, invokedMethod, methodToCall,
+					args);
+		}
+
+		/**
+		 * Invoke {@link Method} by resolving the fragment that implements a suitable method.
+		 *
+		 * @param listener
+		 * @param repositoryInformation
+		 * @param listener
+		 * @param invokedMethod invoked method as per invocation on the interface.
+		 * @param methodToCall backend method that is backing the call.
+		 * @param args
+		 * @return
+		 * @throws Throwable
+		 */
+		@Nullable
+		Object invoke(RepositoryInvocationListener listener, Method invokedMethod, Method methodToCall, Object[] args)
+				throws Throwable {
 
 			RepositoryFragment<?> fragment = fragmentCache.computeIfAbsent(methodToCall, this::findImplementationFragment);
 			Optional<?> optional = fragment.getImplementation();
@@ -415,14 +448,14 @@ public class RepositoryComposition {
 				throw new IllegalArgumentException(String.format("No implementation found for method %s", methodToCall));
 			}
 
-			ImplementationInvocationMetadata invocationMetadata = invocationMetadataCache.get(invokedMethod);
+			RepositoryMethodInvoker invocationMetadata = invocationMetadataCache.get(invokedMethod);
 
 			if (invocationMetadata == null) {
-				invocationMetadata = new ImplementationInvocationMetadata(invokedMethod, methodToCall);
+				invocationMetadata = RepositoryMethodInvoker.forFragmentMethod(invokedMethod, optional.get(), methodToCall);
 				invocationMetadataCache.put(invokedMethod, invocationMetadata);
 			}
 
-			return invocationMetadata.invoke(methodToCall, optional.get(), args);
+			return invocationMetadata.invoke(listener, args);
 		}
 
 		private RepositoryFragment<?> findImplementationFragment(Method key) {

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

@@ -123,6 +123,7 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 	private Optional<Class<?>> repositoryBaseClass;
 	private @Nullable QueryLookupStrategy.Key queryLookupStrategyKey;
 	private List<QueryCreationListener<?>> queryPostProcessors;
+	private List<RepositoryMethodInvocationListener> methodInvocationListeners;
 	private NamedQueries namedQueries;
 	private ClassLoader classLoader;
 	private QueryMethodEvaluationContextProvider evaluationContextProvider;
@@ -142,6 +143,7 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 		this.evaluationContextProvider = QueryMethodEvaluationContextProvider.DEFAULT;
 		this.queryPostProcessors = new ArrayList<>();
 		this.queryPostProcessors.add(collectingListener);
+		this.methodInvocationListeners = new ArrayList<>();
 	}
 
 	/**
@@ -185,7 +187,7 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 	 * queries.
 	 *
 	 * @param evaluationContextProvider can be {@literal null}, defaults to
-	 *          {@link DefaultQueryMethodEvaluationContextProvider#INSTANCE}.
+	 *          {@link QueryMethodEvaluationContextProvider#INSTANCE}.
 	 */
 	public void setEvaluationContextProvider(QueryMethodEvaluationContextProvider evaluationContextProvider) {
 		this.evaluationContextProvider = evaluationContextProvider == null ? QueryMethodEvaluationContextProvider.DEFAULT
@@ -215,6 +217,19 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 		this.queryPostProcessors.add(listener);
 	}
 
+	/**
+	 * Adds a {@link RepositoryMethodInvocationListener} to the factory to plug in functionality triggered right after
+	 * running {@link RepositoryQuery query methods} and {@link Method fragment methods}.
+	 *
+	 * @param listener
+	 * @since 2.4
+	 */
+	public void addInvocationListener(RepositoryMethodInvocationListener listener) {
+
+		Assert.notNull(listener, "Listener must not be null!");
+		this.methodInvocationListeners.add(listener);
+	}
+
 	/**
 	 * Adds {@link RepositoryProxyPostProcessor}s to the factory to allow manipulation of the {@link ProxyFactory} before
 	 * the proxy gets created. Note that the {@link QueryExecutorMethodInterceptor} will be added to the proxy
@@ -326,10 +341,10 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 		Optional<QueryLookupStrategy> queryLookupStrategy = getQueryLookupStrategy(queryLookupStrategyKey,
 				evaluationContextProvider);
 		result.addAdvice(new QueryExecutorMethodInterceptor(information, projectionFactory, queryLookupStrategy,
-				namedQueries, queryPostProcessors));
+				namedQueries, queryPostProcessors, methodInvocationListeners));
 
 		composition = composition.append(RepositoryFragment.implemented(target));
-		result.addAdvice(new ImplementationMethodExecutionInterceptor(composition));
+		result.addAdvice(new ImplementationMethodExecutionInterceptor(information, composition, methodInvocationListeners));
 
 		T repository = (T) result.getProxy(classLoader);
 
@@ -526,12 +541,17 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 	 *
 	 * @author Mark Paluch
 	 */
-	public class ImplementationMethodExecutionInterceptor implements MethodInterceptor {
+	static class ImplementationMethodExecutionInterceptor implements MethodInterceptor {
 
+		private final RepositoryInformation information;
 		private final RepositoryComposition composition;
+		private final List<RepositoryMethodInvocationListener> methodInvocationListeners;
 
-		public ImplementationMethodExecutionInterceptor(RepositoryComposition composition) {
+		public ImplementationMethodExecutionInterceptor(RepositoryInformation information,
+				RepositoryComposition composition, List<RepositoryMethodInvocationListener> methodInvocationListeners) {
+			this.information = information;
 			this.composition = composition;
+			this.methodInvocationListeners = methodInvocationListeners;
 		}
 
 		/*
@@ -544,15 +564,23 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware,
 
 			Method method = invocation.getMethod();
 			Object[] arguments = invocation.getArguments();
+			RepositoryInvocationListener invocationListener = getInvocationListener();
 
 			try {
-				return composition.invoke(method, arguments);
+				return composition.invoke(invocationListener, method, arguments);
 			} catch (Exception e) {
 				ClassUtils.unwrapReflectionException(e);
 			}
 
 			throw new IllegalStateException("Should not occur!");
 		}
+
+		private RepositoryInvocationListener getInvocationListener() {
+			return methodInvocationListeners.isEmpty()
+					? RepositoryInvocationListener.NoOpRepositoryInvocationListener.INSTANCE
+					: new RepositoryInvocationListener.RepositoryInvocationMulticaster(information.getRepositoryInterface(),
+							methodInvocationListeners);
+		}
 	}
 
 	/**

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

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2020 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 java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.List;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Interface to be implemented by repository method listeners. Listeners are notified with the called {@link Method},
+ * arguments and its outcome.
+ *
+ * @author Mark Paluch
+ * @since 2.4
+ */
+interface RepositoryInvocationListener {
+
+	/**
+	 * The repository method invocation to be handled after calling it.
+	 *
+	 * @param method
+	 * @param args
+	 * @param result
+	 * @param exception
+	 */
+	void afterInvocation(Method method, Object[] args, @Nullable Object result, @Nullable Throwable exception);
+
+	/**
+	 * {@link RepositoryInvocationListener} that does nothing upon invocation.
+	 *
+	 * @author Mark Paluch
+	 */
+	enum NoOpRepositoryInvocationListener implements RepositoryInvocationListener {
+
+		INSTANCE;
+
+		@Override
+		public void afterInvocation(Method method, Object[] args, @Nullable Object result, @Nullable Throwable exception) {
+
+		}
+	}
+
+	/**
+	 * {@link RepositoryInvocationListener} implementation that notifies {@link RepositoryMethodInvocationListener} upon
+	 * {@link #afterInvocation(Method, Object[], Object, Throwable)}.
+	 *
+	 * @author Mark Paluch
+	 */
+	class RepositoryInvocationMulticaster implements RepositoryInvocationListener {
+
+		private final Class<?> repositoryInterface;
+		private final List<RepositoryMethodInvocationListener> methodInvocationListeners;
+		private final long startNs;
+
+		RepositoryInvocationMulticaster(Class<?> repositoryInterface,
+				List<RepositoryMethodInvocationListener> methodInvocationListeners) {
+
+			this.repositoryInterface = repositoryInterface;
+			this.methodInvocationListeners = methodInvocationListeners;
+			this.startNs = System.nanoTime();
+		}
+
+		@Override
+		public void afterInvocation(Method method, Object[] args, @Nullable Object result, @Nullable Throwable exception) {
+
+			long durationNs = getDuration(System.nanoTime());
+			RepositoryMethodInvocationListener.Invocation invocation = new RepositoryMethodInvocationListener.Invocation(
+					durationNs, repositoryInterface, method, result,
+					exception instanceof InvocationTargetException ? exception.getCause() : exception);
+
+			for (RepositoryMethodInvocationListener methodInvocationListener : methodInvocationListeners) {
+				methodInvocationListener.afterInvocation(invocation);
+			}
+		}
+
+		private long getDuration(long endNs) {
+
+			if (endNs > startNs) {
+				return endNs - startNs;
+			}
+
+			// end time overflow
+			return (Long.MAX_VALUE - startNs) + endNs;
+		}
+	}
+}

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

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2020 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 java.lang.reflect.Method;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Interface to be implemented by listeners that want to be notified upon repository method invocation. Listeners are
+ * notified with an {@link Invocation} object that describes which repository method was invoked along with invocation
+ * arguments and the call duration.
+ *
+ * @author Mark Paluch
+ * @since 2.4
+ */
+public interface RepositoryMethodInvocationListener {
+
+	/**
+	 * Handle the invocation event. This method is called after the execution has finished.
+	 *
+	 * @param invocation the invocation to respond to.
+	 */
+	void afterInvocation(Invocation invocation);
+
+	/**
+	 * Value object capturing the actual invocation.
+	 */
+	class Invocation {
+
+		private final long durationNs;
+		private final Class<?> repositoryInterface;
+		private final Method method;
+		private final Object result;
+		private final Throwable exception;
+
+		/**
+		 * @param durationNs the duration in {@link TimeUnit#NANOSECONDS}.
+		 * @param repositoryInterface the repository interface that was used to call {@link Method}.
+		 * @param method the actual method that was called.
+		 * @param result the outcome of the method call. May be {@literal null}.
+		 * @param exception may be {@literal null}.
+		 */
+		public Invocation(long durationNs, Class<?> repositoryInterface, Method method, @Nullable Object result,
+				@Nullable Throwable exception) {
+			this.durationNs = durationNs;
+			this.repositoryInterface = repositoryInterface;
+			this.method = method;
+			this.result = result;
+			this.exception = exception;
+		}
+
+		public long getDuration(TimeUnit timeUnit) {
+
+			Assert.notNull(timeUnit, "TimeUnit must not be null");
+
+			return timeUnit.convert(durationNs, TimeUnit.NANOSECONDS);
+		}
+
+		public Class<?> getRepositoryInterface() {
+			return repositoryInterface;
+		}
+
+		public Method getMethod() {
+			return method;
+		}
+
+		@Nullable
+		public Object getResult() {
+			return result;
+		}
+
+		@Nullable
+		public Throwable getException() {
+			return exception;
+		}
+	}
+}

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

@@ -0,0 +1,304 @@
+/*
+ * Copyright 2020 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 java.util.stream.Stream;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.KotlinDetector;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.data.repository.util.ReactiveWrappers;
+import org.springframework.data.util.KotlinReflectionUtils;
+import org.springframework.lang.Nullable;
+
+/**
+ * Invoker for repository methods. Used to invoke query methods and fragment methods. This invoker considers Kotlin
+ * coroutine method adaption by either forwarding the entire call or by bridging invocations over a reactive
+ * implementation.
+ *
+ * @author Mark Paluch
+ * @since 2.4
+ * @see #forFragmentMethod(Method, Object, Method)
+ * @see #forRepositoryQuery(Method, RepositoryQuery)
+ * @see RepositoryQuery
+ * @see RepositoryComposition
+ */
+abstract class RepositoryMethodInvoker {
+
+	private final Method method;
+	private final Class<?> returnedType;
+	private final Invokable invokable;
+	private final boolean suspendedDeclaredMethod;
+	private final boolean returnsReactiveType;
+
+	protected RepositoryMethodInvoker(Method method, Invokable invokable) {
+
+		this.method = method;
+		this.invokable = invokable;
+
+		if (KotlinDetector.isKotlinReflectPresent()) {
+
+			this.suspendedDeclaredMethod = KotlinReflectionUtils.isSuspend(method);
+			this.returnedType = this.suspendedDeclaredMethod ? KotlinReflectionUtils.getReturnType(method)
+					: method.getReturnType();
+		} else {
+
+			this.suspendedDeclaredMethod = false;
+			this.returnedType = method.getReturnType();
+		}
+
+		this.returnsReactiveType = ReactiveWrappers.supports(returnedType);
+	}
+
+	static RepositoryQueryMethodInvoker forRepositoryQuery(Method declaredMethod, RepositoryQuery query) {
+		return new RepositoryQueryMethodInvoker(declaredMethod, query);
+	}
+
+	/**
+	 * Create a {@link RepositoryMethodInvoker} to call a fragment {@link Method}.
+	 *
+	 * @param declaredMethod the declared repository method from the repository interface.
+	 * @param instance fragment instance.
+	 * @param baseMethod the base method to call on fragment {@code instance}.
+	 * @return {@link RepositoryMethodInvoker} to call a fragment {@link Method}.
+	 */
+	static RepositoryMethodInvoker forFragmentMethod(Method declaredMethod, Object instance, Method baseMethod) {
+		return new RepositoryFragmentMethodInvoker(declaredMethod, instance, baseMethod);
+	}
+
+	/**
+	 * Return whether the {@link Method declared method} can be adapted by calling {@link Method baseClassMethod}.
+	 *
+	 * @param declaredMethod the declared repository method from the repository interface.
+	 * @param baseMethod the base method to call on fragment {@code instance}.
+	 * @return
+	 */
+	public static boolean canInvoke(Method declaredMethod, Method baseClassMethod) {
+		return RepositoryFragmentMethodInvoker.CoroutineAdapterInformation.create(declaredMethod, baseClassMethod)
+				.canInvoke();
+	}
+
+	/**
+	 * Invoke the repository method and return its value.
+	 *
+	 * @param listener listener to notify about the call outcome.
+	 * @param args invocation arguments.
+	 * @return
+	 * @throws Exception
+	 */
+	@Nullable
+	public Object invoke(RepositoryInvocationListener listener, Object[] args) throws Exception {
+		return shouldAdaptReactiveToSuspended() ? doInvokeReactiveToSuspended(listener, args) : doInvoke(listener, args);
+	}
+
+	protected boolean shouldAdaptReactiveToSuspended() {
+		return suspendedDeclaredMethod;
+	}
+
+	@Nullable
+	private Object doInvoke(RepositoryInvocationListener listener, Object[] args) throws Exception {
+
+		try {
+			Object result = invokable.invoke(args);
+
+			if (result != null && ReactiveWrappers.supports(result.getClass())) {
+				return ReactiveWrapperConverters.doOnError(
+						ReactiveWrapperConverters.doOnSuccess(result, () -> listener.afterInvocation(method, args, result, null)),
+						ex -> listener.afterInvocation(method, args, result, ex));
+			}
+
+			if (result instanceof Stream) {
+				return ((Stream<?>) result).onClose(() -> listener.afterInvocation(method, args, result, null));
+			}
+
+			listener.afterInvocation(method, args, result, null);
+
+			return result;
+		} catch (Exception e) {
+			listener.afterInvocation(method, args, null, e);
+			throw e;
+		}
+	}
+
+	@Nullable
+	@SuppressWarnings({ "unchecked", "ConstantConditions" })
+	private Object doInvokeReactiveToSuspended(RepositoryInvocationListener listener, Object[] args) throws Exception {
+
+		/*
+		 * Kotlin suspended functions are invoked with a synthetic Continuation parameter that keeps track of the Coroutine context.
+		 * We're invoking a method without Continuation as we expect the method to return any sort of reactive type,
+		 * therefore we need to strip the Continuation parameter.
+		 */
+		Continuation<Object> continuation = (Continuation) args[args.length - 1];
+		args[args.length - 1] = null;
+		try {
+			Object result = invokable.invoke(args);
+
+			if (returnsReactiveType) {
+				listener.afterInvocation(method, args, result, null);
+				return ReactiveWrapperConverters.toWrapper(result, returnedType);
+			}
+
+			Publisher<?> publisher = result instanceof Publisher ? (Publisher<?>) result
+					: ReactiveWrapperConverters.toWrapper(result, Publisher.class);
+
+			publisher = ReactiveWrapperConverters.doOnError(
+					ReactiveWrapperConverters.doOnSuccess(publisher, () -> listener.afterInvocation(method, args, result, null)),
+					ex -> listener.afterInvocation(method, args, result, ex));
+
+			return AwaitKt.awaitFirstOrNull(publisher, continuation);
+		} catch (Exception e) {
+			listener.afterInvocation(method, args, null, e);
+			throw e;
+		}
+	}
+
+	interface Invokable {
+
+		@Nullable
+		Object invoke(Object[] args) throws ReflectiveOperationException;
+	}
+
+	/**
+	 * Implementation to invoke query methods.
+	 */
+	private static class RepositoryQueryMethodInvoker extends RepositoryMethodInvoker {
+		public RepositoryQueryMethodInvoker(Method method, RepositoryQuery repositoryQuery) {
+			super(method, repositoryQuery::execute);
+		}
+	}
+
+	/**
+	 * Implementation to invoke fragment methods.
+	 */
+	private static class RepositoryFragmentMethodInvoker extends RepositoryMethodInvoker {
+
+		private final CoroutineAdapterInformation adapterInformation;
+
+		public RepositoryFragmentMethodInvoker(Method declaredMethod, Object instance, Method baseClassMethod) {
+			this(CoroutineAdapterInformation.create(declaredMethod, baseClassMethod), declaredMethod, instance,
+					baseClassMethod);
+		}
+
+		public RepositoryFragmentMethodInvoker(CoroutineAdapterInformation adapterInformation, Method declaredMethod,
+				Object instance, Method baseClassMethod) {
+			super(declaredMethod, args -> {
+
+				if (adapterInformation.isAdapterMethod()) {
+
+					/*
+					 * Kotlin suspended functions are invoked with a synthetic Continuation parameter that keeps track of the Coroutine context.
+					 * We're invoking a method without Continuation as we expect the method to return any sort of reactive type,
+					 * therefore we need to strip the Continuation parameter.
+					 */
+					Object[] invocationArguments = new Object[args.length - 1];
+					System.arraycopy(args, 0, invocationArguments, 0, invocationArguments.length);
+
+					return baseClassMethod.invoke(instance, invocationArguments);
+				}
+
+				return baseClassMethod.invoke(instance, args);
+			});
+			this.adapterInformation = adapterInformation;
+		}
+
+		@Override
+		protected boolean shouldAdaptReactiveToSuspended() {
+			return adapterInformation.shouldAdaptReactiveToSuspended();
+		}
+
+		/**
+		 * Value object capturing whether a suspended Kotlin method (Coroutine method) can be bridged with a native or
+		 * reactive fragment method.
+		 */
+		static class CoroutineAdapterInformation {
+
+			private static CoroutineAdapterInformation DISABLED = new CoroutineAdapterInformation(false, false, false, 0, 0);
+
+			private final boolean suspendedDeclaredMethod;
+			private final boolean suspendedBaseClassMethod;
+			private final boolean reactiveBaseClassMethod;
+			private final int declaredMethodParameterCount;
+			private final int baseClassMethodParameterCount;
+
+			private CoroutineAdapterInformation(boolean suspendedDeclaredMethod, boolean suspendedBaseClassMethod,
+					boolean reactiveBaseClassMethod, int declaredMethodParameterCount, int baseClassMethodParameterCount) {
+				this.suspendedDeclaredMethod = suspendedDeclaredMethod;
+				this.suspendedBaseClassMethod = suspendedBaseClassMethod;
+				this.reactiveBaseClassMethod = reactiveBaseClassMethod;
+				this.declaredMethodParameterCount = declaredMethodParameterCount;
+				this.baseClassMethodParameterCount = baseClassMethodParameterCount;
+			}
+
+			/**
+			 * Create {@link CoroutineAdapterInformation}.
+			 *
+			 * @param declaredMethod
+			 * @param baseClassMethod
+			 * @return
+			 */
+			public static CoroutineAdapterInformation create(Method declaredMethod, Method baseClassMethod) {
+
+				if (!KotlinDetector.isKotlinReflectPresent()) {
+					return DISABLED;
+				}
+
+				KFunction<?> declaredFunction = KotlinDetector.isKotlinType(declaredMethod.getDeclaringClass())
+						? KotlinReflectionUtils.findKotlinFunction(declaredMethod)
+						: null;
+				KFunction<?> baseClassFunction = KotlinDetector.isKotlinType(baseClassMethod.getDeclaringClass())
+						? KotlinReflectionUtils.findKotlinFunction(baseClassMethod)
+						: null;
+
+				boolean suspendedDeclaredMethod = declaredFunction != null && declaredFunction.isSuspend();
+				boolean suspendedBaseClassMethod = baseClassFunction != null && baseClassFunction.isSuspend();
+				boolean reactiveBaseClassMethod = !suspendedBaseClassMethod
+						&& ReactiveWrapperConverters.supports(baseClassMethod.getReturnType());
+
+				return new CoroutineAdapterInformation(suspendedDeclaredMethod, suspendedBaseClassMethod,
+						reactiveBaseClassMethod, declaredMethod.getParameterCount(), baseClassMethod.getParameterCount());
+			}
+
+			boolean canInvoke() {
+
+				if (suspendedDeclaredMethod == suspendedBaseClassMethod) {
+					return declaredMethodParameterCount == baseClassMethodParameterCount;
+				}
+
+				if (isAdapterMethod()) {
+					return declaredMethodParameterCount - 1 == baseClassMethodParameterCount;
+				}
+
+				return false;
+			}
+
+			boolean isAdapterMethod() {
+				return suspendedDeclaredMethod && reactiveBaseClassMethod;
+			}
+
+			public boolean shouldAdaptReactiveToSuspended() {
+				return suspendedDeclaredMethod && !suspendedBaseClassMethod && reactiveBaseClassMethod;
+			}
+		}
+	}
+}