Add Coroutines support to Spring AOP

This commit adds support for Kotlin Coroutines to Spring AOP
by leveraging CoroutinesUtils#invokeSuspendingFunction in
AopUtils#invokeJoinpointUsingReflection to convert it to the
equivalent Publisher return value, like in other parts of Spring
Framework.

That allows method interceptors with Reactive support to process
related return values.

CglibAopProxy#processReturnType and JdkDynamicAopProxy#invoke
take care of the conversion from the Publisher return value
to Kotlin Coroutines.

Reactive transactional and HTTP service interface support
have been refined to leverage those new generic capabilities.

Closes gh-22462

spring-aop/src/main/java/org/springframework/aop/framework/CglibAopProxy.java

@@ -47,6 +47,7 @@ import org.springframework.cglib.proxy.MethodInterceptor;
 import org.springframework.cglib.proxy.MethodProxy;
 import org.springframework.cglib.proxy.NoOp;
 import org.springframework.core.KotlinDetector;
+import org.springframework.core.MethodParameter;
 import org.springframework.core.SmartClassLoader;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -75,6 +76,7 @@ import org.springframework.util.ReflectionUtils;
  * @author Ramnivas Laddad
  * @author Chris Beams
  * @author Dave Syer
+ * @author Sebastien Deleuze
  * @see org.springframework.cglib.proxy.Enhancer
  * @see AdvisedSupport#setProxyTargetClass
  * @see DefaultAopProxyFactory
