@ -24,11 +24,16 @@ import java.util.Collections;
import java.util.List ;
import java.util.List ;
import java.util.Map ;
import java.util.Map ;
import java.util.Optional ;
import java.util.Optional ;
import java.util.concurrent.CompletableFuture ;
import java.util.concurrent.ConcurrentHashMap ;
import java.util.concurrent.ConcurrentHashMap ;
import java.util.function.Supplier ;
import java.util.function.Supplier ;
import org.apache.commons.logging.Log ;
import org.apache.commons.logging.Log ;
import org.apache.commons.logging.LogFactory ;
import org.apache.commons.logging.LogFactory ;
import org.reactivestreams.Subscriber ;
import org.reactivestreams.Subscription ;
import reactor.core.publisher.Flux ;
import reactor.core.publisher.Mono ;
import org.springframework.aop.framework.AopProxyUtils ;
import org.springframework.aop.framework.AopProxyUtils ;
import org.springframework.aop.support.AopUtils ;
import org.springframework.aop.support.AopUtils ;
@ -43,6 +48,8 @@ import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager ;
import org.springframework.cache.CacheManager ;
import org.springframework.context.expression.AnnotatedElementKey ;
import org.springframework.context.expression.AnnotatedElementKey ;
import org.springframework.core.BridgeMethodResolver ;
import org.springframework.core.BridgeMethodResolver ;
import org.springframework.core.ReactiveAdapter ;
import org.springframework.core.ReactiveAdapterRegistry ;
import org.springframework.expression.EvaluationContext ;
import org.springframework.expression.EvaluationContext ;
import org.springframework.lang.Nullable ;
import org.springframework.lang.Nullable ;
import org.springframework.util.Assert ;
import org.springframework.util.Assert ;
@ -83,12 +90,18 @@ import org.springframework.util.function.SupplierUtils;
public abstract class CacheAspectSupport extends AbstractCacheInvoker
public abstract class CacheAspectSupport extends AbstractCacheInvoker
implements BeanFactoryAware , InitializingBean , SmartInitializingSingleton {
implements BeanFactoryAware , InitializingBean , SmartInitializingSingleton {
private static final boolean reactiveStreamsPresent = ClassUtils . isPresent (
"org.reactivestreams.Publisher" , CacheAspectSupport . class . getClassLoader ( ) ) ;
protected final Log logger = LogFactory . getLog ( getClass ( ) ) ;
protected final Log logger = LogFactory . getLog ( getClass ( ) ) ;
private final Map < CacheOperationCacheKey , CacheOperationMetadata > metadataCache = new ConcurrentHashMap < > ( 1024 ) ;
private final Map < CacheOperationCacheKey , CacheOperationMetadata > metadataCache = new ConcurrentHashMap < > ( 1024 ) ;
private final CacheOperationExpressionEvaluator evaluator = new CacheOperationExpressionEvaluator ( ) ;
private final CacheOperationExpressionEvaluator evaluator = new CacheOperationExpressionEvaluator ( ) ;
@Nullable
private final ReactiveCachingHandler reactiveCachingHandler ;
@Nullable
@Nullable
private CacheOperationSource cacheOperationSource ;
private CacheOperationSource cacheOperationSource ;
@ -103,6 +116,11 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
private boolean initialized = false ;
private boolean initialized = false ;
protected CacheAspectSupport ( ) {
this . reactiveCachingHandler = ( reactiveStreamsPresent ? new ReactiveCachingHandler ( ) : null ) ;
}
/ * *
/ * *
* Configure this aspect with the given error handler , key generator and cache resolver / manager
* Configure this aspect with the given error handler , key generator and cache resolver / manager
* suppliers , applying the corresponding default if a supplier is not resolvable .
* suppliers , applying the corresponding default if a supplier is not resolvable .
@ -371,41 +389,25 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
}
}
@Nullable
@Nullable
private Object execute ( final CacheOperationInvoker invoker , Method method , CacheOperationContexts contexts ) {
private Object execute ( CacheOperationInvoker invoker , Method method , CacheOperationContexts contexts ) {
// Special handling of synchronized invocation
if ( contexts . isSynchronized ( ) ) {
if ( contexts . isSynchronized ( ) ) {
CacheOperationContext context = contexts . get ( CacheableOperation . class ) . iterator ( ) . next ( ) ;
// Special handling of synchronized invocation
if ( isConditionPassing ( context , CacheOperationExpressionEvaluator . NO_RESULT ) ) {
return executeSynchronized ( invoker , method , contexts ) ;
Object key = generateKey ( context , CacheOperationExpressionEvaluator . NO_RESULT ) ;
Cache cache = context . getCaches ( ) . iterator ( ) . next ( ) ;
try {
return wrapCacheValue ( method , handleSynchronizedGet ( invoker , key , cache ) ) ;
}
catch ( Cache . ValueRetrievalException ex ) {
// Directly propagate ThrowableWrapper from the invoker,
// or potentially also an IllegalArgumentException etc.
ReflectionUtils . rethrowRuntimeException ( ex . getCause ( ) ) ;
}
}
else {
// No caching required, just call the underlying method
return invokeOperation ( invoker ) ;
}
}
}
// Process any early evictions
// Process any early evictions
processCacheEvicts ( contexts . get ( CacheEvictOperation . class ) , true ,
processCacheEvicts ( contexts . get ( CacheEvictOperation . class ) , true ,
CacheOperationExpressionEvaluator . NO_RESULT ) ;
CacheOperationExpressionEvaluator . NO_RESULT ) ;
// Check if we have a cached item matching the conditions
// Check if we have a cached value matching the conditions
Cache . ValueWrapper cacheHit = findCachedItem ( contexts . get ( CacheableOperation . class ) ) ;
Object cacheHit = findCachedValue ( contexts . get ( CacheableOperation . class ) ) ;
Object cacheValue ;
Object cacheValue ;
Object returnValue ;
Object returnValue ;
if ( cacheHit ! = null & & ! hasCachePut ( contexts ) ) {
if ( cacheHit ! = null & & ! hasCachePut ( contexts ) ) {
// If there are no put requests, just use the cache hit
// If there are no put requests, just use the cache hit
cacheValue = cacheHit . get ( ) ;
cacheValue = ( cacheHit instanceof Cache . ValueWrapper wrapper ? wrapper . get ( ) : cacheHit ) ;
returnValue = wrapCacheValue ( method , cacheValue ) ;
returnValue = wrapCacheValue ( method , cacheValue ) ;
}
}
else {
else {
@ -414,8 +416,8 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
cacheValue = unwrapReturnValue ( returnValue ) ;
cacheValue = unwrapReturnValue ( returnValue ) ;
}
}
// Collect puts from any @Cacheable miss, if no cached item is found
// Collect puts from any @Cacheable miss, if no cached value is found
List < CachePutRequest > cachePutRequests = new ArrayList < > ( ) ;
List < CachePutRequest > cachePutRequests = new ArrayList < > ( 1 ) ;
if ( cacheHit = = null ) {
if ( cacheHit = = null ) {
collectPutRequests ( contexts . get ( CacheableOperation . class ) , cacheValue , cachePutRequests ) ;
collectPutRequests ( contexts . get ( CacheableOperation . class ) , cacheValue , cachePutRequests ) ;
}
}
@ -425,29 +427,52 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
// Process any collected put requests, either from @CachePut or a @Cacheable miss
// Process any collected put requests, either from @CachePut or a @Cacheable miss
for ( CachePutRequest cachePutRequest : cachePutRequests ) {
for ( CachePutRequest cachePutRequest : cachePutRequests ) {
cachePutRequest . apply ( cacheValue ) ;
Object returnOverride = cachePutRequest . apply ( cacheValue ) ;
if ( returnOverride ! = null ) {
returnValue = returnOverride ;
}
}
}
// Process any late evictions
// Process any late evictions
processCacheEvicts ( contexts . get ( CacheEvictOperation . class ) , false , cacheValue ) ;
Object returnOverride = processCacheEvicts (
contexts . get ( CacheEvictOperation . class ) , false , returnValue ) ;
if ( returnOverride ! = null ) {
returnValue = returnOverride ;
}
return returnValue ;
return returnValue ;
}
}
@Nullable
@Nullable
private Object handleSynchronizedGet ( CacheOperationInvoker invoker , Object key , Cache cache ) {
private Object executeSynchronized ( CacheOperationInvoker invoker , Method method , CacheOperationContexts contexts ) {
InvocationAwareResult invocationResult = new InvocationAwareResult ( ) ;
CacheOperationContext context = contexts . get ( CacheableOperation . class ) . iterator ( ) . next ( ) ;
Object result = cache . get ( key , ( ) - > {
if ( isConditionPassing ( context , CacheOperationExpressionEvaluator . NO_RESULT ) ) {
invocationResult . invoked = true ;
Object key = generateKey ( context , CacheOperationExpressionEvaluator . NO_RESULT ) ;
if ( logger . isTraceEnabled ( ) ) {
Cache cache = context . getCaches ( ) . iterator ( ) . next ( ) ;
logger . trace ( "No cache entry for key '" + key + "' in cache " + cache . getName ( ) ) ;
if ( CompletableFuture . class . isAssignableFrom ( method . getReturnType ( ) ) ) {
return cache . retrieve ( key , ( ) - > ( CompletableFuture < ? > ) invokeOperation ( invoker ) ) ;
}
if ( this . reactiveCachingHandler ! = null ) {
Object returnValue = this . reactiveCachingHandler . executeSynchronized ( invoker , method , cache , key ) ;
if ( returnValue ! = ReactiveCachingHandler . NOT_HANDLED ) {
return returnValue ;
}
}
}
return unwrapReturnValue ( invokeOperation ( invoker ) ) ;
try {
} ) ;
return wrapCacheValue ( method , cache . get ( key , ( ) - > unwrapReturnValue ( invokeOperation ( invoker ) ) ) ) ;
if ( ! invocationResult . invoked & & logger . isTraceEnabled ( ) ) {
}
logger . trace ( "Cache entry for key '" + key + "' found in cache '" + cache . getName ( ) + "'" ) ;
catch ( Cache . ValueRetrievalException ex ) {
// Directly propagate ThrowableWrapper from the invoker,
// or potentially also an IllegalArgumentException etc.
ReflectionUtils . rethrowRuntimeException ( ex . getCause ( ) ) ;
// Never reached
return null ;
}
}
else {
// No caching required, just call the underlying method
return invokeOperation ( invoker ) ;
}
}
return result ;
}
}
@Nullable
@Nullable
@ -467,7 +492,7 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
private boolean hasCachePut ( CacheOperationContexts contexts ) {
private boolean hasCachePut ( CacheOperationContexts contexts ) {
// Evaluate the conditions *without* the result object because we don't have it yet...
// Evaluate the conditions *without* the result object because we don't have it yet...
Collection < CacheOperationContext > cachePutContexts = contexts . get ( CachePutOperation . class ) ;
Collection < CacheOperationContext > cachePutContexts = contexts . get ( CachePutOperation . class ) ;
Collection < CacheOperationContext > excluded = new ArrayList < > ( ) ;
Collection < CacheOperationContext > excluded = new ArrayList < > ( 1 ) ;
for ( CacheOperationContext context : cachePutContexts ) {
for ( CacheOperationContext context : cachePutContexts ) {
try {
try {
if ( ! context . isConditionPassing ( CacheOperationExpressionEvaluator . RESULT_UNAVAILABLE ) ) {
if ( ! context . isConditionPassing ( CacheOperationExpressionEvaluator . RESULT_UNAVAILABLE ) ) {
@ -482,32 +507,55 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
return ( cachePutContexts . size ( ) ! = excluded . size ( ) ) ;
return ( cachePutContexts . size ( ) ! = excluded . size ( ) ) ;
}
}
private void processCacheEvicts (
@Nullable
Collection < CacheOperationContext > contexts , boolean beforeInvocation , @Nullable Object result ) {
private Object processCacheEvicts ( Collection < CacheOperationContext > contexts , boolean beforeInvocation ,
@Nullable Object result ) {
for ( CacheOperationContext context : contexts ) {
if ( contexts . isEmpty ( ) ) {
CacheEvictOperation operation = ( CacheEvictOperation ) context . metadata . operation ;
return null ;
if ( beforeInvocation = = operation . isBeforeInvocation ( ) & & isConditionPassing ( context , result ) ) {
}
performCacheEvict ( context , operation , result ) ;
List < CacheOperationContext > applicable = contexts . stream ( )
. filter ( context - > ( context . metadata . operation instanceof CacheEvictOperation evict & &
beforeInvocation = = evict . isBeforeInvocation ( ) ) ) . toList ( ) ;
if ( applicable . isEmpty ( ) ) {
return null ;
}
if ( result instanceof CompletableFuture < ? > future ) {
return future . whenComplete ( ( value , ex ) - > {
if ( ex = = null ) {
performCacheEvicts ( applicable , result ) ;
}
} ) ;
}
if ( this . reactiveCachingHandler ! = null ) {
Object returnValue = this . reactiveCachingHandler . processCacheEvicts ( applicable , result ) ;
if ( returnValue ! = ReactiveCachingHandler . NOT_HANDLED ) {
return returnValue ;
}
}
}
}
performCacheEvicts ( applicable , result ) ;
return null ;
}
}
private void performCacheEvict (
private void performCacheEvicts ( List < CacheOperationContext > contexts , @Nullable Object result ) {
CacheOperationContext context , CacheEvictOperation operation , @Nullable Object result ) {
for ( CacheOperationContext context : contexts ) {
CacheEvictOperation operation = ( CacheEvictOperation ) context . metadata . operation ;
Object key = null ;
if ( isConditionPassing ( context , result ) ) {
for ( Cache cache : context . getCaches ( ) ) {
Object key = null ;
if ( operation . isCacheWide ( ) ) {
for ( Cache cache : context . getCaches ( ) ) {
logInvalidating ( context , operation , null ) ;
if ( operation . isCacheWide ( ) ) {
doClear ( cache , operation . isBeforeInvocation ( ) ) ;
logInvalidating ( context , operation , null ) ;
}
doClear ( cache , operation . isBeforeInvocation ( ) ) ;
else {
}
if ( key = = null ) {
else {
key = generateKey ( context , result ) ;
if ( key = = null ) {
key = generateKey ( context , result ) ;
}
logInvalidating ( context , operation , key ) ;
doEvict ( cache , key , operation . isBeforeInvocation ( ) ) ;
}
}
}
logInvalidating ( context , operation , key ) ;
doEvict ( cache , key , operation . isBeforeInvocation ( ) ) ;
}
}
}
}
}
}
@ -520,19 +568,21 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
}
}
/ * *
/ * *
* Find a cached item only for { @link CacheableOperation } that passes the condition .
* Find a cached value only for { @link CacheableOperation } that passes the condition .
* @param contexts the cacheable operations
* @param contexts the cacheable operations
* @return a { @link Cache . ValueWrapper } holding the cached item ,
* @return a { @link Cache . ValueWrapper } holding the cached value ,
* or { @code null } if none is found
* or { @code null } if none is found
* /
* /
@Nullable
@Nullable
private Cache . ValueWrapper findCachedItem ( Collection < CacheOperationContext > contexts ) {
private Object findCachedValue ( Collection < CacheOperationContext > contexts ) {
Object result = CacheOperationExpressionEvaluator . NO_RESULT ;
for ( CacheOperationContext context : contexts ) {
for ( CacheOperationContext context : contexts ) {
if ( isConditionPassing ( context , result ) ) {
if ( isConditionPassing ( context , CacheOperationExpressionEvaluator . NO_RESULT ) ) {
Object key = generateKey ( context , result ) ;
Object key = generateKey ( context , CacheOperationExpressionEvaluator . NO_RESULT ) ;
Cache . ValueWrapper cached = findInCaches ( context , key ) ;
Object cached = findInCaches ( context , key ) ;
if ( cached ! = null ) {
if ( cached ! = null ) {
if ( logger . isTraceEnabled ( ) ) {
logger . trace ( "Cache entry for key '" + key + "' found in cache(s) " + context . getCacheNames ( ) ) ;
}
return cached ;
return cached ;
}
}
else {
else {
@ -547,9 +597,9 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
/ * *
/ * *
* Collect the { @link CachePutRequest } for all { @link CacheOperation } using
* Collect the { @link CachePutRequest } for all { @link CacheOperation } using
* the specified result item .
* the specified result value .
* @param contexts the contexts to handle
* @param contexts the contexts to handle
* @param result the result item ( never { @code null } )
* @param result the result value ( never { @code null } )
* @param putRequests the collection to update
* @param putRequests the collection to update
* /
* /
private void collectPutRequests ( Collection < CacheOperationContext > contexts ,
private void collectPutRequests ( Collection < CacheOperationContext > contexts ,
@ -564,15 +614,18 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
}
}
@Nullable
@Nullable
private Cache . ValueWrapper findInCaches ( CacheOperationContext context , Object key ) {
private Object findInCaches ( CacheOperationContext context , Object key ) {
for ( Cache cache : context . getCaches ( ) ) {
for ( Cache cache : context . getCaches ( ) ) {
Cache . ValueWrapper wrapper = doGet ( cache , key ) ;
if ( CompletableFuture . class . isAssignableFrom ( context . getMethod ( ) . getReturnType ( ) ) ) {
if ( wrapper ! = null ) {
return cache . retrieve ( key ) ;
if ( logger . isTraceEnabled ( ) ) {
}
logger . trace ( "Cache entry for key '" + key + "' found in cache '" + cache . getName ( ) + "'" ) ;
if ( this . reactiveCachingHandler ! = null ) {
Object returnValue = this . reactiveCachingHandler . findInCaches ( context , cache , key ) ;
if ( returnValue ! = ReactiveCachingHandler . NOT_HANDLED ) {
return returnValue ;
}
}
return wrapper ;
}
}
return doGet ( cache , key ) ;
}
}
return null ;
return null ;
}
}
@ -625,13 +678,13 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
}
}
private boolean determineSyncFlag ( Method method ) {
private boolean determineSyncFlag ( Method method ) {
List < CacheOperationContext > cacheOperation Contexts = this . contexts . get ( CacheableOperation . class ) ;
List < CacheOperationContext > cacheable Contexts = this . contexts . get ( CacheableOperation . class ) ;
if ( cacheOperation Contexts = = null ) { // no @Cacheable operation at all
if ( cacheable Contexts = = null ) { // no @Cacheable operation at all
return false ;
return false ;
}
}
boolean syncEnabled = false ;
boolean syncEnabled = false ;
for ( CacheOperationContext cacheOperationC ontext : cacheOperation Contexts ) {
for ( CacheOperationContext context : cacheable Contexts ) {
if ( ( ( CacheableOperation ) cacheOperationContext . getOperation ( ) ) . isSync ( ) ) {
if ( context . getOperation ( ) instanceof CacheableOperation cacheable & & cacheable . isSync ( ) ) {
syncEnabled = true ;
syncEnabled = true ;
break ;
break ;
}
}
@ -641,13 +694,13 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
throw new IllegalStateException (
throw new IllegalStateException (
"A sync=true operation cannot be combined with other cache operations on '" + method + "'" ) ;
"A sync=true operation cannot be combined with other cache operations on '" + method + "'" ) ;
}
}
if ( cacheOperation Contexts . size ( ) > 1 ) {
if ( cacheable Contexts . size ( ) > 1 ) {
throw new IllegalStateException (
throw new IllegalStateException (
"Only one sync=true operation is allowed on '" + method + "'" ) ;
"Only one sync=true operation is allowed on '" + method + "'" ) ;
}
}
CacheOperationContext cacheOperationContext = cacheOperation Contexts . iterator ( ) . next ( ) ;
CacheOperationContext cacheableContext = cacheable Contexts . iterator ( ) . next ( ) ;
CacheOperation operation = cacheOperation Context . getOperation ( ) ;
CacheOperation operation = cacheable Context . getOperation ( ) ;
if ( cacheOperation Context . getCaches ( ) . size ( ) > 1 ) {
if ( cacheable Context . getCaches ( ) . size ( ) > 1 ) {
throw new IllegalStateException (
throw new IllegalStateException (
"A sync=true operation is restricted to a single cache on '" + operation + "'" ) ;
"A sync=true operation is restricted to a single cache on '" + operation + "'" ) ;
}
}
@ -720,7 +773,7 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
this . args = extractArgs ( metadata . method , args ) ;
this . args = extractArgs ( metadata . method , args ) ;
this . target = target ;
this . target = target ;
this . caches = CacheAspectSupport . this . getCaches ( this , metadata . cacheResolver ) ;
this . caches = CacheAspectSupport . this . getCaches ( this , metadata . cacheResolver ) ;
this . cacheNames = creat eCacheNames( this . caches ) ;
this . cacheNames = prepar eCacheNames( this . caches ) ;
}
}
@Override
@Override
@ -808,8 +861,8 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
return this . cacheNames ;
return this . cacheNames ;
}
}
private Collection < String > creat eCacheNames( Collection < ? extends Cache > caches ) {
private Collection < String > prepar eCacheNames( Collection < ? extends Cache > caches ) {
Collection < String > names = new ArrayList < > ( ) ;
Collection < String > names = new ArrayList < > ( caches . size ( ) ) ;
for ( Cache cache : caches ) {
for ( Cache cache : caches ) {
names . add ( cache . getName ( ) ) ;
names . add ( cache . getName ( ) ) ;
}
}
@ -818,25 +871,6 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
}
}
private class CachePutRequest {
private final CacheOperationContext context ;
private final Object key ;
public CachePutRequest ( CacheOperationContext context , Object key ) {
this . context = context ;
this . key = key ;
}
public void apply ( @Nullable Object result ) {
for ( Cache cache : this . context . getCaches ( ) ) {
doPut ( cache , this . key , result ) ;
}
}
}
private static final class CacheOperationCacheKey implements Comparable < CacheOperationCacheKey > {
private static final class CacheOperationCacheKey implements Comparable < CacheOperationCacheKey > {
private final CacheOperation cacheOperation ;
private final CacheOperation cacheOperation ;
@ -876,12 +910,168 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
}
}
private class CachePutRequest {
private final CacheOperationContext context ;
private final Object key ;
public CachePutRequest ( CacheOperationContext context , Object key ) {
this . context = context ;
this . key = key ;
}
@Nullable
public Object apply ( @Nullable Object result ) {
if ( result instanceof CompletableFuture < ? > future ) {
return future . whenComplete ( ( value , ex ) - > {
if ( ex ! = null ) {
performEvict ( ex ) ;
}
else {
performPut ( value ) ;
}
} ) ;
}
if ( reactiveCachingHandler ! = null ) {
Object returnValue = reactiveCachingHandler . processPutRequest ( this , result ) ;
if ( returnValue ! = ReactiveCachingHandler . NOT_HANDLED ) {
return returnValue ;
}
}
performPut ( result ) ;
return null ;
}
void performPut ( @Nullable Object value ) {
if ( logger . isTraceEnabled ( ) ) {
logger . trace ( "Creating cache entry for key '" + this . key + "' in cache(s) " +
this . context . getCacheNames ( ) ) ;
}
for ( Cache cache : this . context . getCaches ( ) ) {
doPut ( cache , this . key , value ) ;
}
}
void performEvict ( Throwable cause ) {
if ( logger . isTraceEnabled ( ) ) {
logger . trace ( "Removing cache entry for key '" + this . key + "' from cache(s) " +
this . context . getCacheNames ( ) + " due to exception: " + cause ) ;
}
for ( Cache cache : this . context . getCaches ( ) ) {
doEvict ( cache , this . key , false ) ;
}
}
}
/ * *
/ * *
* Internal holder class for recording that a cache method was invoked .
* Reactive Streams Subscriber collection for collecting a List to cache .
* /
* /
private static class InvocationAwareResult {
private class CachePutListSubscriber implements Subscriber < Object > {
private final CachePutRequest request ;
boolean invoked ;
private final List < Object > cacheValue = new ArrayList < > ( ) ;
public CachePutListSubscriber ( CachePutRequest request ) {
this . request = request ;
}
@Override
public void onSubscribe ( Subscription s ) {
s . request ( Integer . MAX_VALUE ) ;
}
@Override
public void onNext ( Object o ) {
this . cacheValue . add ( o ) ;
}
@Override
public void onError ( Throwable t ) {
this . request . performEvict ( t ) ;
}
@Override
public void onComplete ( ) {
this . request . performPut ( this . cacheValue ) ;
}
}
/ * *
* Inner class to avoid a hard dependency on the Reactive Streams API at runtime .
* /
private class ReactiveCachingHandler {
public static final Object NOT_HANDLED = new Object ( ) ;
private final ReactiveAdapterRegistry registry = ReactiveAdapterRegistry . getSharedInstance ( ) ;
@Nullable
public Object executeSynchronized ( CacheOperationInvoker invoker , Method method , Cache cache , Object key ) {
ReactiveAdapter adapter = this . registry . getAdapter ( method . getReturnType ( ) ) ;
if ( adapter ! = null ) {
if ( adapter . isMultiValue ( ) ) {
// Flux or similar
return adapter . fromPublisher ( Flux . from ( Mono . fromFuture (
cache . retrieve ( key ,
( ) - > Flux . from ( adapter . toPublisher ( invokeOperation ( invoker ) ) ) . collectList ( ) . toFuture ( ) ) ) )
. flatMap ( Flux : : fromIterable ) ) ;
}
else {
// Mono or similar
return adapter . fromPublisher ( Mono . fromFuture (
cache . retrieve ( key ,
( ) - > Mono . from ( adapter . toPublisher ( invokeOperation ( invoker ) ) ) . toFuture ( ) ) ) ) ;
}
}
return NOT_HANDLED ;
}
@Nullable
public Object processCacheEvicts ( List < CacheOperationContext > contexts , @Nullable Object result ) {
ReactiveAdapter adapter = ( result ! = null ? this . registry . getAdapter ( result . getClass ( ) ) : null ) ;
if ( adapter ! = null ) {
return adapter . fromPublisher ( Mono . from ( adapter . toPublisher ( result ) )
. doOnSuccess ( value - > performCacheEvicts ( contexts , result ) ) ) ;
}
return NOT_HANDLED ;
}
@Nullable
public Object findInCaches ( CacheOperationContext context , Cache cache , Object key ) {
ReactiveAdapter adapter = this . registry . getAdapter ( context . getMethod ( ) . getReturnType ( ) ) ;
if ( adapter ! = null ) {
CompletableFuture < ? > cachedFuture = cache . retrieve ( key ) ;
if ( cachedFuture = = null ) {
return null ;
}
if ( adapter . isMultiValue ( ) ) {
return adapter . fromPublisher ( Flux . from ( Mono . fromFuture ( cachedFuture ) )
. flatMap ( v - > ( v instanceof Iterable < ? > iv ? Flux . fromIterable ( iv ) : Flux . just ( v ) ) ) ) ;
}
else {
return adapter . fromPublisher ( Mono . fromFuture ( cachedFuture ) ) ;
}
}
return NOT_HANDLED ;
}
@Nullable
public Object processPutRequest ( CachePutRequest request , @Nullable Object result ) {
ReactiveAdapter adapter = ( result ! = null ? this . registry . getAdapter ( result . getClass ( ) ) : null ) ;
if ( adapter ! = null ) {
if ( adapter . isMultiValue ( ) ) {
Flux < ? > source = Flux . from ( adapter . toPublisher ( result ) ) ;
source . subscribe ( new CachePutListSubscriber ( request ) ) ;
return adapter . fromPublisher ( source ) ;
}
else {
return adapter . fromPublisher ( Mono . from ( adapter . toPublisher ( result ) )
. doOnSuccess ( request : : performPut ) . doOnError ( request : : performEvict ) ) ;
}
}
return NOT_HANDLED ;
}
}
}
}
}