From b667dfa779193e7e3157e031761bea47afe163f8 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Tue, 11 Aug 2020 13:53:30 +0200 Subject: [PATCH] DATACMNS-1764 - Capture reactive invocation on per subscription basis. Original Pull Request: #455 --- .../QueryExecutorMethodInterceptor.java | 20 +- .../core/support/RepositoryComposition.java | 114 ++++- .../support/RepositoryFactorySupport.java | 67 +-- .../support/RepositoryInvocationListener.java | 101 ----- .../RepositoryInvocationMulticaster.java | 80 ++++ .../RepositoryMethodInvocationListener.java | 49 ++- .../core/support/RepositoryMethodInvoker.java | 176 +++++++- .../util/ReactiveWrapperConverters.java | 240 +--------- .../RepositoryFactorySupportUnitTests.java | 25 +- .../RepositoryMethodInvokerUnitTests.java | 414 ++++++++++++++++++ .../ReactiveWrapperConvertersUnitTests.java | 24 - .../CoroutineCrudRepositoryUnitTests.kt | 3 +- 12 files changed, 805 insertions(+), 508 deletions(-) delete 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/RepositoryInvocationMulticaster.java create mode 100644 src/test/java/org/springframework/data/repository/core/support/RepositoryMethodInvokerUnitTests.java 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 478ef59cb..d58b9cb13 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 @@ -23,11 +23,12 @@ import java.util.Optional; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; - import org.springframework.core.ResolvableType; import org.springframework.data.projection.ProjectionFactory; import org.springframework.data.repository.core.NamedQueries; import org.springframework.data.repository.core.RepositoryInformation; +import org.springframework.data.repository.core.support.RepositoryInvocationMulticaster.DefaultRepositoryInvocationMulticaster; +import org.springframework.data.repository.core.support.RepositoryInvocationMulticaster.NoOpRepositoryInvocationMulticaster; import org.springframework.data.repository.query.QueryLookupStrategy; import org.springframework.data.repository.query.QueryMethod; import org.springframework.data.repository.query.RepositoryQuery; @@ -43,6 +44,7 @@ import org.springframework.util.ConcurrentReferenceHashMap; * * @author Oliver Gierke * @author Mark Paluch + * @author Christoph Strobl */ class QueryExecutorMethodInterceptor implements MethodInterceptor { @@ -52,7 +54,7 @@ class QueryExecutorMethodInterceptor implements MethodInterceptor { private final QueryExecutionResultHandler resultHandler; private final NamedQueries namedQueries; private final List> queryPostProcessors; - private final List methodInvocationListeners; + private final RepositoryInvocationMulticaster invocationMulticaster; /** * Creates a new {@link QueryExecutorMethodInterceptor}. Builds a model of {@link QueryMethod}s to be invoked on @@ -66,7 +68,8 @@ class QueryExecutorMethodInterceptor implements MethodInterceptor { this.repositoryInformation = repositoryInformation; this.namedQueries = namedQueries; this.queryPostProcessors = queryPostProcessors; - this.methodInvocationListeners = methodInvocationListeners; + this.invocationMulticaster = methodInvocationListeners.isEmpty() ? NoOpRepositoryInvocationMulticaster.INSTANCE + : new DefaultRepositoryInvocationMulticaster(methodInvocationListeners); this.resultHandler = new QueryExecutionResultHandler(RepositoryFactorySupport.CONVERSION_SERVICE); @@ -138,7 +141,6 @@ class QueryExecutorMethodInterceptor implements MethodInterceptor { if (hasQueryFor(method)) { - RepositoryInvocationListener invocationListener = getInvocationListener(); RepositoryMethodInvoker invocationMetadata = invocationMetadataCache.get(method); if (invocationMetadata == null) { @@ -146,19 +148,13 @@ class QueryExecutorMethodInterceptor implements MethodInterceptor { invocationMetadataCache.put(method, invocationMetadata); } - return invocationMetadata.invoke(invocationListener, invocation.getArguments()); + return invocationMetadata.invoke(repositoryInformation.getRepositoryInterface(), invocationMulticaster, + 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}; * 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 43dfd86a1..53be1cf1f 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 @@ -32,6 +32,9 @@ import java.util.stream.Stream; import org.springframework.data.repository.core.RepositoryMetadata; import org.springframework.data.repository.core.support.MethodLookup.InvokedMethod; import org.springframework.data.repository.core.support.MethodLookup.MethodPredicate; +import org.springframework.data.repository.core.support.RepositoryInvocationMulticaster.NoOpRepositoryInvocationMulticaster; +import org.springframework.data.repository.util.ReactiveWrapperConverters; +import org.springframework.data.repository.util.ReactiveWrappers; import org.springframework.data.util.Streamable; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -56,6 +59,7 @@ import org.springframework.util.ReflectionUtils; * Composition objects are immutable and thread-safe. * * @author Mark Paluch + * @author Christoph Strobl * @soundtrack Masterboy - Anybody (Fj Gauder Mix) * @since 2.0 * @see RepositoryFragment @@ -63,16 +67,50 @@ import org.springframework.util.ReflectionUtils; public class RepositoryComposition { private static final BiFunction PASSTHRU_ARG_CONVERTER = (methodParameter, o) -> o; - private static final RepositoryComposition EMPTY = new RepositoryComposition(RepositoryFragments.empty(), + private static final BiFunction REACTIVE_ARGS_CONVERTER = (method, args) -> { + + if (ReactiveWrappers.isAvailable()) { + + Class[] parameterTypes = method.getParameterTypes(); + + Object[] converted = new Object[args.length]; + for (int i = 0; i < args.length; i++) { + + Object value = args[i]; + Object convertedArg = value; + + 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); + } + } + + converted[i] = convertedArg; + } + + return converted; + } + + return args; + }; + + private static final RepositoryComposition EMPTY = new RepositoryComposition(null, RepositoryFragments.empty(), MethodLookups.direct(), PASSTHRU_ARG_CONVERTER); private final Map methodCache = new ConcurrentReferenceHashMap<>(); private final RepositoryFragments fragments; private final MethodLookup methodLookup; private final BiFunction argumentConverter; + private final @Nullable RepositoryMetadata metadata; - private RepositoryComposition(RepositoryFragments fragments, MethodLookup methodLookup, - BiFunction argumentConverter) { + private RepositoryComposition(@Nullable RepositoryMetadata metadata, RepositoryFragments fragments, + MethodLookup methodLookup, BiFunction argumentConverter) { + + this.metadata = metadata; this.fragments = fragments; this.methodLookup = methodLookup; this.argumentConverter = argumentConverter; @@ -87,6 +125,24 @@ public class RepositoryComposition { return EMPTY; } + /** + * Create an {@link RepositoryComposition} using the provided {@link RepositoryMetadata} to set {@link MethodLookup + * method lookups} depending in the repository type (reactive/imperative). + * + * @return an empty {@link RepositoryComposition}. + * @since 2.4 + */ + public static RepositoryComposition fromMetadata(RepositoryMetadata metadata) { + + if (metadata.isReactiveRepository()) { + return new RepositoryComposition(metadata, RepositoryFragments.empty(), MethodLookups.forReactiveTypes(metadata), + REACTIVE_ARGS_CONVERTER); + } + + return new RepositoryComposition(metadata, RepositoryFragments.empty(), MethodLookups.direct(), + PASSTHRU_ARG_CONVERTER); + } + /** * Create a {@link RepositoryComposition} for just a single {@code implementation} with {@link MethodLookups#direct()) * method lookup. @@ -95,7 +151,7 @@ public class RepositoryComposition { * @return the {@link RepositoryComposition} for a single {@code implementation}. */ public static RepositoryComposition just(Object implementation) { - return new RepositoryComposition(RepositoryFragments.just(implementation), MethodLookups.direct(), + return new RepositoryComposition(null, RepositoryFragments.just(implementation), MethodLookups.direct(), PASSTHRU_ARG_CONVERTER); } @@ -118,7 +174,7 @@ public class RepositoryComposition { * @return the {@link RepositoryComposition} from {@link RepositoryFragment fragments}. */ public static RepositoryComposition of(List> fragments) { - return new RepositoryComposition(RepositoryFragments.from(fragments), MethodLookups.direct(), + return new RepositoryComposition(null, RepositoryFragments.from(fragments), MethodLookups.direct(), PASSTHRU_ARG_CONVERTER); } @@ -130,7 +186,7 @@ public class RepositoryComposition { * @return the {@link RepositoryComposition} from {@link RepositoryFragments fragments}. */ public static RepositoryComposition of(RepositoryFragments fragments) { - return new RepositoryComposition(fragments, MethodLookups.direct(), PASSTHRU_ARG_CONVERTER); + return new RepositoryComposition(null, fragments, MethodLookups.direct(), PASSTHRU_ARG_CONVERTER); } /** @@ -141,7 +197,7 @@ public class RepositoryComposition { * @return the new {@link RepositoryComposition}. */ public RepositoryComposition append(RepositoryFragment fragment) { - return new RepositoryComposition(fragments.append(fragment), methodLookup, argumentConverter); + return new RepositoryComposition(metadata, fragments.append(fragment), methodLookup, argumentConverter); } /** @@ -153,7 +209,7 @@ public class RepositoryComposition { * @return the new {@link RepositoryComposition}. */ public RepositoryComposition append(RepositoryFragments fragments) { - return new RepositoryComposition(this.fragments.append(fragments), methodLookup, argumentConverter); + return new RepositoryComposition(metadata, this.fragments.append(fragments), methodLookup, argumentConverter); } /** @@ -163,7 +219,7 @@ public class RepositoryComposition { * @return the new {@link RepositoryComposition}. */ public RepositoryComposition withArgumentConverter(BiFunction argumentConverter) { - return new RepositoryComposition(fragments, methodLookup, argumentConverter); + return new RepositoryComposition(metadata, fragments, methodLookup, argumentConverter); } /** @@ -173,7 +229,18 @@ public class RepositoryComposition { * @return the new {@link RepositoryComposition}. */ public RepositoryComposition withMethodLookup(MethodLookup methodLookup) { - return new RepositoryComposition(fragments, methodLookup, argumentConverter); + return new RepositoryComposition(metadata, fragments, methodLookup, argumentConverter); + } + + /** + * Create a new {@link RepositoryComposition} retaining current configuration and set {@code metadata}. + * + * @param metadata must not be {@literal null}. + * @return new instance of {@link RepositoryComposition}. + * @since 2.4 + */ + public RepositoryComposition withMetadata(RepositoryMetadata metadata) { + return new RepositoryComposition(metadata, fragments, methodLookup, argumentConverter); } /** @@ -194,7 +261,7 @@ public class RepositoryComposition { * @throws Throwable */ public Object invoke(Method method, Object... args) throws Throwable { - return invoke(RepositoryInvocationListener.NoOpRepositoryInvocationListener.INSTANCE, method, args); + return invoke(NoOpRepositoryInvocationMulticaster.INSTANCE, method, args); } /** @@ -205,7 +272,7 @@ public class RepositoryComposition { * @return * @throws Throwable */ - Object invoke(RepositoryInvocationListener listener, Method method, Object[] args) throws Throwable { + Object invoke(RepositoryInvocationMulticaster listener, Method method, Object[] args) throws Throwable { Method methodToCall = getMethod(method); @@ -215,7 +282,8 @@ public class RepositoryComposition { ReflectionUtils.makeAccessible(methodToCall); - return fragments.invoke(listener, method, methodToCall, argumentConverter.apply(methodToCall, args)); + return fragments.invoke(metadata != null ? metadata.getRepositoryInterface() : method.getDeclaringClass(), listener, + method, methodToCall, argumentConverter.apply(methodToCall, args)); } /** @@ -306,6 +374,7 @@ public class RepositoryComposition { private final List> fragments; private RepositoryFragments(List> fragments) { + this.fragments = fragments; } @@ -421,8 +490,7 @@ public class RepositoryComposition { */ @Nullable public Object invoke(Method invokedMethod, Method methodToCall, Object[] args) throws Throwable { - return invoke(RepositoryInvocationListener.NoOpRepositoryInvocationListener.INSTANCE, invokedMethod, methodToCall, - args); + return invoke(null, NoOpRepositoryInvocationMulticaster.INSTANCE, invokedMethod, methodToCall, args); } /** @@ -438,8 +506,8 @@ public class RepositoryComposition { * @throws Throwable */ @Nullable - Object invoke(RepositoryInvocationListener listener, Method invokedMethod, Method methodToCall, Object[] args) - throws Throwable { + Object invoke(Class repositoryInterface, RepositoryInvocationMulticaster listener, Method invokedMethod, + Method methodToCall, Object[] args) throws Throwable { RepositoryFragment fragment = fragmentCache.computeIfAbsent(methodToCall, this::findImplementationFragment); Optional optional = fragment.getImplementation(); @@ -448,14 +516,16 @@ public class RepositoryComposition { throw new IllegalArgumentException(String.format("No implementation found for method %s", methodToCall)); } - RepositoryMethodInvoker invocationMetadata = invocationMetadataCache.get(invokedMethod); + RepositoryMethodInvoker repositoryMethodInvoker = invocationMetadataCache.get(invokedMethod); - if (invocationMetadata == null) { - invocationMetadata = RepositoryMethodInvoker.forFragmentMethod(invokedMethod, optional.get(), methodToCall); - invocationMetadataCache.put(invokedMethod, invocationMetadata); + if (repositoryMethodInvoker == null) { + + repositoryMethodInvoker = RepositoryMethodInvoker.forFragmentMethod(invokedMethod, optional.get(), + methodToCall); + invocationMetadataCache.put(invokedMethod, repositoryMethodInvoker); } - return invocationMetadata.invoke(listener, args); + return repositoryMethodInvoker.invoke(repositoryInterface, 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 265e4580e..86ebeb49a 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 @@ -22,14 +22,12 @@ import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.function.BiFunction; import java.util.stream.Collectors; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.interceptor.ExposeInvocationInterceptor; import org.springframework.beans.BeanUtils; @@ -49,6 +47,8 @@ import org.springframework.data.repository.core.NamedQueries; import org.springframework.data.repository.core.RepositoryInformation; import org.springframework.data.repository.core.RepositoryMetadata; import org.springframework.data.repository.core.support.RepositoryComposition.RepositoryFragments; +import org.springframework.data.repository.core.support.RepositoryInvocationMulticaster.DefaultRepositoryInvocationMulticaster; +import org.springframework.data.repository.core.support.RepositoryInvocationMulticaster.NoOpRepositoryInvocationMulticaster; import org.springframework.data.repository.query.QueryLookupStrategy; import org.springframework.data.repository.query.QueryLookupStrategy.Key; import org.springframework.data.repository.query.QueryMethod; @@ -56,8 +56,6 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro import org.springframework.data.repository.query.RepositoryQuery; 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.ReflectionUtils; import org.springframework.lang.Nullable; import org.springframework.transaction.interceptor.TransactionalProxy; @@ -78,39 +76,8 @@ import org.springframework.util.ObjectUtils; */ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware, BeanFactoryAware { - private static final BiFunction REACTIVE_ARGS_CONVERTER = (method, args) -> { - - if (ReactiveWrappers.isAvailable()) { - - Class[] parameterTypes = method.getParameterTypes(); - - Object[] converted = new Object[args.length]; - for (int i = 0; i < args.length; i++) { - - Object value = args[i]; - Object convertedArg = value; - - 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); - } - } - - converted[i] = convertedArg; - } - - return converted; - } - - return args; - }; - final static GenericConversionService CONVERSION_SERVICE = new DefaultConversionService(); - private static final Log logger = LogFactory.getLog(RepositoryFactorySupport.class); + private static final Log logger = LogFactory.getLog(RepositoryFactorySupport.class); static { QueryExecutionConverters.registerConvertersIn(CONVERSION_SERVICE); @@ -260,15 +227,7 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware, * @return */ private RepositoryComposition getRepositoryComposition(RepositoryMetadata metadata) { - - RepositoryComposition composition = RepositoryComposition.empty(); - - if (metadata.isReactiveRepository()) { - return composition.withMethodLookup(MethodLookups.forReactiveTypes(metadata)) - .withArgumentConverter(REACTIVE_ARGS_CONVERTER); - } - - return composition.withMethodLookup(MethodLookups.forRepositoryTypes(metadata)); + return RepositoryComposition.fromMetadata(metadata); } /** @@ -349,7 +308,8 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware, T repository = (T) result.getProxy(classLoader); if (logger.isDebugEnabled()) { - logger.debug(LogMessage.format("Finished creation of repository instance for {}.", repositoryInterface.getName())); + logger + .debug(LogMessage.format("Finished creation of repository instance for {}.", repositoryInterface.getName())); } return repository; @@ -545,13 +505,14 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware, private final RepositoryInformation information; private final RepositoryComposition composition; - private final List methodInvocationListeners; + private final RepositoryInvocationMulticaster invocationMulticaster; public ImplementationMethodExecutionInterceptor(RepositoryInformation information, RepositoryComposition composition, List methodInvocationListeners) { this.information = information; this.composition = composition; - this.methodInvocationListeners = methodInvocationListeners; + this.invocationMulticaster = methodInvocationListeners.isEmpty() ? NoOpRepositoryInvocationMulticaster.INSTANCE + : new DefaultRepositoryInvocationMulticaster(methodInvocationListeners); } /* @@ -564,23 +525,15 @@ public abstract class RepositoryFactorySupport implements BeanClassLoaderAware, Method method = invocation.getMethod(); Object[] arguments = invocation.getArguments(); - RepositoryInvocationListener invocationListener = getInvocationListener(); try { - return composition.invoke(invocationListener, method, arguments); + return composition.invoke(invocationMulticaster, 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 deleted file mode 100644 index c65a16192..000000000 --- a/src/main/java/org/springframework/data/repository/core/support/RepositoryInvocationListener.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * 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/RepositoryInvocationMulticaster.java b/src/main/java/org/springframework/data/repository/core/support/RepositoryInvocationMulticaster.java new file mode 100644 index 000000000..24f8745a5 --- /dev/null +++ b/src/main/java/org/springframework/data/repository/core/support/RepositoryInvocationMulticaster.java @@ -0,0 +1,80 @@ +/* + * 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.List; + +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocation; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocationResult; + +/** + * 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 RepositoryInvocationMulticaster { + + /** + * The repository method invocation to be handled after calling it. + * + * @param method + * @param args + * @param result + */ + void notifyListeners(Method method, Object[] args, RepositoryMethodInvocation result); + + /** + * {@link RepositoryInvocationMulticaster} that does nothing upon invocation. + * + * @author Mark Paluch + */ + enum NoOpRepositoryInvocationMulticaster implements RepositoryInvocationMulticaster { + + INSTANCE; + + @Override + public void notifyListeners(Method method, Object[] args, RepositoryMethodInvocation result) { + + } + } + + /** + * {@link RepositoryInvocationMulticaster} implementation that notifies {@link RepositoryMethodInvocationListener} + * upon {@link #notifyListeners(Method, Object[], RepositoryMethodInvocationResult)}. + * + * @author Mark Paluch + */ + class DefaultRepositoryInvocationMulticaster implements RepositoryInvocationMulticaster { + + private final List methodInvocationListeners; + + DefaultRepositoryInvocationMulticaster(List methodInvocationListeners) { + + this.methodInvocationListeners = methodInvocationListeners; + } + + @Override + public void notifyListeners(Method method, Object[] args, RepositoryMethodInvocation result) { + + for (RepositoryMethodInvocationListener methodInvocationListener : methodInvocationListeners) { + methodInvocationListener.afterInvocation(result); + } + } + } +} 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 index 2a989a3d0..4ed606268 100644 --- a/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvocationListener.java +++ b/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvocationListener.java @@ -16,17 +16,20 @@ package org.springframework.data.repository.core.support; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.concurrent.TimeUnit; import org.springframework.lang.Nullable; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * 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. + * notified with an {@link RepositoryMethodInvocation} object that describes which repository method was invoked along + * with invocation arguments and the call duration. * * @author Mark Paluch + * @author Christoph Strobl * @since 2.4 */ public interface RepositoryMethodInvocationListener { @@ -34,35 +37,33 @@ public interface RepositoryMethodInvocationListener { /** * Handle the invocation event. This method is called after the execution has finished. * - * @param invocation the invocation to respond to. + * @param repositoryMethodInvocation the invocation to respond to. */ - void afterInvocation(Invocation invocation); + void afterInvocation(RepositoryMethodInvocation repositoryMethodInvocation); /** * Value object capturing the actual invocation. */ - class Invocation { + class RepositoryMethodInvocation { private final long durationNs; private final Class repositoryInterface; private final Method method; - private final Object result; - private final Throwable exception; + private final RepositoryMethodInvocationResult result; /** - * @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}. + * @param result the outcome of the invocation. Must not be {@literal null}. + * @param durationNs the duration in {@link TimeUnit#NANOSECONDS}. */ - public Invocation(long durationNs, Class repositoryInterface, Method method, @Nullable Object result, - @Nullable Throwable exception) { + public RepositoryMethodInvocation(Class repositoryInterface, Method method, + RepositoryMethodInvocationResult result, long durationNs) { + this.durationNs = durationNs; this.repositoryInterface = repositoryInterface; this.method = method; this.result = result; - this.exception = exception; } public long getDuration(TimeUnit timeUnit) { @@ -81,13 +82,29 @@ public interface RepositoryMethodInvocationListener { } @Nullable - public Object getResult() { + public RepositoryMethodInvocationResult getResult() { return result; } + @Override + public String toString() { + + return String.format("Invocation %s.%s(%s): %s ms - %s", repositoryInterface.getSimpleName(), method.getName(), + StringUtils.arrayToCommaDelimitedString( + Arrays.stream(method.getParameterTypes()).map(Class::getSimpleName).toArray()), + getDuration(TimeUnit.MILLISECONDS), result.getState()); + } + } + + interface RepositoryMethodInvocationResult { + + State getState(); + @Nullable - public Throwable getException() { - return exception; + Throwable getError(); + + public enum State { + SUCCESS, ERROR, CANCELED, RUNNING } } } 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 index 00b266728..199cd1938 100644 --- a/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvoker.java +++ b/src/main/java/org/springframework/data/repository/core/support/RepositoryMethodInvoker.java @@ -18,13 +18,18 @@ package org.springframework.data.repository.core.support; import kotlin.coroutines.Continuation; import kotlin.reflect.KFunction; import kotlinx.coroutines.reactive.AwaitKt; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.stream.Stream; import org.reactivestreams.Publisher; - import org.springframework.core.KotlinDetector; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocation; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocationResult; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocationResult.State; import org.springframework.data.repository.query.RepositoryQuery; import org.springframework.data.repository.util.ReactiveWrapperConverters; import org.springframework.data.repository.util.ReactiveWrappers; @@ -37,6 +42,7 @@ import org.springframework.lang.Nullable; * implementation. * * @author Mark Paluch + * @author Christoph Strobl * @since 2.4 * @see #forFragmentMethod(Method, Object, Method) * @see #forRepositoryQuery(Method, RepositoryQuery) @@ -90,7 +96,7 @@ abstract class RepositoryMethodInvoker { * 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}. + * @param baseClassMethod the base method to call on fragment {@code instance}. * @return */ public static boolean canInvoke(Method declaredMethod, Method baseClassMethod) { @@ -101,14 +107,16 @@ abstract class RepositoryMethodInvoker { /** * Invoke the repository method and return its value. * - * @param listener listener to notify about the call outcome. + * @param multicaster 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); + public Object invoke(Class repositoryInterface, RepositoryInvocationMulticaster multicaster, Object[] args) + throws Exception { + return shouldAdaptReactiveToSuspended() ? doInvokeReactiveToSuspended(repositoryInterface, multicaster, args) + : doInvoke(repositoryInterface, multicaster, args); } protected boolean shouldAdaptReactiveToSuspended() { @@ -116,33 +124,38 @@ abstract class RepositoryMethodInvoker { } @Nullable - private Object doInvoke(RepositoryInvocationListener listener, Object[] args) throws Exception { + private Object doInvoke(Class repositoryInterface, RepositoryInvocationMulticaster multicaster, Object[] args) + throws Exception { + + RepositoryMethodInvocationCaptor invocationResultCaptor = RepositoryMethodInvocationCaptor + .captureInvocationOn(repositoryInterface); 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)); + return inferReactiveInvocationCallbacks(repositoryInterface, multicaster, args, result); } if (result instanceof Stream) { - return ((Stream) result).onClose(() -> listener.afterInvocation(method, args, result, null)); + return ((Stream) result).onClose( + () -> multicaster.notifyListeners(method, args, computeInvocationResult(invocationResultCaptor.success()))); } - listener.afterInvocation(method, args, result, null); + multicaster.notifyListeners(method, args, computeInvocationResult(invocationResultCaptor.success())); return result; } catch (Exception e) { - listener.afterInvocation(method, args, null, e); + multicaster.notifyListeners(method, args, computeInvocationResult(invocationResultCaptor.error(e))); throw e; } } @Nullable @SuppressWarnings({ "unchecked", "ConstantConditions" }) - private Object doInvokeReactiveToSuspended(RepositoryInvocationListener listener, Object[] args) throws Exception { + private Object doInvokeReactiveToSuspended(Class repositoryInterface, RepositoryInvocationMulticaster multicaster, + Object[] args) throws Exception { /* * Kotlin suspended functions are invoked with a synthetic Continuation parameter that keeps track of the Coroutine context. @@ -151,28 +164,67 @@ abstract class RepositoryMethodInvoker { */ Continuation continuation = (Continuation) args[args.length - 1]; args[args.length - 1] = null; + + RepositoryMethodInvocationCaptor invocationResultCaptor = RepositoryMethodInvocationCaptor + .captureInvocationOn(repositoryInterface); try { - Object result = invokable.invoke(args); + + Publisher result = inferReactiveInvocationCallbacks(repositoryInterface, multicaster, args, + 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); + return AwaitKt.awaitFirstOrNull(result, continuation); } catch (Exception e) { - listener.afterInvocation(method, args, null, e); + multicaster.notifyListeners(method, args, computeInvocationResult(invocationResultCaptor.error(e))); throw e; } } + private RepositoryMethodInvocation computeInvocationResult(RepositoryMethodInvocationCaptor captured) { + return new RepositoryMethodInvocation(captured.getRepositoryInterface(), method, captured.getCapturedResult(), + captured.getDuration()); + } + + private Publisher inferReactiveInvocationCallbacks(Class repositoryInterface, + RepositoryInvocationMulticaster multicaster, Object[] args, Object result) { + + if (result instanceof Mono) { + return Mono.usingWhen( + Mono.fromSupplier(() -> RepositoryMethodInvocationCaptor.captureInvocationOn(repositoryInterface)), it -> { + it.trackStart(); + return ReactiveWrapperConverters.toWrapper(result, Mono.class); + }, it -> { + multicaster.notifyListeners(method, args, computeInvocationResult(it.success())); + return Mono.empty(); + }, (it, e) -> { + multicaster.notifyListeners(method, args, computeInvocationResult(it.error(e))); + return Mono.empty(); + }, it -> { + multicaster.notifyListeners(method, args, computeInvocationResult(it.canceled())); + return Mono.empty(); + }); + } + + return Flux.usingWhen( + Mono.fromSupplier(() -> RepositoryMethodInvocationCaptor.captureInvocationOn(repositoryInterface)), it -> { + it.trackStart(); + return result instanceof Publisher ? (Publisher) result + : ReactiveWrapperConverters.toWrapper(result, Publisher.class); + }, it -> { + multicaster.notifyListeners(method, args, computeInvocationResult(it.success())); + return Mono.empty(); + }, (it, e) -> { + multicaster.notifyListeners(method, args, computeInvocationResult(it.error(e))); + return Mono.empty(); + }, it -> { + multicaster.notifyListeners(method, args, computeInvocationResult(it.canceled())); + return Mono.empty(); + }); + } + interface Invokable { @Nullable @@ -301,4 +353,80 @@ abstract class RepositoryMethodInvoker { } } } + + private static class RepositoryMethodInvocationCaptor { + + private final Class repositoryInterface; + private long startTime; + private @Nullable Long endTime; + private final State state; + private final @Nullable Throwable error; + + protected RepositoryMethodInvocationCaptor(Class repositoryInterface, long startTime, Long endTime, State state, + @Nullable Throwable exception) { + + this.repositoryInterface = repositoryInterface; + this.startTime = startTime; + this.endTime = endTime; + this.state = state; + this.error = exception instanceof InvocationTargetException ? exception.getCause() : exception; + } + + public static RepositoryMethodInvocationCaptor captureInvocationOn(Class repositoryInterface) { + return new RepositoryMethodInvocationCaptor(repositoryInterface, System.nanoTime(), null, State.RUNNING, null); + } + + public RepositoryMethodInvocationCaptor error(Throwable exception) { + return new RepositoryMethodInvocationCaptor(repositoryInterface, startTime, System.nanoTime(), State.ERROR, + exception); + } + + public RepositoryMethodInvocationCaptor success() { + return new RepositoryMethodInvocationCaptor(repositoryInterface, startTime, System.nanoTime(), State.SUCCESS, + null); + } + + public RepositoryMethodInvocationCaptor canceled() { + return new RepositoryMethodInvocationCaptor(repositoryInterface, startTime, System.nanoTime(), State.CANCELED, + null); + } + + Class getRepositoryInterface() { + return repositoryInterface; + } + + void trackStart() { + startTime = System.nanoTime(); + } + + public State getState() { + return state; + } + + @Nullable + public Throwable getError() { + return error; + } + + long getDuration() { + return (endTime != null ? endTime : System.nanoTime()) - startTime; + } + + RepositoryMethodInvocationResult getCapturedResult() { + return new RepositoryMethodInvocationResult() { + + @Override + public State getState() { + return RepositoryMethodInvocationCaptor.this.getState(); + } + + @Nullable + @Override + public Throwable getError() { + return RepositoryMethodInvocationCaptor.this.getError(); + } + }; + } + + } } diff --git a/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java index d3c68c0f4..c88b8e4b5 100644 --- a/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java +++ b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java @@ -27,13 +27,12 @@ import rx.Single; import java.util.ArrayList; import java.util.List; -import java.util.function.Consumer; +import java.util.Optional; 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; @@ -59,6 +58,7 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Oliver Gierke + * @author Christoph Strobl * @since 2.0 * @see ReactiveWrappers * @see ReactiveAdapterRegistry @@ -183,55 +183,15 @@ public abstract class ReactiveWrapperConverters { Assert.notNull(reactiveObject, "Reactive source object must not be null!"); Assert.notNull(converter, "Converter must not be null!"); - return REACTIVE_WRAPPERS.stream()// - .filter(it -> ClassUtils.isAssignable(it.getWrapperClass(), reactiveObject.getClass()))// - .findFirst()// + return getFirst(reactiveObject)// .map(it -> (T) it.map(reactiveObject, converter))// .orElseThrow(() -> new IllegalStateException(String.format("Cannot apply converter to %s", reactiveObject))); } - /** - * Apply a {@link Runnable} when the reactive type emits a completion signal. - * - * @param reactiveObject must not be {@literal null}. - * @param onSuccess must not be {@literal null}. - * @return - * @since 2.4 - */ - @SuppressWarnings("unchecked") - public static T doOnSuccess(Object reactiveObject, Runnable onSuccess) { - - Assert.notNull(reactiveObject, "Reactive source object must not be null!"); - Assert.notNull(onSuccess, "onSuccess callback must not be null!"); - + private static Optional> getFirst(Object reactiveObject) { return REACTIVE_WRAPPERS.stream()// .filter(it -> ClassUtils.isAssignable(it.getWrapperClass(), reactiveObject.getClass()))// - .findFirst()// - .map(it -> (T) it.doOnSuccess(reactiveObject, onSuccess))// - .orElseThrow( - () -> new IllegalStateException(String.format("Cannot apply onSuccess callback to %s", reactiveObject))); - } - - /** - * Apply a {@link Consumer} when the reactive type emits an error signal. - * - * @param reactiveObject must not be {@literal null}. - * @param onError must not be {@literal null}. - * @return - * @since 2.4 - */ - @SuppressWarnings("unchecked") - public static T doOnError(Object reactiveObject, Consumer onError) { - - Assert.notNull(reactiveObject, "Reactive source object must not be null!"); - Assert.notNull(onError, "onError callback must not be null!"); - - return REACTIVE_WRAPPERS.stream()// - .filter(it -> ClassUtils.isAssignable(it.getWrapperClass(), reactiveObject.getClass()))// - .findFirst()// - .map(it -> (T) it.doOnError(reactiveObject, onError))// - .orElseThrow( - () -> new IllegalStateException(String.format("Cannot apply onError callback to %s", reactiveObject))); + .findFirst(); } /** @@ -275,25 +235,6 @@ public abstract class ReactiveWrapperConverters { */ Object map(Object wrapper, Function function); - /** - * Apply a {@link Runnable} when the reactive type emits a completion signal. - * - * @param wrapper the reactive type, must not be {@literal null}. - * @param onSuccess the signal callback, must not be {@literal null}. - * @return the reactive type with {@code onSuccess} attached. - * @since 2.4 - */ - Object doOnSuccess(Object wrapper, Runnable onSuccess); - - /** - * Apply a {@link Consumer} when the reactive type emits an error signal. - * - * @param wrapper the reactive type, must not be {@literal null}. - * @param onError the error consumer, must not be {@literal null}. - * @return the reactive type with {@code onError} attached. - * @since 2.4 - */ - Object doOnError(Object wrapper, Consumer onError); } /** @@ -312,16 +253,6 @@ public abstract class ReactiveWrapperConverters { public Mono map(Object wrapper, Function function) { return ((Mono) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((Mono) wrapper).doOnSuccess(o -> onSuccess.run()); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((Mono) wrapper).doOnError(onError); - } } /** @@ -339,16 +270,6 @@ public abstract class ReactiveWrapperConverters { public Flux map(Object wrapper, Function function) { return ((Flux) wrapper).map(function); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((Flux) wrapper).doOnComplete(onSuccess); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((Flux) wrapper).doOnError(onError); - } } /** @@ -368,26 +289,6 @@ public abstract class ReactiveWrapperConverters { public Flow map(Object wrapper, Function function) { return FlowKt.map((Flow) wrapper, (o, continuation) -> function.apply(o)); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return FlowKt.onCompletion((Flow) wrapper, (collector, ex, continuation) -> { - if (ex == null) { - onSuccess.run(); - } - return collector; - }); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return FlowKt.onCompletion((Flow) wrapper, (collector, ex, continuation) -> { - if (ex != null) { - onError.accept(ex); - } - return collector; - }); - } } /** @@ -415,34 +316,6 @@ public abstract class ReactiveWrapperConverters { return FluxWrapper.INSTANCE.map(Flux.from((Publisher) wrapper), function); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - - if (wrapper instanceof Mono) { - return MonoWrapper.INSTANCE.doOnSuccess(wrapper, onSuccess); - } - - if (wrapper instanceof Flux) { - return FluxWrapper.INSTANCE.doOnSuccess(wrapper, onSuccess); - } - - return FluxWrapper.INSTANCE.doOnSuccess(Flux.from((Publisher) wrapper), onSuccess); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - - if (wrapper instanceof Mono) { - return MonoWrapper.INSTANCE.doOnError(wrapper, onError); - } - - if (wrapper instanceof Flux) { - return FluxWrapper.INSTANCE.doOnError(wrapper, onError); - } - - return FluxWrapper.INSTANCE.doOnError(Flux.from((Publisher) wrapper), onError); - } } // ------------------------------------------------------------------------- @@ -465,16 +338,6 @@ public abstract class ReactiveWrapperConverters { public Single map(Object wrapper, Function function) { return ((Single) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((Single) wrapper).doOnSuccess(o -> onSuccess.run()); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((Single) wrapper).doOnError(onError::accept); - } } /** @@ -493,16 +356,6 @@ public abstract class ReactiveWrapperConverters { public Observable map(Object wrapper, Function function) { return ((Observable) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((Observable) wrapper).doOnCompleted(onSuccess::run); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((Observable) wrapper).doOnError(onError::accept); - } } // ------------------------------------------------------------------------- @@ -525,16 +378,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Single map(Object wrapper, Function function) { return ((io.reactivex.Single) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.Single) wrapper).doOnSuccess(o -> onSuccess.run()); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.Single) wrapper).doOnError(onError::accept); - } } /** @@ -553,16 +396,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Maybe map(Object wrapper, Function function) { return ((io.reactivex.Maybe) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.Maybe) wrapper).doOnSuccess(o -> onSuccess.run()); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.Maybe) wrapper).doOnError(onError::accept); - } } /** @@ -581,16 +414,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Observable map(Object wrapper, Function function) { return ((io.reactivex.Observable) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.Observable) wrapper).doOnComplete(onSuccess::run); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.Observable) wrapper).doOnError(onError::accept); - } } /** @@ -609,16 +432,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Flowable map(Object wrapper, Function function) { return ((io.reactivex.Flowable) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.Flowable) wrapper).doOnComplete(onSuccess::run); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.Flowable) wrapper).doOnError(onError::accept); - } } // ------------------------------------------------------------------------- @@ -641,16 +454,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Single map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Single) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.rxjava3.core.Single) wrapper).doOnSuccess(o -> onSuccess.run()); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.rxjava3.core.Single) wrapper).doOnError(onError::accept); - } } /** @@ -669,16 +472,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Maybe map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Maybe) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.rxjava3.core.Maybe) wrapper).doOnSuccess(o -> onSuccess.run()); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.rxjava3.core.Maybe) wrapper).doOnError(onError::accept); - } } /** @@ -697,16 +490,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Observable map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Observable) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.rxjava3.core.Observable) wrapper).doOnComplete(onSuccess::run); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.rxjava3.core.Observable) wrapper).doOnError(onError::accept); - } } /** @@ -725,16 +508,6 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Flowable map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Flowable) wrapper).map(function::apply); } - - @Override - public Object doOnSuccess(Object wrapper, Runnable onSuccess) { - return ((io.reactivex.rxjava3.core.Flowable) wrapper).doOnComplete(onSuccess::run); - } - - @Override - public Object doOnError(Object wrapper, Consumer onError) { - return ((io.reactivex.rxjava3.core.Flowable) wrapper).doOnError(onError::accept); - } } // ------------------------------------------------------------------------- @@ -799,8 +572,7 @@ public abstract class ReactiveWrapperConverters { /** * A {@link ConverterFactory} that adapts between reactive types using {@link ReactiveAdapterRegistry}. */ - private enum ReactiveAdapterConverterFactory - implements ConverterFactory, ConditionalConverter { + private enum ReactiveAdapterConverterFactory implements ConverterFactory, ConditionalConverter { INSTANCE; 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 ce4e9820b..6f1803749 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 @@ -40,7 +40,6 @@ import org.mockito.Mockito; import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; - import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.dao.EmptyResultDataAccessException; @@ -58,6 +57,8 @@ import org.springframework.data.repository.core.NamedQueries; import org.springframework.data.repository.core.RepositoryInformation; import org.springframework.data.repository.core.RepositoryMetadata; import org.springframework.data.repository.core.support.RepositoryComposition.RepositoryFragments; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocation; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocationResult.State; import org.springframework.data.repository.query.RepositoryQuery; import org.springframework.data.repository.sample.User; import org.springframework.lang.Nullable; @@ -213,12 +214,8 @@ class RepositoryFactorySupportUnitTests { verify(factory.queryOne, times(1)).execute(any(Object[].class)); - ArgumentCaptor captor = ArgumentCaptor - .forClass(RepositoryMethodInvocationListener.Invocation.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(RepositoryMethodInvocation.class); verify(invocationListener).afterInvocation(captor.capture()); - - RepositoryMethodInvocationListener.Invocation value = captor.getValue(); - assertThat(value.getResult()).isEqualTo(reference); } @Test // DATACMNS-1764, DATACMNS-1764 @@ -235,13 +232,13 @@ class RepositoryFactorySupportUnitTests { fail("Missing exception"); } catch (IllegalStateException e) {} - ArgumentCaptor captor = ArgumentCaptor - .forClass(RepositoryMethodInvocationListener.Invocation.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(RepositoryMethodInvocation.class); verify(invocationListener).afterInvocation(captor.capture()); - RepositoryMethodInvocationListener.Invocation invocation = captor.getValue(); - assertThat(invocation.getDuration(TimeUnit.NANOSECONDS)).isGreaterThan(0); - assertThat(invocation.getException()).isInstanceOf(IllegalStateException.class); + RepositoryMethodInvocation repositoryMethodInvocation = captor.getValue(); + assertThat(repositoryMethodInvocation.getDuration(TimeUnit.NANOSECONDS)).isGreaterThan(0); + assertThat(repositoryMethodInvocation.getResult().getState()).isEqualTo(State.ERROR); + assertThat(repositoryMethodInvocation.getResult().getError()).isInstanceOf(IllegalStateException.class); } @Test // DATACMNS-509, DATACMNS-1764 @@ -258,12 +255,8 @@ class RepositoryFactorySupportUnitTests { assertThat(result).hasSize(1); assertThat(result.iterator().next()).isEqualTo("Dave"); - ArgumentCaptor captor = ArgumentCaptor - .forClass(RepositoryMethodInvocationListener.Invocation.class); + ArgumentCaptor captor = ArgumentCaptor.forClass(RepositoryMethodInvocation.class); verify(invocationListener).afterInvocation(captor.capture()); - - RepositoryMethodInvocationListener.Invocation value = captor.getValue(); - assertThat(value.getResult()).isEqualTo(names); } @Test // DATACMNS-509 diff --git a/src/test/java/org/springframework/data/repository/core/support/RepositoryMethodInvokerUnitTests.java b/src/test/java/org/springframework/data/repository/core/support/RepositoryMethodInvokerUnitTests.java new file mode 100644 index 000000000..dd5b13f34 --- /dev/null +++ b/src/test/java/org/springframework/data/repository/core/support/RepositoryMethodInvokerUnitTests.java @@ -0,0 +1,414 @@ +/* + * 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 static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; + +import kotlin.coroutines.Continuation; +import kotlin.coroutines.CoroutineContext; +import kotlinx.coroutines.flow.Flow; +import kotlinx.coroutines.flow.FlowKt; +import kotlinx.coroutines.reactor.ReactorContext; +import lombok.AllArgsConstructor; +import lombok.NoArgsConstructor; +import lombok.ToString; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Queue; +import java.util.Spliterator; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.stream.Stream; + +import org.assertj.core.api.Assertions; +import org.assertj.core.data.Percentage; +import org.jetbrains.annotations.NotNull; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.internal.stubbing.answers.AnswersWithDelay; +import org.mockito.internal.stubbing.answers.Returns; +import org.mockito.junit.jupiter.MockitoExtension; +import org.reactivestreams.Subscription; +import org.springframework.data.repository.CrudRepository; +import org.springframework.data.repository.core.support.CoroutineRepositoryMetadataUnitTests.MyCoroutineRepository; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocation; +import org.springframework.data.repository.core.support.RepositoryMethodInvocationListener.RepositoryMethodInvocationResult.State; +import org.springframework.data.repository.query.RepositoryQuery; +import org.springframework.data.repository.reactive.ReactiveCrudRepository; +import org.springframework.lang.Nullable; +import org.springframework.util.CollectionUtils; +import org.springframework.util.ReflectionUtils; + +/** + * @author Christoph Strobl + */ +@ExtendWith(MockitoExtension.class) +class RepositoryMethodInvokerUnitTests { + + @Mock RepositoryQuery query; + CapturingRepositoryInvocationMulticaster multicaster; + + @BeforeEach + void beforeEach() { + multicaster = new CapturingRepositoryInvocationMulticaster(); + } + + @Test // DATACMNS-1764 + void usesRepositoryInterfaceNameForMethodsDefinedOnCrudRepository() throws Exception { + + when(query.execute(any())).thenReturn(new TestDummy()); + + repositoryMethodInvoker("findAll").invoke(); + + assertThat(multicaster.first().getMethod().getName()).isEqualTo("findAll"); + assertThat(multicaster.first().getRepositoryInterface()).isEqualTo(DummyRepository.class); + } + + @Test // DATACMNS-1764 + void usesRepositoryInterfaceNameForMethodsOnTheInterface() throws Exception { + + when(query.execute(any())).thenReturn(new TestDummy()); + + repositoryMethodInvoker("findByName").invoke(); + + assertThat(multicaster.first().getMethod().getName()).isEqualTo("findByName"); + assertThat(multicaster.first().getRepositoryInterface()).isEqualTo(DummyRepository.class); + } + + @Test // DATACMNS-1764 + void usesRepositoryInterfaceNameForMethodsDefinedOnReactiveCrudRepository() throws Exception { + + when(query.execute(any())).thenReturn(Flux.just(new TestDummy())); + + repositoryMethodInvokerForReactive("findAll").> invoke().subscribe(); + + assertThat(multicaster.first().getMethod().getName()).isEqualTo("findAll"); + assertThat(multicaster.first().getRepositoryInterface()).isEqualTo(ReactiveDummyRepository.class); + } + + @Test // DATACMNS-1764 + void capturesImperativeDurationCorrectly() throws Exception { + + when(query.execute(any())).thenAnswer(new AnswersWithDelay(250, new Returns(new TestDummy()))); + + repositoryMethodInvoker("findAll").invoke(); + + assertThat(multicaster.first().getDuration(TimeUnit.MILLISECONDS)).isCloseTo(250, Percentage.withPercentage(10)); + } + + @Test // DATACMNS-1764 + void capturesReactiveDurationCorrectly() throws Exception { + + when(query.execute(any())).thenReturn(Flux.just(new TestDummy()).doOnSubscribe(delays(250)::delay)); + + repositoryMethodInvokerForReactive("findAll").> invoke().subscribe(); + + assertThat(multicaster.first().getDuration(TimeUnit.MILLISECONDS)).isCloseTo(250, Percentage.withPercentage(10)); + } + + @Test // DATACMNS-1764 + void capturesReactiveExecutionOnSubscribe() throws Exception { + + when(query.execute(any())).thenReturn(Flux.just(new TestDummy())); + + Flux findAll = repositoryMethodInvokerForReactive("findAll").invoke(); + + assertThat(multicaster).isEmpty(); + + findAll.subscribe(); + + assertThat(multicaster).hasSize(1); + } + + @Test // DATACMNS-1764 + void capturesReactiveDurationPerSubscriptionCorrectly() throws Exception { + + when(query.execute(any())).thenReturn(Flux.just(new TestDummy()).doOnSubscribe(delays(250, 100)::delay)); + + Flux findAll = repositoryMethodInvokerForReactive("findAll").invoke(); + + findAll.subscribe(); + findAll.subscribe(); + + assertThat(multicaster.first().getDuration(TimeUnit.MILLISECONDS)).isCloseTo(250, Percentage.withPercentage(10)); + assertThat(multicaster.last().getDuration(TimeUnit.MILLISECONDS)).isCloseTo(100, Percentage.withPercentage(10)); + } + + @Test // DATACMNS-1764 + void capturesStreamDurationOnClose() throws Exception { + + when(query.execute(any())).thenReturn(Stream.generate(TestDummy::new)); + + Stream stream = repositoryMethodInvoker("streamAll").invoke(); + assertThat(multicaster).isEmpty(); + + stream.iterator().next(); + assertThat(multicaster).isEmpty(); + + stream.close(); + assertThat(multicaster).hasSize(1); + } + + @Test // DATACMNS-1764 + void capturesStreamDurationAsSumOfDelayTillCancel() throws Exception { + + Delays delays = delays(250, 100); + when(query.execute(any())).thenReturn(Stream.generate(() -> { + + delays.delay(); + return new TestDummy(); + })); + + Stream stream = repositoryMethodInvoker("streamAll").invoke(); + stream.limit(2).forEach(it -> {}); + stream.close(); + + assertThat(multicaster.first().getDuration(TimeUnit.MILLISECONDS)).isCloseTo(350, Percentage.withPercentage(10)); + } + + @Test // DATACMNS-1764 + void capturesImperativeSuccessCorrectly() throws Exception { + + repositoryMethodInvoker("findAll").invoke(); + + assertThat(multicaster.first().getResult().getState()).isEqualTo(State.SUCCESS); + assertThat(multicaster.first().getResult().getError()).isNull(); + } + + @Test // DATACMNS-1764 + void capturesReactiveCompletionCorrectly() throws Exception { + + when(query.execute(any())).thenReturn(Mono.just(new TestDummy())); + + repositoryMethodInvokerForReactive("findByName").> invoke().subscribe(); + + assertThat(multicaster.first().getResult().getState()).isEqualTo(State.SUCCESS); + assertThat(multicaster.first().getResult().getError()).isNull(); + } + + @Test // DATACMNS-1764 + void capturesImperativeErrorCorrectly() { + + when(query.execute(any())).thenThrow(new IllegalStateException("I'll be back!")); + assertThatIllegalStateException().isThrownBy(() -> repositoryMethodInvoker("findAll").invoke()); + + assertThat(multicaster.first().getResult().getState()).isEqualTo(State.ERROR); + assertThat(multicaster.first().getResult().getError()).isInstanceOf(IllegalStateException.class); + } + + @Test // DATACMNS-1764 + void capturesReactiveErrorCorrectly() throws Exception { + + when(query.execute(any())).thenReturn(Mono.fromSupplier(() -> { + throw new IllegalStateException("I'll be back!"); + })); + + repositoryMethodInvokerForReactive("findByName").> invoke().as(StepVerifier::create).verifyError(); + + assertThat(multicaster.first().getResult().getState()).isEqualTo(State.ERROR); + assertThat(multicaster.first().getResult().getError()).isInstanceOf(IllegalStateException.class); + } + + @Test // DATACMNS-1764 + void capturesReactiveCancellationCorrectly() throws Exception { + + when(query.execute(any())).thenReturn(Flux.just(new TestDummy(), new TestDummy())); + + repositoryMethodInvokerForReactive("findAll").> invoke().take(1).subscribe(); + + assertThat(multicaster.first().getResult().getState()).isEqualTo(State.CANCELED); + assertThat(multicaster.first().getResult().getError()).isNull(); + } + + @Test // DATACMNS-1764 + void capturesKotlinSuspendFunctionsCorrectly() throws Exception { + + Flux result = Flux.just(new TestDummy()); + when(query.execute(any())).thenReturn(result); + + Flow flow = new RepositoryMethodInvokerStub(MyCoroutineRepository.class, multicaster, + "suspendedQueryMethod", query::execute).invoke(mock(Continuation.class)); + + assertThat(multicaster).isEmpty(); + + FlowKt.toCollection(flow, new ArrayList<>(), new Continuation>() { + + ReactorContext ctx = new ReactorContext(reactor.util.context.Context.empty()); + + @NotNull + @Override + public CoroutineContext getContext() { + return ctx; + } + + @Override + public void resumeWith(@NotNull Object o) { + + } + }); + + assertThat(multicaster.first().getResult().getState()).isEqualTo(State.SUCCESS); + assertThat(multicaster.first().getResult().getError()).isNull(); + } + + RepositoryMethodInvokerStub repositoryMethodInvoker(String methodName) { + return new RepositoryMethodInvokerStub(DummyRepository.class, multicaster, methodName, query::execute); + } + + RepositoryMethodInvokerStub repositoryMethodInvokerForReactive(String methodName) { + return new RepositoryMethodInvokerStub(ReactiveDummyRepository.class, multicaster, methodName, query::execute); + } + + static Delays delays(Integer... delays) { + return new Delays(delays); + } + + static class Delays { + + private final Queue delays; + + Delays(Integer... delays) { + + this.delays = new ArrayBlockingQueue(delays.length); + for (Integer delay : delays) { + this.delays.add(delay); + } + } + + void delay() { + delay(null); + } + + void delay(Subscription it) { + + if (delays.isEmpty()) { + return; + } + + delayBy(delays.poll()); + } + + private static void delayBy(Integer ms) { + + if (ms == null) { + return; + } + + try { + TimeUnit.MILLISECONDS.sleep(ms); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } + + static class RepositoryMethodInvokerStub extends RepositoryMethodInvoker { + + private final Class repositoryInterface; + private final RepositoryInvocationMulticaster multicaster; + + RepositoryMethodInvokerStub(Class repositoryInterface, RepositoryInvocationMulticaster multicaster, + String methodName, Invokable invokable) { + + super(methodByName(repositoryInterface, methodName), invokable); + this.repositoryInterface = repositoryInterface; + this.multicaster = multicaster; + } + + public T invoke(Object... args) throws Exception { + return (T) super.invoke(repositoryInterface, multicaster, args); + } + + @Nullable + public T invoke(RepositoryInvocationMulticaster multicaster, Object... args) throws Exception { + return (T) super.invoke(repositoryInterface, multicaster, args); + } + + static Method methodByName(Class repository, String name) { + return ReflectionUtils.findMethod(repository, name, null); + } + + } + + static class CapturingRepositoryInvocationMulticaster + implements RepositoryInvocationMulticaster, Iterable { + + private final List invocations = new ArrayList<>(); + + @Override + public void notifyListeners(Method method, Object[] args, RepositoryMethodInvocation result) { + invocations.add(result); + } + + RepositoryMethodInvocation first() { + + Assertions.assertThat(invocations).isNotEmpty(); + return CollectionUtils.firstElement(invocations); + } + + RepositoryMethodInvocation last() { + + Assertions.assertThat(invocations).isNotEmpty(); + return CollectionUtils.lastElement(invocations); + } + + @NotNull + @Override + public Iterator iterator() { + return invocations.iterator(); + } + + @Override + public Spliterator spliterator() { + return invocations.spliterator(); + } + + @Override + public void forEach(Consumer action) { + invocations.forEach(action); + } + } + + interface DummyRepository extends CrudRepository { + + TestDummy findByName(String name); + + Stream streamAll(); + } + + interface ReactiveDummyRepository extends ReactiveCrudRepository { + + Mono findByName(String name); + } + + @ToString + @AllArgsConstructor + @NoArgsConstructor + static class TestDummy { + String id; + String name; + } +} diff --git a/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java b/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java index b5c2518b7..fdb40daf8 100644 --- a/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java +++ b/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java @@ -229,30 +229,6 @@ class ReactiveWrapperConvertersUnitTests { assertThat(map.block()).isEqualTo(1L); } - @Test // DATACMNS-1763 - void shouldDoOnSuccess() { - - Mono foo = Mono.just("foo"); - AtomicBoolean success = new AtomicBoolean(); - Mono doOn = ReactiveWrapperConverters.doOnSuccess(foo, () -> success.set(true)); - - doOn.as(StepVerifier::create).expectNext("foo").verifyComplete(); - - assertThat(success.get()).isTrue(); - } - - @Test // DATACMNS-1763 - void shouldDoOnError() { - - Mono foo = Mono.error(new IllegalStateException()); - AtomicReference error = new AtomicReference<>(); - Mono doOn = ReactiveWrapperConverters.doOnError(foo, error::set); - - doOn.as(StepVerifier::create).verifyError(); - - assertThat(error.get()).isInstanceOf(IllegalStateException.class); - } - @Test // DATACMNS-836 void shouldMapFlux() { 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 e2772a389..c8a340259 100644 --- a/src/test/kotlin/org/springframework/data/repository/kotlin/CoroutineCrudRepositoryUnitTests.kt +++ b/src/test/kotlin/org/springframework/data/repository/kotlin/CoroutineCrudRepositoryUnitTests.kt @@ -194,9 +194,8 @@ class CoroutineCrudRepositoryUnitTests { } assertThat(result).isNotNull().isEqualTo(sample) - val captor = ArgumentCaptor.forClass(RepositoryMethodInvocationListener.Invocation::class.java) + val captor = ArgumentCaptor.forClass(RepositoryMethodInvocationListener.RepositoryMethodInvocation::class.java) Mockito.verify(invocationListener).afterInvocation(captor.capture()) - assertThat(captor.value.result).isInstanceOf(Mono::class.java) } @Test // DATACMNS-1508