Merge 050c2be776 into 033df68da6

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/CoRequestCacheable.kt

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache
+
+/**
+ * Annotation indicating that the result of invoking a *suspend method* can be cached
+ * for the lifespan of the underlying web request coroutine.
+ *
+ * Example:
+ *
+ * ```
+ * class MyServiceBean {
+ *
+ * 	@CoRequestCacheable(key = "#userName")
+ * 	suspend fun fetchUserAgeFromDownstreamService(userName: String, authHeader: String): Int {
+ * 		// prepare request and fetch the user info
+ * 		return userInfo.age
+ * 	}
+ *
+ * }
+ * ```
+ *
+ * Each time an advised suspend method is invoked, caching behavior will be applied,
+ * checking whether the method has been already invoked for the given arguments *within the same web request execution*.
+ * A sensible default simply uses the method parameters to compute the key, but
+ * a SpEL expression can be provided via the [key] attribute.
+ *
+ * If no value is found in the cache for the computed key, the target method
+ * will be invoked and the returned value will be stored in the coroutine context.
+ *
+ * Note that breaking
+ * [structured concurrency](https://kotlinlang.org/docs/coroutines-basics.html#coroutine-scope-and-structured-concurrency)
+ * by invoking the annotated method in a coroutine scope not tied to the web request, will prevent any caching behaviour.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ * @see EnableCoRequestCaching
+ */
+@Retention(AnnotationRetention.RUNTIME)
+@Target(AnnotationTarget.FUNCTION)
+annotation class CoRequestCacheable(
+
+	/**
+	 * Spring Expression Language (SpEL) expression for computing the key dynamically.
+	 *
+	 * The default value is `""`, meaning all method parameters are considered as a key.
+	 *
+	 * Method arguments can be accessed by index. For instance the second argument
+	 * can be accessed via `#p1` or `#a1`.
+	 * Arguments can also be accessed by name if that information is available.
+	 */
+	val key: String = ""
+)

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/EnableCoRequestCaching.kt

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache
+
+import org.springframework.context.annotation.AdviceMode
+import org.springframework.context.annotation.Import
+import org.springframework.core.Ordered
+import org.springframework.web.reactive.function.server.cache.config.CoRequestCacheConfigurationSelector
+
+/**
+ * Enables request-scoped cache management capability for Spring WebFlux servers with
+ * [Kotlin coroutines support enabled](https://docs.spring.io/spring-framework/reference/languages/kotlin/coroutines.html#dependencies).
+ *
+ * To be used together with @[Configuration](org.springframework.context.annotation.Configuration) classes as follows:
+ *
+ * ```
+ * @Configuration
+ * @EnableCoRequestCaching
+ * class AppConfig {
+ *
+ * 	@Bean
+ * 	fun myService(): MyService {
+ * 		// configure and return a class having @CoRequestCacheable suspend methods
+ * 		return MyService()
+ * 	}
+ *
+ * }
+ * ```
+ *
+ * Note that the only supported advice [mode] is [AdviceMode.PROXY],
+ * so local calls within the same class cannot get intercepted.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ * @see CoRequestCacheable
+ */
+@Target(AnnotationTarget.CLASS)
+@Retention(AnnotationRetention.RUNTIME)
+@Import(CoRequestCacheConfigurationSelector::class)
+annotation class EnableCoRequestCaching(
+
+	/**
+	 * Indicate whether subclass-based (CGLIB) proxies are to be created as opposed
+	 * to standard Java interface-based proxies. The default is `false`.
+	 */
+	val proxyTargetClass: Boolean = false,
+
+	/**
+	 * Indicate the ordering of the execution of the co-request caching advisor
+	 * when multiple advices are applied at a specific joinpoint.
+	 *
+	 * The default is [Ordered.LOWEST_PRECEDENCE].
+	 */
+	val order: Int = Ordered.LOWEST_PRECEDENCE,
+
+	/**
+	 * Indicate how caching advice should be applied.
+	 * The default and *only supported mode* is [AdviceMode.PROXY].
+	 */
+	val mode: AdviceMode = AdviceMode.PROXY
+)

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/config/CoRequestCacheConfiguration.kt

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.config
+
+import org.aopalliance.intercept.MethodInterceptor
+import org.springframework.aop.Advisor
+import org.springframework.beans.factory.config.BeanDefinition
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.ImportAware
+import org.springframework.context.annotation.Role
+import org.springframework.core.annotation.AnnotationAttributes
+import org.springframework.core.type.AnnotationMetadata
+import org.springframework.web.reactive.function.server.cache.EnableCoRequestCaching
+import org.springframework.web.reactive.function.server.cache.context.CoRequestCacheWebFilter
+import org.springframework.web.reactive.function.server.cache.interceptor.CoRequestCacheAdvisor
+import org.springframework.web.reactive.function.server.cache.interceptor.CoRequestCacheInterceptor
+import org.springframework.web.reactive.function.server.cache.interceptor.CoRequestCacheKeyGenerator
+import org.springframework.web.reactive.function.server.cache.operation.CoRequestCacheOperationSource
+import org.springframework.web.server.CoWebFilter
+import kotlin.reflect.jvm.jvmName
+
+/**
+ * `@Configuration` class that registers the Spring infrastructure beans necessary
+ * to enable proxy-based request-scoped cache for Spring WebFlux servers
+ * with Kotlin coroutines support enabled.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ * @see CoRequestCacheConfigurationSelector
+ */
+@Configuration(proxyBeanMethods = false)
+@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+internal class CoRequestCacheConfiguration : ImportAware {
+	lateinit var enableCoRequestCaching: AnnotationAttributes
+
+	override fun setImportMetadata(importMetadata: AnnotationMetadata) {
+		enableCoRequestCaching =
+			AnnotationAttributes.fromMap(
+				importMetadata.getAnnotationAttributes(EnableCoRequestCaching::class.jvmName)
+			) ?: throw IllegalArgumentException(
+				"@EnableCoRequestCaching is not present on importing class ${importMetadata.className}"
+			)
+	}
+
+	@Bean
+	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+	fun coRequestCacheOperationSource(): CoRequestCacheOperationSource = CoRequestCacheOperationSource()
+
+	@Bean
+	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+	fun coRequestCacheInterceptor(coRequestCacheOperationSource: CoRequestCacheOperationSource): MethodInterceptor =
+		CoRequestCacheInterceptor(
+			CoRequestCacheKeyGenerator(
+				coRequestCacheOperationSource
+			)
+		)
+
+	@Bean
+	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+	fun coRequestCacheAdvisor(
+		coRequestCacheOperationSource: CoRequestCacheOperationSource,
+		coRequestCacheInterceptor: MethodInterceptor
+	): Advisor =
+		CoRequestCacheAdvisor(
+			coRequestCacheOperationSource,
+			coRequestCacheInterceptor
+		)
+			.apply {
+				if (::enableCoRequestCaching.isInitialized) {
+					order = enableCoRequestCaching.getNumber("order")
+				}
+			}
+
+	@Bean
+	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+	fun coRequestCacheWebFilter(): CoWebFilter = CoRequestCacheWebFilter()
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/config/CoRequestCacheConfigurationSelector.kt

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.config
+
+import org.springframework.context.annotation.AdviceMode
+import org.springframework.context.annotation.AdviceModeImportSelector
+import org.springframework.context.annotation.AutoProxyRegistrar
+import org.springframework.web.reactive.function.server.cache.EnableCoRequestCaching
+import kotlin.reflect.jvm.jvmName
+
+private const val UNSUPPORTED_ADVISE_MODE_MESSAGE = "CoRequestCaching does not support aspectj advice mode"
+
+/**
+ * Select which classes to import according to the value of [EnableCoRequestCaching.mode] on the importing
+ * `@Configuration` class.
+ *
+ * Only [AdviceMode.PROXY] is currently supported.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ * @see CoRequestCacheConfiguration
+ */
+internal class CoRequestCacheConfigurationSelector : AdviceModeImportSelector<EnableCoRequestCaching>() {
+	override fun selectImports(adviceMode: AdviceMode): Array<out String> =
+		when (adviceMode) {
+			AdviceMode.PROXY -> arrayOf(AutoProxyRegistrar::class.jvmName, CoRequestCacheConfiguration::class.jvmName)
+			AdviceMode.ASPECTJ -> throw UnsupportedOperationException(UNSUPPORTED_ADVISE_MODE_MESSAGE)
+		}
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/context/CoRequestCacheContext.kt

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.context
+
+import org.reactivestreams.Publisher
+import java.util.concurrent.ConcurrentHashMap
+import kotlin.coroutines.AbstractCoroutineContextElement
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * A coroutine context element holding the [cache] values for
+ * [@CoRequestCacheable][org.springframework.web.reactive.function.server.cache.CoRequestCacheable]
+ * annotated methods.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal class CoRequestCacheContext(
+	val cache: ConcurrentHashMap<Any, Publisher<*>> = ConcurrentHashMap()
+) : AbstractCoroutineContextElement(Key) {
+	companion object Key : CoroutineContext.Key<CoRequestCacheContext>
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/context/CoRequestCacheWebFilter.kt

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.context
+
+import kotlinx.coroutines.withContext
+import org.springframework.web.server.CoWebFilter
+import org.springframework.web.server.CoWebFilterChain
+import org.springframework.web.server.ServerWebExchange
+
+/**
+ * Add a [CoRequestCacheContext] element to the web request coroutine context.
+ *
+ * This [CoWebFilter] is automatically registered when
+ * [EnableCoRequestCaching][org.springframework.web.reactive.function.server.cache.EnableCoRequestCaching]
+ * is applied to the app configuration.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal class CoRequestCacheWebFilter : CoWebFilter() {
+	override suspend fun filter(
+		exchange: ServerWebExchange,
+		chain: CoWebFilterChain
+	) {
+		return withContext(CoRequestCacheContext()) {
+			chain.filter(exchange)
+		}
+	}
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/interceptor/CoRequestCacheAdvisor.kt

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.interceptor
+
+import org.aopalliance.aop.Advice
+import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor
+import org.springframework.web.reactive.function.server.cache.operation.CoRequestCacheOperationSource
+import java.lang.reflect.Method
+
+/**
+ * Advisor driven by a [coRequestCacheOperationSource], used to match suspend methods that are cacheable for the lifespan
+ * of the coroutine handling a web request.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal class CoRequestCacheAdvisor(
+	val coRequestCacheOperationSource: CoRequestCacheOperationSource,
+	coRequestCacheAdvice: Advice,
+) : StaticMethodMatcherPointcutAdvisor(coRequestCacheAdvice) {
+	override fun matches(method: Method, targetClass: Class<*>): Boolean =
+		coRequestCacheOperationSource.hasCacheOperations(method, targetClass)
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/interceptor/CoRequestCacheInterceptor.kt

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.interceptor
+
+import org.aopalliance.intercept.MethodInterceptor
+import org.aopalliance.intercept.MethodInvocation
+import org.apache.commons.logging.LogFactory
+import org.springframework.cache.interceptor.KeyGenerator
+import org.springframework.web.reactive.function.server.cache.context.CoRequestCacheContext
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import kotlin.coroutines.Continuation
+import kotlin.reflect.jvm.jvmName
+
+private val logger = LogFactory.getLog(CoRequestCacheInterceptor::class.java)
+
+/**
+ * AOP Alliance MethodInterceptor for request-scoped caching of Kotlin suspend method invocations.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal class CoRequestCacheInterceptor(private val keyGenerator: KeyGenerator) : MethodInterceptor {
+
+	/**
+	 * Use the provided [keyGenerator] to generate a unique key for the intercepted suspend method call.
+	 *
+	 * When not already present for the generated key, create and store in the [CoRequestCacheContext] element of the
+	 * web request coroutine a lazy cached version of the [invocation] result:
+	 *
+	 * - A [shared Mono][Mono.share], for [Mono] type.
+	 * - A [buffered][Flux.buffer], [flattened][Flux.flatMapIterable], [replayed Flux][Flux.replay], for [Flux] type.
+	 *
+	 * The suspend method result is expected to be already converted to a reactive type by the
+	 * [AopUtils.invokeJoinpointUsingReflection][org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection]
+	 * AOP utility.
+	 *
+	 * @author Angelo Bracaglia
+	 * @since 7.0
+	 */
+	override fun invoke(invocation: MethodInvocation): Any? {
+		val coRequestCache =
+			(invocation.arguments.lastOrNull() as? Continuation<*>)
+				?.context[CoRequestCacheContext.Key]?.cache
+				?: run {
+					if (logger.isWarnEnabled) {
+						logger.warn(
+							"Skip CoRequestCaching for ${invocation.method}: coroutine cache context not available"
+						)
+					}
+					return invocation.proceed()
+				}
+
+		val targetObject = checkNotNull(invocation.getThis())
+
+		val coRequestCacheKey = keyGenerator.generate(targetObject, invocation.method, *invocation.arguments)
+
+		return coRequestCache.computeIfAbsent(coRequestCacheKey) {
+			when (val publisher = invocation.proceed()) {
+				is Mono<*> -> publisher.share()
+				is Flux<*> ->
+					publisher
+						.buffer()
+						.flatMapIterable { it }
+						.replay()
+						.refCount(1)
+
+				else -> throw IllegalArgumentException("Unexpected type ${publisher?.let { it::class.jvmName }}")
+			}
+		}
+	}
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/interceptor/CoRequestCacheKeyGenerator.kt

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.interceptor
+
+import org.springframework.aop.framework.AopProxyUtils
+import org.springframework.aop.support.AopUtils
+import org.springframework.cache.interceptor.KeyGenerator
+import org.springframework.cache.interceptor.SimpleKey
+import org.springframework.context.expression.AnnotatedElementKey
+import org.springframework.context.expression.CachedExpressionEvaluator
+import org.springframework.context.expression.MethodBasedEvaluationContext
+import org.springframework.core.BridgeMethodResolver
+import org.springframework.expression.Expression
+import org.springframework.web.reactive.function.server.cache.operation.CoRequestCacheOperationSource
+import java.lang.reflect.Method
+import java.lang.reflect.Proxy
+import java.util.concurrent.ConcurrentHashMap
+import kotlin.coroutines.Continuation
+
+/**
+ * Key generator for suspend method annotated by
+ * [@CoRequestCacheable][org.springframework.web.reactive.function.server.cache.CoRequestCacheable].
+ *
+ * If the only method parameter is the [Continuation] object, return a [NullaryMethodKey] instance,
+ * so that different beans with same method name still have distinct keys.
+ *
+ * If the method has other parameters, look for the
+ * [key expression][org.springframework.web.reactive.function.server.cache.CoRequestCacheable.key]
+ * using the [coRequestCacheOperationSource], and return a [SimpleKey] combining the nullary identity with
+ * the expression evaluation result, or with all the other parameters for a default blank key.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal class CoRequestCacheKeyGenerator(
+	private val coRequestCacheOperationSource: CoRequestCacheOperationSource,
+) : KeyGenerator, CachedExpressionEvaluator() {
+	private val bakedExpressions: MutableMap<ExpressionKey, Expression> = ConcurrentHashMap()
+
+	override fun generate(target: Any, method: Method, vararg params: Any?): Any {
+		check(params.lastOrNull() is Continuation<*>)
+
+		val targetClass = AopProxyUtils.ultimateTargetClass(target)
+		val nullaryMethodKey = NullaryMethodKey(targetClass, method.name)
+
+		if (params.size == 1) {
+			return nullaryMethodKey
+		}
+
+		val keyExpression: String = coRequestCacheKeyExpression(method, targetClass)
+
+		return if (keyExpression.isBlank()) {
+			SimpleKey(nullaryMethodKey, *params.copyOfRange(0, params.size - 1))
+		} else {
+			val targetMethod = ultimateTargetMethod(method, targetClass)
+
+			val expression =
+				getExpression(
+					this.bakedExpressions,
+					AnnotatedElementKey(targetMethod, targetClass),
+					keyExpression
+				)
+
+			val context =
+				MethodBasedEvaluationContext(
+					target,
+					targetMethod,
+					params,
+					parameterNameDiscoverer,
+				)
+
+			SimpleKey(nullaryMethodKey, expression.getValue(context))
+		}
+	}
+
+	private fun coRequestCacheKeyExpression(method: Method, targetClass: Class<*>): String {
+		val coRequestCacheOperations = coRequestCacheOperationSource.getCacheOperations(method, targetClass)
+
+		check(1 == coRequestCacheOperations?.size)
+
+		return coRequestCacheOperations.first().key
+	}
+
+	private fun ultimateTargetMethod(
+		method: Method,
+		targetClass: Class<*>
+	): Method {
+		var method = BridgeMethodResolver.findBridgedMethod(method)
+
+		if (!Proxy.isProxyClass(targetClass)) {
+			method = AopUtils.getMostSpecificMethod(method, targetClass)
+		}
+		return method
+	}
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/interceptor/NullaryMethodKey.kt

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.interceptor
+
+/**
+ * A unique identifier for a nullary method of a class.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal data class NullaryMethodKey(
+	val clazz: Class<*>,
+	val methodName: String
+)

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/operation/CoRequestCacheOperationSource.kt

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.operation
+
+import org.springframework.cache.interceptor.AbstractFallbackCacheOperationSource
+import org.springframework.cache.interceptor.CacheOperation
+import org.springframework.core.KotlinDetector
+import org.springframework.core.annotation.AnnotatedElementUtils
+import org.springframework.web.reactive.function.server.cache.CoRequestCacheable
+import java.lang.reflect.Method
+
+/**
+ * Implementation of [CacheOperationSource][org.springframework.cache.interceptor.CacheOperationSource]
+ * interface detecting [CoRequestCacheable] annotations on suspend methods and exposing the corresponding
+ * [CoRequestCacheableOperation].
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal class CoRequestCacheOperationSource : AbstractFallbackCacheOperationSource() {
+	override fun findCacheOperations(type: Class<*>): Collection<CacheOperation>? = null
+
+	override fun findCacheOperations(method: Method): Collection<CacheOperation>? {
+		if (!KotlinDetector.isSuspendingFunction(method)) return null
+
+		val coRequestCacheable =
+			AnnotatedElementUtils
+				.findMergedAnnotation(method, CoRequestCacheable::class.java) ?: return null
+
+		val coRequestCacheableOperation =
+			CoRequestCacheableOperation
+				.Builder()
+				.apply { key = coRequestCacheable.key }
+				.build()
+
+		return listOf(coRequestCacheableOperation)
+	}
+}

spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/server/cache/operation/CoRequestCacheableOperation.kt

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.operation
+
+import org.springframework.cache.interceptor.CacheOperation
+
+/**
+ * Class describing a cache 'CoRequestCacheable' operation.
+ *
+ * @author Angelo Bracaglia
+ * @since 7.0
+ */
+internal class CoRequestCacheableOperation(builder: Builder) : CacheOperation(builder) {
+
+	/**
+	 * A builder that can be used to create a [CoRequestCacheableOperation].
+	 *
+	 * @author Angelo Bracaglia
+	 * @since 7.0
+	 */
+	class Builder : CacheOperation.Builder() {
+		override fun build(): CacheOperation {
+			return CoRequestCacheableOperation(this)
+		}
+	}
+}

spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/cache/interceptor/CoRequestCacheAdvisorTests.kt

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.interceptor
+
+import io.mockk.every
+import io.mockk.mockk
+import org.aopalliance.aop.Advice
+import org.junit.jupiter.api.Test
+import org.springframework.web.reactive.function.server.cache.operation.CoRequestCacheOperationSource
+import java.lang.reflect.Method
+
+/**
+ * Tests for [CoRequestCacheAdvisor].
+ *
+ * @author Angelo Bracaglia
+ */
+class CoRequestCacheAdvisorTests {
+	private val target = mockk<Any>()
+	private val method = mockk<Method>()
+	private val coRequestCacheOperationSource = mockk<CoRequestCacheOperationSource>()
+
+	private val underTest = CoRequestCacheAdvisor(coRequestCacheOperationSource, mockk<Advice>())
+
+	@Test
+	fun `should match when operation source has cache operations`() {
+		every { coRequestCacheOperationSource.hasCacheOperations(method, target::class.java) } returns true
+		assert(underTest.matches(method, target::class.java))
+	}
+
+	@Test
+	fun `should not match when operation source does not have cache operations`() {
+		every { coRequestCacheOperationSource.hasCacheOperations(method, target::class.java) } returns false
+		assert(!underTest.matches(method, target::class.java))
+	}
+}

spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/cache/interceptor/CoRequestCacheInterceptorTests.kt

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.interceptor
+
+import io.mockk.every
+import io.mockk.mockk
+import org.aopalliance.intercept.MethodInvocation
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertInstanceOf
+import org.springframework.cache.interceptor.KeyGenerator
+import org.springframework.web.reactive.function.server.cache.context.CoRequestCacheContext
+import reactor.core.publisher.Mono
+import java.lang.reflect.Method
+import kotlin.coroutines.Continuation
+
+/**
+ * Tests for [CoRequestCacheInterceptor].
+ *
+ * @author Angelo Bracaglia
+ */
+class CoRequestCacheInterceptorTests {
+	private val coRequestCacheContext = CoRequestCacheContext()
+	private val continuation = mockk<Continuation<*>>()
+	private val target = mockk<Any>()
+	private val method = mockk<Method>()
+	private val argumentsWithContinuation = arrayOf("firstArgument", continuation)
+	private val keyGenerator = mockk<KeyGenerator>()
+	private lateinit var invocation: MethodInvocation
+
+	private val underTest = CoRequestCacheInterceptor(keyGenerator)
+
+	@BeforeEach
+	fun setup() {
+		invocation = mockk<MethodInvocation>()
+		every { invocation.`this` } returns target
+		every { invocation.method } returns method
+		every { invocation.arguments } returns argumentsWithContinuation
+		every { continuation.context[CoRequestCacheContext] } returns coRequestCacheContext
+		every { keyGenerator.generate(target, method, *argumentsWithContinuation) } returns "cacheKey"
+	}
+
+	@Test
+	fun `should cache the result of the intercepted suspend function within the same coroutine context`() {
+		every { invocation.proceed() } returns createExecutionsCounterMono()
+
+		val firsInvocationResult = underTest.invoke(invocation)
+		val secondInvocationResult = underTest.invoke(invocation)
+
+		assertThat(firsInvocationResult).isSameAs(secondInvocationResult)
+
+		val sharedMono = assertInstanceOf<Mono<Int>>(firsInvocationResult)
+
+		repeat(3) {
+			assertThat(sharedMono.block()).isEqualTo(1)
+		}
+	}
+
+	@Test
+	fun `should cache different results of the intercepted suspend function for different coroutine contexts`() {
+		every { invocation.proceed() } returns createExecutionsCounterMono()
+
+		val firstCoRequestCacheContext = CoRequestCacheContext()
+		every { continuation.context[CoRequestCacheContext] } returns firstCoRequestCacheContext
+		val firsInvocationResult = underTest.invoke(invocation)
+
+		val secondCoRequestCacheContext = CoRequestCacheContext()
+		every { continuation.context[CoRequestCacheContext] } returns secondCoRequestCacheContext
+		val secondInvocationResult = underTest.invoke(invocation)
+
+		assertThat(firsInvocationResult).isNotSameAs(secondInvocationResult)
+
+		val firstSharedMono = assertInstanceOf<Mono<Int>>(firsInvocationResult)
+		val secondSharedMono = assertInstanceOf<Mono<Int>>(secondInvocationResult)
+
+		repeat(3) {
+			assertThat(firstSharedMono.block()).isEqualTo(1)
+			assertThat(secondSharedMono.block()).isEqualTo(2)
+		}
+	}
+
+	@Test
+	fun `should cache different results of the intercepted suspend function for different cache keys`() {
+		every { invocation.proceed() } returns createExecutionsCounterMono()
+
+		val firstInvocationCacheKey = "firstCacheKey"
+		every { keyGenerator.generate(target, method, *argumentsWithContinuation) } returns firstInvocationCacheKey
+		val firsInvocationResult = underTest.invoke(invocation)
+
+		val secondInvocationCacheKey = "secondCacheKey"
+		every { keyGenerator.generate(target, method, *argumentsWithContinuation) } returns secondInvocationCacheKey
+		val secondInvocationResult = underTest.invoke(invocation)
+
+		assertThat(firsInvocationResult).isNotSameAs(secondInvocationResult)
+
+		val firstSharedMono = assertInstanceOf<Mono<Int>>(firsInvocationResult)
+		val secondSharedMono = assertInstanceOf<Mono<Int>>(secondInvocationResult)
+
+		repeat(3) {
+			assertThat(firstSharedMono.block()).isEqualTo(1)
+			assertThat(secondSharedMono.block()).isEqualTo(2)
+		}
+	}
+
+	@Test
+	fun `should skip caching of the intercepted function when no coroutine cache context is available`() {
+		every { invocation.proceed() } returns createExecutionsCounterMono()
+		every { continuation.context[CoRequestCacheContext] } returns null
+
+		val mono = assertInstanceOf<Mono<Int>>(underTest.invoke(invocation))
+
+		repeat(3) {
+			val expectedExecutionsCount = it + 1
+			assertThat(mono.block()).isEqualTo(expectedExecutionsCount)
+		}
+	}
+
+	private fun createExecutionsCounterMono(): Mono<Int> {
+		var executionsCount = 0
+		return Mono.defer {
+			executionsCount++
+			Mono.just(executionsCount)
+		}
+	}
+}

spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/cache/interceptor/CoRequestCacheKeyGeneratorTests.kt

@@ -0,0 +1,125 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.interceptor
+
+import io.mockk.clearMocks
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.delay
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import org.springframework.cache.interceptor.SimpleKey
+import org.springframework.web.reactive.function.server.cache.operation.CoRequestCacheOperationSource
+import org.springframework.web.reactive.function.server.cache.operation.CoRequestCacheableOperation
+import java.lang.reflect.Method
+import kotlin.coroutines.Continuation
+
+private const val SAMPLE_METHOD_NAME = "sampleMethodName"
+
+/**
+ * Tests for [CoRequestCacheKeyGenerator].
+ *
+ * @author Angelo Bracaglia
+ */
+class CoRequestCacheKeyGeneratorTests {
+	private val coRequestCacheOperationSource = mockk<CoRequestCacheOperationSource>()
+	private val target = mockk<Any>()
+	private val method = mockk<Method>()
+	private val continuation = mockk<Continuation<*>>()
+	private val underTest = CoRequestCacheKeyGenerator(coRequestCacheOperationSource)
+
+	@BeforeEach
+	fun setup() {
+		every { method.name } returns SAMPLE_METHOD_NAME
+	}
+
+	@AfterEach
+	fun teardown() {
+		clearMocks(coRequestCacheOperationSource, target, method)
+	}
+
+	@Test
+	fun `should throw an IllegalStateException when used for a not-suspend method`() {
+		assertThrows<IllegalStateException> {
+			underTest.generate(target, method, "notContinuationObject")
+		}
+	}
+
+	@Test
+	fun `should return a NullaryMethodKey when the only method parameter is a continuation object`() {
+		val key = underTest.generate(target, method, continuation)
+
+		val expectedKey = NullaryMethodKey(target::class.java, SAMPLE_METHOD_NAME)
+		assertThat(key).isEqualTo(expectedKey)
+	}
+
+	@Test
+	fun `should return a SimpleKey combining the NullaryMethodKey and all the arguments for empty key expression`() {
+		val coRequestCacheableOperation = CoRequestCacheableOperation.Builder().apply { key = "" }.build()
+		every { coRequestCacheOperationSource.getCacheOperations(method, target::class.java) } returns
+				listOf(coRequestCacheableOperation)
+
+		val firstParameterValue = "firstParameterValue"
+		val secondParameterValue = 2
+		val key = underTest.generate(target, method, firstParameterValue, secondParameterValue, continuation)
+
+		val expectedKey = SimpleKey(
+			NullaryMethodKey(target::class.java, SAMPLE_METHOD_NAME),
+			firstParameterValue,
+			secondParameterValue
+		)
+		assertThat(key).isEqualTo(expectedKey)
+	}
+
+	@Test
+	fun `should return a SimpleKey combining the NullaryMethodKey and evaluated key expression`() {
+		class SampleBean {
+			@Suppress("Unused")
+			suspend fun sampleMethod(firstParameter: String, secondParameter: Int) {
+				delay(100)
+			}
+		}
+
+		val sampleBeanInstance = SampleBean()
+		val sampleMethod = SampleBean::class.java.declaredMethods.first()
+
+		val coRequestCacheableOperation =
+			CoRequestCacheableOperation.Builder().apply { key = "#firstParameter" }.build()
+		every {
+			coRequestCacheOperationSource.getCacheOperations(sampleMethod, sampleBeanInstance::class.java)
+		} returns listOf(coRequestCacheableOperation)
+
+		val firstParameterValue = "firstParameterValue"
+		val secondParameterValue = 2
+
+		val key = underTest.generate(
+			sampleBeanInstance,
+			sampleMethod,
+			firstParameterValue, secondParameterValue,
+			continuation
+		)
+
+		val expectedKey = SimpleKey(
+			NullaryMethodKey(SampleBean::class.java, sampleMethod.name),
+			firstParameterValue
+		)
+		assertThat(key).isEqualTo(expectedKey)
+	}
+}

spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/server/cache/operation/CoRequestCacheOperationSourceTests.kt

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2002-present 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.web.reactive.function.server.cache.operation
+
+import kotlinx.coroutines.delay
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertInstanceOf
+import org.junit.jupiter.api.assertNotNull
+import org.springframework.web.reactive.function.server.cache.CoRequestCacheable
+
+private const val SAMPLE_CACHE_KEY = "sampleCacheKey"
+private const val ANNOTATED_SUSPEND_METHOD_NAME = "annotatedSuspendMethod"
+private const val ANNOTATED_METHOD_NAME = "annotatedMethod"
+private const val NOT_ANNOTATED_SUSPEND_METHOD_NAME = "notAnnotatedSuspendMethod"
+
+/**
+ * Tests for [CoRequestCacheOperationSource].
+ *
+ * @author Angelo Bracaglia
+ */
+class CoRequestCacheOperationSourceTests {
+	class SampleBean {
+		@Suppress("Unused")
+		@CoRequestCacheable(key = SAMPLE_CACHE_KEY)
+		suspend fun annotatedSuspendMethod() {
+			delay(10)
+		}
+
+		@Suppress("Unused")
+		@CoRequestCacheable(key = SAMPLE_CACHE_KEY)
+		fun annotatedMethod() {
+		}
+
+		@Suppress("Unused")
+		suspend fun notAnnotatedSuspendMethod() {
+			delay(10)
+		}
+	}
+
+	private val underTest = CoRequestCacheOperationSource()
+
+	@Test
+	fun `should have CoRequestCacheableOperation when the given method is suspend and annotated by CoRequestCacheable`() {
+		val target = SampleBean()
+		val method = target::class.java.declaredMethods.first { it.name == ANNOTATED_SUSPEND_METHOD_NAME }
+
+		assert(underTest.hasCacheOperations(method, SampleBean::class.java))
+
+		val cacheOperations = underTest.getCacheOperations(method, SampleBean::class.java)
+		assertNotNull(cacheOperations)
+		assertThat(cacheOperations.size).isEqualTo(1)
+
+		val coRequestCacheableOperation = assertInstanceOf<CoRequestCacheableOperation>(cacheOperations.first())
+		assertThat(coRequestCacheableOperation.key).isEqualTo(SAMPLE_CACHE_KEY)
+	}
+
+	@Test
+	fun `should not have CoRequestCacheableOperation when the given method is annotated by CoRequestCacheable but not suspend`() {
+		val target = SampleBean()
+		val method = target::class.java.declaredMethods.first { it.name == ANNOTATED_METHOD_NAME }
+
+		assert(!underTest.hasCacheOperations(method, SampleBean::class.java))
+	}
+
+	@Test
+	fun `should not have CoRequestCacheableOperation when the given method is suspend but not annotated by CoRequestCacheable`() {
+		val target = SampleBean()
+		val method = target::class.java.declaredMethods.first { it.name == NOT_ANNOTATED_SUSPEND_METHOD_NAME }
+
+		assert(!underTest.hasCacheOperations(method, SampleBean::class.java))
+	}
+}