From e77947f0a5eaa026fb8e3bb8ed700394e0ab78d1 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 8 Jul 2020 15:40:18 +0200 Subject: [PATCH] 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 --- .../ImplementationInvocationMetadata.java | 112 ------- .../core/support/MethodLookups.java | 5 +- .../QueryExecutorMethodInterceptor.java | 85 ++--- .../core/support/RepositoryComposition.java | 45 ++- .../support/RepositoryFactorySupport.java | 40 ++- .../support/RepositoryInvocationListener.java | 101 ++++++ .../RepositoryMethodInvocationListener.java | 93 ++++++ .../core/support/RepositoryMethodInvoker.java | 304 ++++++++++++++++++ ...eryExecutorMethodInterceptorUnitTests.java | 4 +- .../RepositoryFactorySupportUnitTests.java | 54 +++- .../CoroutineCrudRepositoryUnitTests.kt | 13 +- 11 files changed, 654 insertions(+), 202 deletions(-) delete mode 100644 src/main/java/org/springframework/data/repository/core/support/ImplementationInvocationMetadata.java create mode 100644 src/main/java/org/springframework/data/repository/core/support/RepositoryInvocationListener.java create mode 100644 src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvocationListener.java create mode 100644 src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvoker.java diff --git a/src/main/java/org/springframework/data/repository/core/support/ImplementationInvocationMetadata.java b/src/main/java/org/springframework/data/repository/core/support/ImplementationInvocationMetadata.java deleted file mode 100644 index 3b28f51f8..000000000 --- a/src/main/java/org/springframework/data/repository/core/support/ImplementationInvocationMetadata.java +++ /dev/null @@ -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; - } -} diff --git a/src/main/java/org/springframework/data/repository/core/support/MethodLookups.java b/src/main/java/org/springframework/data/repository/core/support/MethodLookups.java index 5bbe6d01a..a02180f19 100644 --- a/src/main/java/org/springframework/data/repository/core/support/MethodLookups.java +++ b/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 methodParameters(Method invokedMethod, Method baseClassMethod) { diff --git a/src/main/java/org/springframework/data/repository/core/support/QueryExecutorMethodInterceptor.java b/src/main/java/org/springframework/data/repository/core/support/QueryExecutorMethodInterceptor.java index c1940b014..478ef59cb 100644 --- a/src/main/java/org/springframework/data/repository/core/support/QueryExecutorMethodInterceptor.java +++ b/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 queries; - private final Map invocationMetadataCache = new ConcurrentReferenceHashMap<>(); + private final Map invocationMetadataCache = new ConcurrentReferenceHashMap<>(); private final QueryExecutionResultHandler resultHandler; private final NamedQueries namedQueries; private final List> queryPostProcessors; + private final List 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, NamedQueries namedQueries, - List> queryPostProcessors) { + List> queryPostProcessors, + List 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 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); - } - } } diff --git a/src/main/java/org/springframework/data/repository/core/support/RepositoryComposition.java b/src/main/java/org/springframework/data/repository/core/support/RepositoryComposition.java index 4fd0d3b19..43dfd86a1 100644 --- a/src/main/java/org/springframework/data/repository/core/support/RepositoryComposition.java +++ b/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> fragmentCache = new ConcurrentReferenceHashMap<>(); - private final Map invocationMetadataCache = new ConcurrentHashMap<>(); + private final Map invocationMetadataCache = new ConcurrentHashMap<>(); private final List> fragments; private RepositoryFragments(List> 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) { diff --git a/src/main/java/org/springframework/data/repository/core/support/RepositoryFactorySupport.java b/src/main/java/org/springframework/data/repository/core/support/RepositoryFactorySupport.java index 051058692..265e4580e 100644 --- a/src/main/java/org/springframework/data/repository/core/support/RepositoryFactorySupport.java +++ b/src/main/java/org/springframework/data/repository/core/support/RepositoryFactorySupport.java @@ -123,6 +123,7 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware, private Optional> repositoryBaseClass; private @Nullable QueryLookupStrategy.Key queryLookupStrategyKey; private List> queryPostProcessors; + private List 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 = 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 methodInvocationListeners; - public ImplementationMethodExecutionInterceptor(RepositoryComposition composition) { + public ImplementationMethodExecutionInterceptor(RepositoryInformation information, + RepositoryComposition composition, List 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); + } } /** diff --git a/src/main/java/org/springframework/data/repository/core/support/RepositoryInvocationListener.java b/src/main/java/org/springframework/data/repository/core/support/RepositoryInvocationListener.java new file mode 100644 index 000000000..c65a16192 --- /dev/null +++ b/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 methodInvocationListeners; + private final long startNs; + + RepositoryInvocationMulticaster(Class repositoryInterface, + List 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; + } + } +} diff --git a/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvocationListener.java b/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvocationListener.java new file mode 100644 index 000000000..2a989a3d0 --- /dev/null +++ b/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; + } + } +} diff --git a/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvoker.java b/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvoker.java new file mode 100644 index 000000000..00b266728 --- /dev/null +++ b/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 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; + } + } + } +} diff --git a/src/test/java/org/springframework/data/repository/core/support/QueryExecutorMethodInterceptorUnitTests.java b/src/test/java/org/springframework/data/repository/core/support/QueryExecutorMethodInterceptorUnitTests.java index b1ea6b4b1..b59e3929f 100755 --- a/src/test/java/org/springframework/data/repository/core/support/QueryExecutorMethodInterceptorUnitTests.java +++ b/src/test/java/org/springframework/data/repository/core/support/QueryExecutorMethodInterceptorUnitTests.java @@ -50,14 +50,14 @@ class QueryExecutorMethodInterceptorUnitTests { assertThatIllegalStateException() .isThrownBy(() -> new QueryExecutorMethodInterceptor(information, new SpelAwareProxyProjectionFactory(), - Optional.empty(), PropertiesBasedNamedQueries.EMPTY, Collections.emptyList())); + Optional.empty(), PropertiesBasedNamedQueries.EMPTY, Collections.emptyList(), Collections.emptyList())); } @Test // DATACMNS-1508 void skipsQueryLookupsIfQueryLookupStrategyIsNotPresent() { new QueryExecutorMethodInterceptor(information, new SpelAwareProxyProjectionFactory(), Optional.empty(), - PropertiesBasedNamedQueries.EMPTY, Collections.emptyList()); + PropertiesBasedNamedQueries.EMPTY, Collections.emptyList(), Collections.emptyList()); verify(strategy, times(0)).resolveQuery(any(), any(), any(), any()); } diff --git a/src/test/java/org/springframework/data/repository/core/support/RepositoryFactorySupportUnitTests.java b/src/test/java/org/springframework/data/repository/core/support/RepositoryFactorySupportUnitTests.java index 3c5285169..ce4e9820b 100755 --- a/src/test/java/org/springframework/data/repository/core/support/RepositoryFactorySupportUnitTests.java +++ b/src/test/java/org/springframework/data/repository/core/support/RepositoryFactorySupportUnitTests.java @@ -29,10 +29,12 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.junit.jupiter.MockitoExtension; @@ -86,6 +88,7 @@ class RepositoryFactorySupportUnitTests { @Mock PlainQueryCreationListener otherListener; @Mock RepositoryProxyPostProcessor repositoryPostProcessor; + @Mock RepositoryMethodInvocationListener invocationListener; @BeforeEach void setUp() { @@ -118,13 +121,16 @@ class RepositoryFactorySupportUnitTests { verify(repositoryPostProcessor, times(1)).postProcess(any(ProxyFactory.class), any(RepositoryInformation.class)); } - @Test + @Test // DATACMNS-1764 void routesCallToRedeclaredMethodIntoTarget() { + factory.addInvocationListener(invocationListener); + ObjectRepository repository = factory.getRepository(ObjectRepository.class); repository.save(repository); verify(backingRepo, times(1)).save(any(Object.class)); + verify(invocationListener).afterInvocation(any()); } @Test @@ -177,7 +183,7 @@ class RepositoryFactorySupportUnitTests { assertThat(ReflectionTestUtils.getField(factory, "classLoader")).isEqualTo(ClassUtils.getDefaultClassLoader()); } - @Test // DATACMNS-489 + @Test // DATACMNS-489, DATACMNS-1764 void wrapsExecutionResultIntoFutureIfConfigured() throws Exception { final Object reference = new Object(); @@ -187,6 +193,8 @@ class RepositoryFactorySupportUnitTests { return reference; }); + factory.addInvocationListener(invocationListener); + ConvertingRepository repository = factory.getRepository(ConvertingRepository.class); AsyncAnnotationBeanPostProcessor processor = new AsyncAnnotationBeanPostProcessor(); @@ -204,13 +212,44 @@ class RepositoryFactorySupportUnitTests { assertThat(future.get()).isEqualTo(reference); verify(factory.queryOne, times(1)).execute(any(Object[].class)); + + ArgumentCaptor captor = ArgumentCaptor + .forClass(RepositoryMethodInvocationListener.Invocation.class); + verify(invocationListener).afterInvocation(captor.capture()); + + RepositoryMethodInvocationListener.Invocation value = captor.getValue(); + assertThat(value.getResult()).isEqualTo(reference); } - @Test // DATACMNS-509 + @Test // DATACMNS-1764, DATACMNS-1764 + void capturesFailureFromInvocation() { + + when(factory.queryOne.execute(any(Object[].class))).thenThrow(new IllegalStateException()); + + factory.addInvocationListener(invocationListener); + + ConvertingRepository repository = factory.getRepository(ConvertingRepository.class); + + try { + repository.findByLastname("Foo"); + fail("Missing exception"); + } catch (IllegalStateException e) {} + + ArgumentCaptor captor = ArgumentCaptor + .forClass(RepositoryMethodInvocationListener.Invocation.class); + verify(invocationListener).afterInvocation(captor.capture()); + + RepositoryMethodInvocationListener.Invocation invocation = captor.getValue(); + assertThat(invocation.getDuration(TimeUnit.NANOSECONDS)).isGreaterThan(0); + assertThat(invocation.getException()).isInstanceOf(IllegalStateException.class); + } + + @Test // DATACMNS-509, DATACMNS-1764 void convertsWithSameElementType() { List names = singletonList("Dave"); + factory.addInvocationListener(invocationListener); when(factory.queryOne.execute(any(Object[].class))).thenReturn(names); ConvertingRepository repository = factory.getRepository(ConvertingRepository.class); @@ -218,6 +257,13 @@ class RepositoryFactorySupportUnitTests { assertThat(result).hasSize(1); assertThat(result.iterator().next()).isEqualTo("Dave"); + + ArgumentCaptor captor = ArgumentCaptor + .forClass(RepositoryMethodInvocationListener.Invocation.class); + verify(invocationListener).afterInvocation(captor.capture()); + + RepositoryMethodInvocationListener.Invocation value = captor.getValue(); + assertThat(value.getResult()).isEqualTo(names); } @Test // DATACMNS-509 @@ -461,6 +507,8 @@ class RepositoryFactorySupportUnitTests { Set convertListToObjectSet(); + Future findByLastname(String lastname); + @Async Future findByFirstname(String firstname); diff --git a/src/test/kotlin/org/springframework/data/repository/kotlin/CoroutineCrudRepositoryUnitTests.kt b/src/test/kotlin/org/springframework/data/repository/kotlin/CoroutineCrudRepositoryUnitTests.kt index 6f4756a13..e2772a389 100644 --- a/src/test/kotlin/org/springframework/data/repository/kotlin/CoroutineCrudRepositoryUnitTests.kt +++ b/src/test/kotlin/org/springframework/data/repository/kotlin/CoroutineCrudRepositoryUnitTests.kt @@ -25,9 +25,11 @@ import kotlinx.coroutines.runBlocking import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test +import org.mockito.ArgumentCaptor import org.mockito.Mockito import org.reactivestreams.Publisher import org.springframework.data.repository.core.support.DummyReactiveRepositoryFactory +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener import org.springframework.data.repository.reactive.ReactiveCrudRepository import org.springframework.data.repository.sample.User import reactor.core.publisher.Flux @@ -45,14 +47,17 @@ class CoroutineCrudRepositoryUnitTests { val backingRepository = mockk>() lateinit var factory: DummyReactiveRepositoryFactory lateinit var coRepository: MyCoRepository + lateinit var invocationListener: RepositoryMethodInvocationListener @BeforeEach fun before() { factory = DummyReactiveRepositoryFactory(backingRepository) + invocationListener = Mockito.mock(RepositoryMethodInvocationListener::class.java) + factory.addInvocationListener(invocationListener) coRepository = factory.getRepository(MyCoRepository::class.java) } - @Test // DATACMNS-1508 + @Test // DATACMNS-1508, DATACMNS-1764 fun shouldInvokeFindAll() { val sample = User() @@ -64,6 +69,7 @@ class CoroutineCrudRepositoryUnitTests { } assertThat(result).hasSize(1).containsOnly(sample) + Mockito.verify(invocationListener).afterInvocation(Mockito.any()) } @Test // DATACMNS-1508 @@ -176,7 +182,7 @@ class CoroutineCrudRepositoryUnitTests { verify { backingRepository.deleteAll(any>()) } } - @Test // DATACMNS-1508 + @Test // DATACMNS-1508, DATACMNS-1764 fun shouldBridgeQueryMethod() { val sample = User() @@ -188,6 +194,9 @@ class CoroutineCrudRepositoryUnitTests { } assertThat(result).isNotNull().isEqualTo(sample) + val captor = ArgumentCaptor.forClass(RepositoryMethodInvocationListener.Invocation::class.java) + Mockito.verify(invocationListener).afterInvocation(captor.capture()) + assertThat(captor.value.result).isInstanceOf(Mono::class.java) } @Test // DATACMNS-1508