@@ -98,6 +100,8 @@ class CglibAopProxy implements AopProxy, Serializable {
 	/** Keeps track of the Classes that we have validated for final methods. */
 	private static final Map<Class<?>, Boolean> validatedClasses = new WeakHashMap<>();
 
+	private static final String COROUTINES_FLOW_CLASS_NAME = "kotlinx.coroutines.flow.Flow";
+
 
 	/** The configuration used to configure this proxy. */
 	protected final AdvisedSupport advised;
@@ -399,10 +403,11 @@ class CglibAopProxy implements AopProxy, Serializable {
 	/**
 	 * Process a return value. Wraps a return of {@code this} if necessary to be the
 	 * {@code proxy} and also verifies that {@code null} is not returned as a primitive.
+	 * Also takes care of the conversion from {@code Mono} to Kotlin Coroutines if needed.
 	 */
 	@Nullable
 	private static Object processReturnType(
-			Object proxy, @Nullable Object target, Method method, @Nullable Object returnValue) {
+			Object proxy, @Nullable Object target, Method method, Object[] arguments, @Nullable Object returnValue) {
 
 		// Massage return value if necessary
 		if (returnValue != null && returnValue == target &&
@@ -416,6 +421,11 @@ class CglibAopProxy implements AopProxy, Serializable {
 			throw new AopInvocationException(
 					"Null return value from advice does not match primitive return type for: " + method);
 		}
+		if (KotlinDetector.isSuspendingFunction(method)) {
+			return COROUTINES_FLOW_CLASS_NAME.equals(new MethodParameter(method, -1).getParameterType().getName()) ?
+					CoroutinesUtils.asFlow(returnValue) :
+					CoroutinesUtils.awaitSingleOrNull(returnValue, arguments[arguments.length - 1]);
+		}
 		return returnValue;
 	}
 
@@ -446,7 +456,7 @@ class CglibAopProxy implements AopProxy, Serializable {
 		@Nullable
 		public Object intercept(Object proxy, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
 			Object retVal = AopUtils.invokeJoinpointUsingReflection(this.target, method, args);
-			return processReturnType(proxy, this.target, method, retVal);
+			return processReturnType(proxy, this.target, method, args, retVal);
 		}
 	}
 
@@ -471,7 +481,7 @@ class CglibAopProxy implements AopProxy, Serializable {
 			try {
 				oldProxy = AopContext.setCurrentProxy(proxy);
 				Object retVal = AopUtils.invokeJoinpointUsingReflection(this.target, method, args);
-				return processReturnType(proxy, this.target, method, retVal);
+				return processReturnType(proxy, this.target, method, args, retVal);
 			}
 			finally {
 				AopContext.setCurrentProxy(oldProxy);
@@ -499,7 +509,7 @@ class CglibAopProxy implements AopProxy, Serializable {
 			Object target = this.targetSource.getTarget();
 			try {
 				Object retVal = AopUtils.invokeJoinpointUsingReflection(target, method, args);
-				return processReturnType(proxy, target, method, retVal);
+				return processReturnType(proxy, target, method, args, retVal);
 			}
 			finally {
 				if (target != null) {
@@ -529,7 +539,7 @@ class CglibAopProxy implements AopProxy, Serializable {
 			try {
 				oldProxy = AopContext.setCurrentProxy(proxy);
 				Object retVal = AopUtils.invokeJoinpointUsingReflection(target, method, args);
-				return processReturnType(proxy, target, method, retVal);
+				return processReturnType(proxy, target, method, args, retVal);
 			}
 			finally {
 				AopContext.setCurrentProxy(oldProxy);
@@ -656,7 +666,7 @@ class CglibAopProxy implements AopProxy, Serializable {
 					proxy, this.target, method, args, this.targetClass, this.adviceChain, methodProxy);
 			// If we get here, we need to create a MethodInvocation.
 			Object retVal = invocation.proceed();
-			retVal = processReturnType(proxy, this.target, method, retVal);
+			retVal = processReturnType(proxy, this.target, method, args, retVal);
 			return retVal;
 		}
 	}
@@ -706,7 +716,7 @@ class CglibAopProxy implements AopProxy, Serializable {
 					// We need to create a method invocation...
 					retVal = new CglibMethodInvocation(proxy, target, method, args, targetClass, chain, methodProxy).proceed();
 				}
-				return processReturnType(proxy, target, method, retVal);
+				return processReturnType(proxy, target, method, args, retVal);
 			}
 			finally {
 				if (target != null && !targetSource.isStatic()) {

spring-aop/src/main/java/org/springframework/aop/framework/CoroutinesUtils.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2002-2023 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.aop.framework;
+
+import kotlin.coroutines.Continuation;
+import kotlinx.coroutines.reactive.ReactiveFlowKt;
+import kotlinx.coroutines.reactor.MonoKt;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Package-visible class designed to avoid a hard dependency on Kotlin and Coroutines dependency at runtime.
+ *
+ * @author Sebastien Deleuze
+ * @since 6.1.0
+ */
+abstract class CoroutinesUtils {
+
+	static Object asFlow(Object publisher) {
+		return ReactiveFlowKt.asFlow((Publisher<?>) publisher);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Nullable
+	static Object awaitSingleOrNull(Object mono, Object continuation) {
+		return MonoKt.awaitSingleOrNull((Mono<?>) mono, (Continuation<Object>) continuation);
+	}
+
+}

spring-aop/src/main/java/org/springframework/aop/framework/JdkDynamicAopProxy.java

@@ -31,6 +31,8 @@ import org.springframework.aop.RawTargetAccess;
 import org.springframework.aop.TargetSource;
 import org.springframework.aop.support.AopUtils;
 import org.springframework.core.DecoratingProxy;
+import org.springframework.core.KotlinDetector;
+import org.springframework.core.MethodParameter;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -58,6 +60,7 @@ import org.springframework.util.ClassUtils;
  * @author Rob Harrop
  * @author Dave Syer
  * @author Sergey Tsypanov
+ * @author Sebastien Deleuze
  * @see java.lang.reflect.Proxy
  * @see AdvisedSupport
  * @see ProxyFactory
@@ -80,6 +83,8 @@ final class JdkDynamicAopProxy implements AopProxy, InvocationHandler, Serializa
 	/** We use a static Log to avoid serialization issues. */
 	private static final Log logger = LogFactory.getLog(JdkDynamicAopProxy.class);
 
+	private static final String COROUTINES_FLOW_CLASS_NAME = "kotlinx.coroutines.flow.Flow";
+
 	/** Config used to configure this proxy. */
 	private final AdvisedSupport advised;
 
@@ -258,6 +263,10 @@ final class JdkDynamicAopProxy implements AopProxy, InvocationHandler, Serializa
 				throw new AopInvocationException(
 						"Null return value from advice does not match primitive return type for: " + method);
 			}
+			if (KotlinDetector.isSuspendingFunction(method)) {
+				return COROUTINES_FLOW_CLASS_NAME.equals(new MethodParameter(method, -1).getParameterType().getName()) ?
+						CoroutinesUtils.asFlow(retVal) : CoroutinesUtils.awaitSingleOrNull(retVal, args[args.length - 1]);
+			}
 			return retVal;
 		}
 		finally {

spring-aop/src/main/java/org/springframework/aop/support/AopUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -25,6 +25,11 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 
+import kotlin.coroutines.Continuation;
+import kotlin.coroutines.CoroutineContext;
+import kotlinx.coroutines.Job;
+import org.reactivestreams.Publisher;
+
 import org.springframework.aop.Advisor;
 import org.springframework.aop.AopInvocationException;
 import org.springframework.aop.IntroductionAdvisor;
@@ -35,6 +40,8 @@ import org.springframework.aop.PointcutAdvisor;
 import org.springframework.aop.SpringProxy;
 import org.springframework.aop.TargetClassAware;
 import org.springframework.core.BridgeMethodResolver;
+import org.springframework.core.CoroutinesUtils;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.MethodIntrospector;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -53,6 +60,7 @@ import org.springframework.util.ReflectionUtils;
  * @author Rod Johnson
  * @author Juergen Hoeller
  * @author Rob Harrop
+ * @author Sebastien Deleuze
  * @see org.springframework.aop.framework.AopProxyUtils
  */
 public abstract class AopUtils {
@@ -340,7 +348,8 @@ public abstract class AopUtils {
 		// Use reflection to invoke the method.
 		try {
 			ReflectionUtils.makeAccessible(method);
-			return method.invoke(target, args);
+			return KotlinDetector.isSuspendingFunction(method) ?
+					KotlinDelegate.invokeSuspendingFunction(method, target, args) : method.invoke(target, args);
 		}
 		catch (InvocationTargetException ex) {
 			// Invoked method threw a checked exception.
@@ -356,4 +365,17 @@ public abstract class AopUtils {
 		}
 	}
 
+	/**
+	 * Inner class to avoid a hard dependency on Kotlin at runtime.
+	 */
+	private static class KotlinDelegate {
+
+		public static Publisher<?> invokeSuspendingFunction(Method method, Object target, Object... args) {
+			Continuation<?> continuation = (Continuation<?>) args[args.length -1];
+			CoroutineContext context = continuation.getContext().minusKey(Job.Key);
+			return CoroutinesUtils.invokeSuspendingFunction(context, method, target, args);
+		}
+
+	}
+
 }