Use error handler for reactive cache aspect
This change ensures that the cache error handler is used in case of future-based or publisher-based asynchronous caching completing with an exception. Closes gh-33073
This commit is contained in:
@@ -18,8 +18,10 @@ package org.springframework.cache.annotation;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import reactor.core.publisher.Flux;
|
||||
@@ -29,12 +31,15 @@ import org.springframework.cache.Cache;
|
||||
import org.springframework.cache.CacheManager;
|
||||
import org.springframework.cache.concurrent.ConcurrentMapCache;
|
||||
import org.springframework.cache.concurrent.ConcurrentMapCacheManager;
|
||||
import org.springframework.cache.interceptor.CacheErrorHandler;
|
||||
import org.springframework.cache.interceptor.LoggingCacheErrorHandler;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.lang.Nullable;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.AssertionsForClassTypes.catchThrowable;
|
||||
|
||||
/**
|
||||
* Tests for annotation-based caching methods that use reactive operators.
|
||||
@@ -113,6 +118,51 @@ class ReactiveCachingTests {
|
||||
ctx.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void cacheErrorHandlerWithLoggingCacheErrorHandler() {
|
||||
AnnotationConfigApplicationContext ctx =
|
||||
new AnnotationConfigApplicationContext(ExceptionCacheManager.class, ReactiveCacheableService.class, ErrorHandlerCachingConfiguration.class);
|
||||
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
|
||||
|
||||
Object key = new Object();
|
||||
Long r1 = service.cacheFuture(key).join();
|
||||
|
||||
assertThat(r1).isNotNull();
|
||||
assertThat(r1).as("cacheFuture").isEqualTo(0L);
|
||||
|
||||
key = new Object();
|
||||
|
||||
r1 = service.cacheMono(key).block();
|
||||
|
||||
assertThat(r1).isNotNull();
|
||||
assertThat(r1).as("cacheMono").isEqualTo(1L);
|
||||
|
||||
key = new Object();
|
||||
|
||||
r1 = service.cacheFlux(key).blockFirst();
|
||||
|
||||
assertThat(r1).isNotNull();
|
||||
assertThat(r1).as("cacheFlux blockFirst").isEqualTo(2L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void cacheErrorHandlerWithSimpleCacheErrorHandler() {
|
||||
AnnotationConfigApplicationContext ctx =
|
||||
new AnnotationConfigApplicationContext(ExceptionCacheManager.class, ReactiveCacheableService.class);
|
||||
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
|
||||
|
||||
Throwable completableFuturThrowable = catchThrowable(() -> service.cacheFuture(new Object()).join());
|
||||
assertThat(completableFuturThrowable).isInstanceOf(CompletionException.class)
|
||||
.extracting(Throwable::getCause)
|
||||
.isInstanceOf(UnsupportedOperationException.class);
|
||||
|
||||
Throwable monoThrowable = catchThrowable(() -> service.cacheMono(new Object()).block());
|
||||
assertThat(monoThrowable).isInstanceOf(UnsupportedOperationException.class);
|
||||
|
||||
Throwable fluxThrowable = catchThrowable(() -> service.cacheFlux(new Object()).blockFirst());
|
||||
assertThat(fluxThrowable).isInstanceOf(UnsupportedOperationException.class);
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@ValueSource(classes = {EarlyCacheHitDeterminationConfig.class,
|
||||
EarlyCacheHitDeterminationWithoutNullValuesConfig.class,
|
||||
@@ -139,7 +189,6 @@ class ReactiveCachingTests {
|
||||
ctx.close();
|
||||
}
|
||||
|
||||
|
||||
@CacheConfig(cacheNames = "first")
|
||||
static class ReactiveCacheableService {
|
||||
|
||||
@@ -242,4 +291,41 @@ class ReactiveCachingTests {
|
||||
}
|
||||
}
|
||||
|
||||
@Configuration
|
||||
static class ErrorHandlerCachingConfiguration implements CachingConfigurer {
|
||||
|
||||
@Bean
|
||||
@Override
|
||||
public CacheErrorHandler errorHandler() {
|
||||
return new LoggingCacheErrorHandler();
|
||||
}
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableCaching
|
||||
static class ExceptionCacheManager {
|
||||
|
||||
@Bean
|
||||
CacheManager cacheManager() {
|
||||
return new ConcurrentMapCacheManager("first") {
|
||||
@Override
|
||||
protected Cache createConcurrentMapCache(String name) {
|
||||
return new ConcurrentMapCache(name, isAllowNullValues()) {
|
||||
@Override
|
||||
public CompletableFuture<?> retrieve(Object key) {
|
||||
return CompletableFuture.supplyAsync(() -> {
|
||||
throw new UnsupportedOperationException("Test exception on retrieve");
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void put(Object key, @Nullable Object value) {
|
||||
throw new UnsupportedOperationException("Test exception on put");
|
||||
}
|
||||
};
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user