Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support asynchronous Cache.retrieve(…) in RedisCache.

Browse Source 
Closes #2650
Original pull request: #2717
This commit is contained in:
John Blum
2023-09-22 17:20:20 -07:00
committed by Mark Paluch
parent 9bae67eae2
commit a72c4268a3
7 changed files with 770 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java vendored
View File

@@ -15,20 +15,29 @@
*/
package org.springframework.data.redis.cache;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
import org.springframework.dao.PessimisticLockingFailureException;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.data.redis.util.ByteUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* {@link RedisCacheWriter} implementation capable of reading/writing binary data from/to Redis in {@literal standalone}
* and {@literal cluster} environments, and uses a given {@link RedisConnectionFactory} to obtain the actual
@@ -114,8 +123,8 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
Assert.notNull(key, "Key must not be null");
byte[] result = shouldExpireWithin(ttl)
? execute(name, connection -> connection.stringCommands().getEx(key, Expiration.from(ttl)))
: execute(name, connection -> connection.stringCommands().get(key));
? execute(name, connection -> connection.stringCommands().getEx(key, Expiration.from(ttl)))
: execute(name, connection -> connection.stringCommands().get(key));
statistics.incGets(name);
@@ -128,6 +137,81 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
return result;
}
@Override
public boolean isRetrieveSupported() {
return isReactive();
}
@Override
public CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl) {
Assert.notNull(name, "Name must not be null");
Assert.notNull(key, "Key must not be null");
CompletableFuture<byte[]> result = nonBlockingRetrieveFunction(name).apply(key, ttl);
result = result.thenApply(cachedValue -> {
statistics.incGets(name);
if (cachedValue != null) {
statistics.incHits(name);
} else {
statistics.incMisses(name);
}
return cachedValue;
});
return result;
}
private BiFunction<byte[], Duration, CompletableFuture<byte[]>> nonBlockingRetrieveFunction(String cacheName) {
return isReactive() ? reactiveRetrieveFunction(cacheName) : asyncRetrieveFunction(cacheName);
}
// TODO: Possibly remove if we rely on the default Cache.retrieve(..) behavior
// after assessing RedisCacheWriter.isRetrieveSupported().
// Function applied for Cache.retrieve(key) when a non-reactive Redis driver is used, such as Jedis.
private BiFunction<byte[], Duration, CompletableFuture<byte[]>> asyncRetrieveFunction(String cacheName) {
return (key, ttl) -> {
Supplier<byte[]> getKey = () -> execute(cacheName, connection -> connection.stringCommands().get(key));
Supplier<byte[]> getKeyWithExpiration = () -> execute(cacheName, connection ->
connection.stringCommands().getEx(key, Expiration.from(ttl)));
return shouldExpireWithin(ttl)
? CompletableFuture.supplyAsync(getKeyWithExpiration)
: CompletableFuture.supplyAsync(getKey);
};
}
// Function applied for Cache.retrieve(key) when a reactive Redis driver is used, such as Lettuce.
private BiFunction<byte[], Duration, CompletableFuture<byte[]>> reactiveRetrieveFunction(String cacheName) {
return (key, ttl) -> {
ByteBuffer wrappedKey = ByteBuffer.wrap(key);
Flux<?> cacheLockCheckFlux = Flux.interval(Duration.ZERO, this.sleepTime).takeUntil(count ->
executeLockFree(connection -> !doCheckLock(cacheName, connection)));
Mono<ByteBuffer> getMono = shouldExpireWithin(ttl)
? executeReactively(connection -> connection.stringCommands().getEx(wrappedKey, Expiration.from(ttl)))
: executeReactively(connection -> connection.stringCommands().get(wrappedKey));
Mono<ByteBuffer> result = cacheLockCheckFlux.then(getMono);
@SuppressWarnings("all")
Mono<byte[]> byteArrayResult = result.map(DefaultRedisCacheWriter::nullSafeGetBytes);
return byteArrayResult.toFuture();
};
}
@Override
public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
@@ -282,32 +366,42 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
return connection.keyCommands().del(createCacheLockKey(name));
}
boolean doCheckLock(String name, RedisConnection connection) {
return isTrue(connection.keyCommands().exists(createCacheLockKey(name)));
}
/**
* @return {@literal true} if {@link RedisCacheWriter} uses locks.
*/
private boolean isLockingCacheWriter() {
return !sleepTime.isZero() && !sleepTime.isNegative();
}
private <T> T execute(String name, Function<RedisConnection, T> callback) {
try (RedisConnection connection = connectionFactory.getConnection()) {
try (RedisConnection connection = this.connectionFactory.getConnection()) {
checkAndPotentiallyWaitUntilUnlocked(name, connection);
return callback.apply(connection);
}
}
private void executeLockFree(Consumer<RedisConnection> callback) {
private <T> T executeLockFree(Function<RedisConnection, T> callback) {
try (RedisConnection connection = connectionFactory.getConnection()) {
callback.accept(connection);
try (RedisConnection connection = this.connectionFactory.getConnection()) {
return callback.apply(connection);
}
}
private <T> T executeReactively(Function<ReactiveRedisConnection, T> callback) {
ReactiveRedisConnection connection = getReactiveRedisConnectionFactory().getReactiveConnection();
try {
return callback.apply(connection);
}
finally {
connection.closeLater();
}
}
/**
* Determines whether this {@link RedisCacheWriter} uses locks during caching operations.
*
* @return {@literal true} if {@link RedisCacheWriter} uses locks.
*/
private boolean isLockingCacheWriter() {
return !this.sleepTime.isZero() && !this.sleepTime.isNegative();
}
private void checkAndPotentiallyWaitUntilUnlocked(String name, RedisConnection connection) {
if (!isLockingCacheWriter()) {
@@ -318,29 +412,46 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
try {
while (doCheckLock(name, connection)) {
Thread.sleep(sleepTime.toMillis());
Thread.sleep(this.sleepTime.toMillis());
}
} catch (InterruptedException cause) {
// Re-interrupt current thread, to allow other participants to react.
// Re-interrupt current Thread to allow other participants to react.
Thread.currentThread().interrupt();
String message = String.format("Interrupted while waiting to unlock cache %s", name);
throw new PessimisticLockingFailureException(message, cause);
} finally {
statistics.incLockTime(name, System.nanoTime() - lockWaitTimeNs);
this.statistics.incLockTime(name, System.nanoTime() - lockWaitTimeNs);
}
}
boolean doCheckLock(String name, RedisConnection connection) {
return isTrue(connection.keyCommands().exists(createCacheLockKey(name)));
}
private boolean isReactive() {
return this.connectionFactory instanceof ReactiveRedisConnectionFactory;
}
private ReactiveRedisConnectionFactory getReactiveRedisConnectionFactory() {
return (ReactiveRedisConnectionFactory) this.connectionFactory;
}
private static byte[] createCacheLockKey(String name) {
return (name + "~lock").getBytes(StandardCharsets.UTF_8);
}
private boolean isTrue(@Nullable Boolean value) {
private static boolean isTrue(@Nullable Boolean value) {
return Boolean.TRUE.equals(value);
}
@Nullable
private static byte[] nullSafeGetBytes(@Nullable ByteBuffer value) {
return value != null ? ByteUtils.getBytes(value) : null;
}
private static boolean shouldExpireWithin(@Nullable Duration ttl) {
return ttl != null && !ttl.isZero() && !ttl.isNegative();
}

28
src/main/java/org/springframework/data/redis/cache/RedisCache.java vendored
View File

@@ -46,7 +46,7 @@ import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
/**
* {@link org.springframework.cache.Cache} implementation using for Redis as the underlying store for cache data.
* {@link AbstractValueAdaptingCache Cache} implementation using Redis as the underlying store for cache data.
* <p>
* Use {@link RedisCacheManager} to create {@link RedisCache} instances.
*
@@ -61,7 +61,7 @@ import org.springframework.util.ReflectionUtils;
@SuppressWarnings("unused")
public class RedisCache extends AbstractValueAdaptingCache {
private static final byte[] BINARY_NULL_VALUE = RedisSerializer.java().serialize(NullValue.INSTANCE);
static final byte[] BINARY_NULL_VALUE = RedisSerializer.java().serialize(NullValue.INSTANCE);
private final Lock lock = new ReentrantLock();
@@ -293,14 +293,38 @@ public class RedisCache extends AbstractValueAdaptingCache {
@Override
public CompletableFuture<?> retrieve(Object key) {
if (getCacheWriter().isRetrieveSupported()) {
return retrieveValue(key).thenApply(this::nullSafeDeserializedStoreValue);
}
return super.retrieve(key);
}
@Override
@SuppressWarnings("unchecked")
public <T> CompletableFuture<T> retrieve(Object key, Supplier<CompletableFuture<T>> valueLoader) {
if (getCacheWriter().isRetrieveSupported()) {
return retrieveValue(key)
.thenApply(this::nullSafeDeserializedStoreValue)
.thenCompose(cachedValue -> cachedValue != null
? CompletableFuture.completedFuture((T) cachedValue)
: valueLoader.get());
}
return super.retrieve(key, valueLoader);
}
CompletableFuture<byte[]> retrieveValue(Object key) {
return getCacheWriter().retrieve(getName(), createAndConvertCacheKey(key));
}
@Nullable
Object nullSafeDeserializedStoreValue(@Nullable byte[] value) {
return value != null ? fromStoreValue(deserializeCacheValue(value)) : null;
}
/**
* Serialize the given {@link String cache key}.
*

47
src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java vendored
View File

@@ -16,6 +16,8 @@
package org.springframework.data.redis.cache;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.lang.Nullable;
@@ -135,6 +137,51 @@ public interface RedisCacheWriter extends CacheStatisticsProvider {
return get(name, key);
}
/**
* Determines whether the asynchronous {@link #retrieve(String, byte[])}
* and {@link #retrieve(String, byte[], Duration)} cache operations are supported by the implementation.
* <p>
* The main factor for whether the {@literal retrieve} operation can be supported will primarily be determined by
* the Redis driver in use at runtime.
* <p>
* Returns {@literal false} by default. This will have an effect of {@link RedisCache#retrieve(Object)}
* and {@link RedisCache#retrieve(Object, Supplier)} throwing an {@link UnsupportedOperationException}.
*
* @return {@literal true} if asynchronous {@literal retrieve} operations are supported by the implementation.
*/
default boolean isRetrieveSupported() {
return false;
}
/**
* Returns the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@link byte[] key}.
* <p>
* This operation is non-blocking.
*
* @param name {@link String} with the name of the {@link RedisCache}.
* @param key {@link byte[] key} mapped to the {@link CompletableFuture value} in the {@link RedisCache}.
* @return the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@link byte[] key}.
* @see #retrieve(String, byte[], Duration)
* @since 3.2.0
*/
default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
return retrieve(name, key, null);
}
/**
* Returns the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@link byte[] key}
* setting the {@link Duration TTL expiration} for the cache entry.
* <p>
* This operation is non-blocking.
*
* @param name {@link String} with the name of the {@link RedisCache}.
* @param key {@link byte[] key} mapped to the {@link CompletableFuture value} in the {@link RedisCache}.
* @param ttl {@link Duration} specifying the {@literal expiration timeout} for the cache entry.
* @return the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@link byte[] key}.
* @since 3.2.0
*/
CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl);
/**
* Write the given key/value pair to Redis and set the expiration time if defined.
*

241
src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterUnitTests.java vendored
View File

@@ -16,15 +16,19 @@
package org.springframework.data.redis.cache;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -32,10 +36,17 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.ReactiveStringCommands;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisKeyCommands;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.types.Expiration;
import reactor.core.publisher.Mono;
/**
* Unit tests for {@link DefaultRedisCacheWriter}
*
@@ -45,19 +56,39 @@ import org.springframework.data.redis.core.types.Expiration;
class DefaultRedisCacheWriterUnitTests {
@Mock
private RedisConnection mockConnection;
private CacheStatisticsCollector mockCacheStatisticsCollector = mock(CacheStatisticsCollector.class);
@Mock
private RedisConnection mockConnection;
@Mock(strictness = Mock.Strictness.LENIENT)
private RedisConnectionFactory mockConnectionFactory;
@Mock
private ReactiveRedisConnection mockReactiveConnection;
@Mock(strictness = Mock.Strictness.LENIENT)
private TestReactiveRedisConnectionFactory mockReactiveConnectionFactory;
@BeforeEach
void setup() {
doReturn(this.mockConnection).when(this.mockConnectionFactory).getConnection();
doReturn(this.mockConnection).when(this.mockReactiveConnectionFactory).getConnection();
doReturn(this.mockReactiveConnection).when(this.mockReactiveConnectionFactory).getReactiveConnection();
}
private RedisCacheWriter newRedisCacheWriter() {
return new DefaultRedisCacheWriter(this.mockConnectionFactory, mock(BatchStrategy.class))
.withStatisticsCollector(mock(CacheStatisticsCollector.class));
.withStatisticsCollector(this.mockCacheStatisticsCollector);
}
private RedisCacheWriter newReactiveRedisCacheWriter() {
return newReactiveRedisCacheWriter(Duration.ZERO);
}
private RedisCacheWriter newReactiveRedisCacheWriter(Duration sleepTime) {
return new DefaultRedisCacheWriter(this.mockReactiveConnectionFactory, sleepTime, mock(BatchStrategy.class))
.withStatisticsCollector(this.mockCacheStatisticsCollector);
}
@Test // GH-2351
@@ -69,15 +100,19 @@ class DefaultRedisCacheWriterUnitTests {
Duration ttl = Duration.ofSeconds(15);
Expiration expiration = Expiration.from(ttl);
doReturn(value).when(this.mockConnection).getEx(any(), any());
RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
doReturn(value).when(mockStringCommands).getEx(any(), any());
RedisCacheWriter cacheWriter = newRedisCacheWriter();
assertThat(cacheWriter.get("TestCache", key, ttl)).isEqualTo(value);
verify(this.mockConnection, times(1)).getEx(eq(key), eq(expiration));
verify(this.mockConnection, times(1)).stringCommands();
verify(mockStringCommands, times(1)).getEx(eq(key), eq(expiration));
verify(this.mockConnection).close();
verifyNoMoreInteractions(this.mockConnection);
verifyNoMoreInteractions(this.mockConnection, mockStringCommands);
}
@Test // GH-2351
@@ -86,14 +121,204 @@ class DefaultRedisCacheWriterUnitTests {
byte[] key = "TestKey".getBytes();
byte[] value = "TestValue".getBytes();
doReturn(value).when(this.mockConnection).get(any());
RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
doReturn(value).when(mockStringCommands).get(any());
RedisCacheWriter cacheWriter = newRedisCacheWriter();
assertThat(cacheWriter.get("TestCache", key, null)).isEqualTo(value);
verify(this.mockConnection, times(1)).get(eq(key));
verify(this.mockConnection, times(1)).stringCommands();
verify(mockStringCommands, times(1)).get(eq(key));
verify(this.mockConnection).close();
verifyNoMoreInteractions(this.mockConnection);
verifyNoMoreInteractions(this.mockConnection, mockStringCommands);
}
@Test // GH-2650
@SuppressWarnings("all")
void retrieveWithNoCacheName() {
byte[] key = "TestKey".getBytes();
RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
assertThatIllegalArgumentException()
.isThrownBy(() -> cacheWriter.retrieve(null, key))
.withMessage("Name must not be null")
.withNoCause();
verifyNoInteractions(this.mockReactiveConnectionFactory);
}
@Test // GH-2650
@SuppressWarnings("all")
void retrieveWithNoKey() {
RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
assertThatIllegalArgumentException()
.isThrownBy(() -> cacheWriter.retrieve("TestCacheName", null))
.withMessage("Key must not be null")
.withNoCause();
verifyNoInteractions(this.mockReactiveConnectionFactory);
}
@Test // GH-2650
void retrieveReturnsAsyncFutureWithValue() throws Exception {
byte[] key = "TestKey".getBytes();
RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
doReturn("test".getBytes()).when(mockStringCommands).get(any(byte[].class));
RedisCacheWriter cacheWriter = newRedisCacheWriter();
CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key);
assertThat(result).isNotNull();
verifyNoInteractions(this.mockCacheStatisticsCollector);
byte[] value = result.get();
assertThat(value).isNotNull();
assertThat(new String(value)).isEqualTo("test");
verify(mockStringCommands, times(1)).get(eq(key));
verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
verifyNoMoreInteractions(mockStringCommands, this.mockCacheStatisticsCollector);
}
@Test // GH-2650
void retrieveWithExpirationReturnsAsyncFutureWithValue() throws Exception {
byte[] key = "TestKey".getBytes();
Duration thirtySeconds = Duration.ofSeconds(30L);
RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
doReturn("test".getBytes()).when(mockStringCommands).getEx(any(byte[].class), any(Expiration.class));
RedisCacheWriter cacheWriter = newRedisCacheWriter();
CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key, thirtySeconds);
assertThat(result).isNotNull();
byte[] value = result.get();
assertThat(value).isNotNull();
assertThat(new String(value)).isEqualTo("test");
verify(mockStringCommands, times(1)).getEx(eq(key), eq(Expiration.from(thirtySeconds)));
verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
verifyNoMoreInteractions(mockStringCommands, this.mockCacheStatisticsCollector);
}
@Test // GH-2650
void retrieveReturnsReactiveFutureWithValue() throws Exception {
byte[] key = "TestKey".getBytes();
Duration sixtySeconds = Duration.ofMillis(60L);
RedisKeyCommands mockKeyCommands = mock(RedisKeyCommands.class);
ReactiveStringCommands mockStringCommands = mock(ReactiveStringCommands.class);
doReturn(mockKeyCommands).when(this.mockConnection).keyCommands();
doReturn(false).when(mockKeyCommands).exists(any(byte[].class));
doReturn(mockStringCommands).when(this.mockReactiveConnection).stringCommands();
doReturn(Mono.just(ByteBuffer.wrap("test".getBytes()))).when(mockStringCommands).get(any(ByteBuffer.class));
RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter(sixtySeconds);
CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key);
assertThat(result).isNotNull();
byte[] value = result.get();
assertThat(value).isNotNull();
assertThat(new String(value)).isEqualTo("test");
verify(mockKeyCommands, times(1)).exists(any(byte[].class));
verify(mockStringCommands, times(1)).get(eq(ByteBuffer.wrap(key)));
verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
verifyNoMoreInteractions(mockKeyCommands, mockStringCommands, this.mockCacheStatisticsCollector);
}
@Test // GH-2650
void retrieveReturnsReactiveFutureWithNoValue() throws Exception {
byte[] key = "TestKey".getBytes();
RedisKeyCommands mockKeyCommands = mock(RedisKeyCommands.class);
ReactiveStringCommands mockStringCommands = mock(ReactiveStringCommands.class);
doReturn(mockKeyCommands).when(this.mockConnection).keyCommands();
doReturn(false).when(mockKeyCommands).exists(any(byte[].class));
doReturn(mockStringCommands).when(this.mockReactiveConnection).stringCommands();
doReturn(Mono.empty()).when(mockStringCommands).get(any(ByteBuffer.class));
RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key);
assertThat(result).isNotNull();
byte[] value = result.get();
assertThat(value).isNull();
verify(mockKeyCommands, times(1)).exists(any(byte[].class));
verify(mockStringCommands, times(1)).get(eq(ByteBuffer.wrap(key)));
verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
verify(this.mockCacheStatisticsCollector, times(1)).incMisses(eq("TestCacheName"));
verifyNoMoreInteractions(mockKeyCommands, mockStringCommands, this.mockCacheStatisticsCollector);
}
@Test // GH-2650
void retrieveWithExpirationReturnsReactiveFutureWithValue() throws Exception {
byte[] key = "TestKey".getBytes();
Duration twoMinutes = Duration.ofMinutes(2L);
RedisKeyCommands mockKeyCommands = mock(RedisKeyCommands.class);
ReactiveStringCommands mockStringCommands = mock(ReactiveStringCommands.class);
doReturn(mockKeyCommands).when(this.mockConnection).keyCommands();
doReturn(false).when(mockKeyCommands).exists(any(byte[].class));
doReturn(mockStringCommands).when(this.mockReactiveConnection).stringCommands();
doReturn(Mono.just(ByteBuffer.wrap("test".getBytes()))).when(mockStringCommands).getEx(any(ByteBuffer.class), any());
RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key, twoMinutes);
assertThat(result).isNotNull();
byte[] value = result.get();
assertThat(value).isNotNull();
assertThat(new String(value)).isEqualTo("test");
verify(mockKeyCommands, times(1)).exists(any(byte[].class));
verify(mockStringCommands, times(1)).getEx(eq(ByteBuffer.wrap(key)), eq(Expiration.from(twoMinutes)));
verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
verifyNoMoreInteractions(mockKeyCommands, mockStringCommands, this.mockCacheStatisticsCollector);
}
interface TestReactiveRedisConnectionFactory extends ReactiveRedisConnectionFactory, RedisConnectionFactory { }
}

235
src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java vendored
View File

@@ -15,30 +15,37 @@
*/
package org.springframework.data.redis.cache;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assumptions.*;
import static org.awaitility.Awaitility.*;
import io.netty.util.concurrent.DefaultThreadFactory;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import static org.assertj.core.api.Assumptions.assumeThat;
import static org.awaitility.Awaitility.await;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.Month;
import java.time.ZoneOffset;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.IntStream;
import org.junit.jupiter.api.BeforeEach;
import org.springframework.cache.Cache.ValueWrapper;
import org.springframework.cache.interceptor.SimpleKey;
import org.springframework.cache.interceptor.SimpleKeyGenerator;
@@ -46,13 +53,18 @@ import org.springframework.cache.support.NullValue;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.test.condition.EnabledOnCommand;
import org.springframework.data.redis.test.condition.EnabledOnRedisDriver;
import org.springframework.data.redis.test.condition.RedisDriver;
import org.springframework.data.redis.test.extension.parametrized.MethodSource;
import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest;
import org.springframework.lang.Nullable;
import io.netty.util.concurrent.DefaultThreadFactory;
/**
* Tests for {@link RedisCache} with {@link DefaultRedisCacheWriter} using different {@link RedisSerializer} and
* {@link RedisConnectionFactory} pairs.
@@ -103,9 +115,7 @@ public class RedisCacheTests {
cache.put("key-1", sample);
doWithConnection(connection -> {
assertThat(connection.exists(binaryCacheKey)).isTrue();
});
doWithConnection(connection -> assertThat(connection.exists(binaryCacheKey)).isTrue());
}
@ParameterizedRedisTest // GH-2379
@@ -116,9 +126,7 @@ public class RedisCacheTests {
String keyPattern = "*" + key.substring(1);
cache.clear(keyPattern);
doWithConnection(connection -> {
assertThat(connection.exists(binaryCacheKey)).isFalse();
});
doWithConnection(connection -> assertThat(connection.exists(binaryCacheKey)).isFalse());
}
@ParameterizedRedisTest // GH-2379
@@ -129,9 +137,7 @@ public class RedisCacheTests {
String keyPattern = "*" + key.substring(1) + "tail";
cache.clear(keyPattern);
doWithConnection(connection -> {
assertThat(connection.exists(binaryCacheKey)).isTrue();
});
doWithConnection(connection -> assertThat(connection.exists(binaryCacheKey)).isTrue());
}
@ParameterizedRedisTest // DATAREDIS-481
@@ -177,9 +183,7 @@ public class RedisCacheTests {
assertThat(result).isNotNull();
assertThat(result.get()).isEqualTo(sample);
doWithConnection(connection -> {
assertThat(connection.get(binaryCacheKey)).isEqualTo(binarySample);
});
doWithConnection(connection -> assertThat(connection.get(binaryCacheKey)).isEqualTo(binarySample));
}
@ParameterizedRedisTest // DATAREDIS-481
@@ -192,17 +196,13 @@ public class RedisCacheTests {
assertThat(result).isNotNull();
assertThat(result.get()).isNull();
doWithConnection(connection -> {
assertThat(connection.get(binaryCacheKey)).isEqualTo(binaryNullValue);
});
doWithConnection(connection -> assertThat(connection.get(binaryCacheKey)).isEqualTo(binaryNullValue));
}
@ParameterizedRedisTest // DATAREDIS-481
void getShouldRetrieveEntry() {
doWithConnection(connection -> {
connection.set(binaryCacheKey, binarySample);
});
doWithConnection(connection -> connection.set(binaryCacheKey, binarySample));
ValueWrapper result = cache.get(key);
assertThat(result).isNotNull();
@@ -237,6 +237,7 @@ public class RedisCacheTests {
cache.put(key, sample);
ValueWrapper result = cache.get(key);
assertThat(result).isNotNull();
assertThat(result.get()).isEqualTo(sample);
}
@@ -249,11 +250,10 @@ public class RedisCacheTests {
@ParameterizedRedisTest // DATAREDIS-481
void getShouldReturnValueWrapperHoldingNullIfNullValueStored() {
doWithConnection(connection -> {
connection.set(binaryCacheKey, binaryNullValue);
});
doWithConnection(connection -> connection.set(binaryCacheKey, binaryNullValue));
ValueWrapper result = cache.get(key);
assertThat(result).isNotNull();
assertThat(result.get()).isEqualTo(null);
}
@@ -353,11 +353,8 @@ public class RedisCacheTests {
cacheWithCustomPrefix.put("key-1", sample);
doWithConnection(connection -> {
assertThat(connection.stringCommands().get("_cache_key-1".getBytes(StandardCharsets.UTF_8)))
.isEqualTo(binarySample);
});
doWithConnection(connection -> assertThat(connection.stringCommands()
.get("_cache_key-1".getBytes(StandardCharsets.UTF_8))).isEqualTo(binarySample));
}
@ParameterizedRedisTest // DATAREDIS-1041
@@ -369,11 +366,8 @@ public class RedisCacheTests {
cacheWithCustomPrefix.put("key-1", sample);
doWithConnection(connection -> {
assertThat(connection.stringCommands().get("redis::cache::key-1".getBytes(StandardCharsets.UTF_8)))
.isEqualTo(binarySample);
});
doWithConnection(connection -> assertThat(connection.stringCommands()
.get("redis::cache::key-1".getBytes(StandardCharsets.UTF_8))).isEqualTo(binarySample));
}
@ParameterizedRedisTest // DATAREDIS-715
@@ -400,6 +394,7 @@ public class RedisCacheTests {
ValueWrapper target = cache
.get(SimpleKeyGenerator.generateKey(Collections.singletonList("my-cache-key-in-a-list")));
assertThat(target.get()).isEqualTo(sample);
}
@@ -410,6 +405,7 @@ public class RedisCacheTests {
cache.put(key, sample);
ValueWrapper target = cache.get(SimpleKeyGenerator.generateKey("my-cache-key-in-an-array"));
assertThat(target.get()).isEqualTo(sample);
}
@@ -422,6 +418,7 @@ public class RedisCacheTests {
ValueWrapper target = cache.get(SimpleKeyGenerator
.generateKey(Collections.singletonList(new ComplexKey(sample.getFirstname(), sample.getBirthdate()))));
assertThat(target.get()).isEqualTo(sample);
}
@@ -434,6 +431,7 @@ public class RedisCacheTests {
ValueWrapper target = cache.get(SimpleKeyGenerator
.generateKey(Collections.singletonMap("map-key", new ComplexKey(sample.getFirstname(), sample.getBirthdate()))));
assertThat(target.get()).isEqualTo(sample);
}
@@ -442,6 +440,7 @@ public class RedisCacheTests {
Object key = SimpleKeyGenerator
.generateKey(Collections.singletonList(new InvalidKey(sample.getFirstname(), sample.getBirthdate())));
assertThatIllegalStateException().isThrownBy(() -> cache.put(key, sample));
}
@@ -458,11 +457,6 @@ public class RedisCacheTests {
cache = new RedisCache("foo", new RedisCacheWriter() {
@Override
public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
storage.set(value);
}
@Override
public byte[] get(String name, byte[] key) {
return get(name, key, null);
@@ -481,6 +475,17 @@ public class RedisCacheTests {
return storage.get();
}
@Override
public CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl) {
byte[] value = get(name, key);
return CompletableFuture.completedFuture(value);
}
@Override
public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
storage.set(value);
}
@Override
public byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
return new byte[0];
@@ -567,13 +572,129 @@ public class RedisCacheTests {
assertThat(cache.get(this.cacheKey, Person.class)).isNull();
}
@Nullable
private Object unwrap(@Nullable Object value) {
return value instanceof ValueWrapper wrapper ? wrapper.get() : value;
@ParameterizedRedisTest
void retrieveCacheValueUsingJedis() {
// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.JEDIS) does not work!
assumeThat(this.connectionFactory instanceof JedisConnectionFactory).isTrue();
assertThatExceptionOfType(UnsupportedOperationException.class)
.isThrownBy(() -> this.cache.retrieve(this.binaryCacheKey))
.withMessageContaining(RedisCache.class.getName())
.withNoCause();
}
private RedisCacheWriter usingRedisCacheWriter() {
return RedisCacheWriter.nonLockingRedisCacheWriter(this.connectionFactory);
@ParameterizedRedisTest
void retrieveCacheValueWithLoaderUsingJedis() {
// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.JEDIS) does not work!
assumeThat(this.connectionFactory instanceof JedisConnectionFactory).isTrue();
assertThatExceptionOfType(UnsupportedOperationException.class)
.isThrownBy(() -> this.cache.retrieve(this.binaryCacheKey, () -> CompletableFuture.completedFuture("TEST")))
.withMessageContaining(RedisCache.class.getName())
.withNoCause();
}
@ParameterizedRedisTest // GH-2650
@SuppressWarnings("unchecked")
void retrieveReturnsCachedValue() throws Exception {
// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.LETTUCE) does not work!
assumeThat(this.connectionFactory instanceof LettuceConnectionFactory).isTrue();
doWithConnection(connection -> connection.stringCommands().set(this.binaryCacheKey, this.binarySample));
RedisCache cache = new RedisCache("cache", usingLockingRedisCacheWriter(), usingRedisCacheConfiguration());
CompletableFuture<Person> value = (CompletableFuture<Person>) cache.retrieve(this.key);
assertThat(value).isNotNull();
assertThat(value.get()).isEqualTo(this.sample);
assertThat(value).isDone();
}
@ParameterizedRedisTest // GH-2650
@SuppressWarnings("unchecked")
void retrieveReturnsCachedValueWhenLockIsReleased() throws Exception {
// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.LETTUCE) does not work!
assumeThat(this.connectionFactory instanceof LettuceConnectionFactory).isTrue();
String mockValue = "MockValue";
String testValue = "TestValue";
byte[] binaryCacheValue = this.serializer.serialize(testValue);
doWithConnection(connection -> connection.stringCommands().set(this.binaryCacheKey, binaryCacheValue));
RedisCache cache = new RedisCache("cache", usingLockingRedisCacheWriter(Duration.ofMillis(5L)),
usingRedisCacheConfiguration());
RedisCacheWriter cacheWriter = cache.getCacheWriter();
assertThat(cacheWriter).isInstanceOf(DefaultRedisCacheWriter.class);
((DefaultRedisCacheWriter) cacheWriter).lock("cache");
CompletableFuture<String> value = (CompletableFuture<String>) cache.retrieve(this.key);
assertThat(value).isNotNull();
assertThat(value.getNow(mockValue)).isEqualTo(mockValue);
assertThat(value).isNotDone();
((DefaultRedisCacheWriter) cacheWriter).unlock("cache");
assertThat(value.get(15L, TimeUnit.MILLISECONDS)).isEqualTo(testValue);
assertThat(value).isDone();
}
@ParameterizedRedisTest // GH-2650
void retrieveReturnsLoadedValue() throws Exception {
// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.LETTUCE) does not work!
assumeThat(this.connectionFactory instanceof LettuceConnectionFactory).isTrue();
RedisCache cache = new RedisCache("cache", usingLockingRedisCacheWriter(), usingRedisCacheConfiguration());
AtomicBoolean loaded = new AtomicBoolean(false);
Date birthdate = Date.from(LocalDateTime.of(2023, Month.SEPTEMBER, 22, 17, 3)
.toInstant(ZoneOffset.UTC));
Person jon = new Person("Jon", birthdate);
CompletableFuture<Person> valueLoader = CompletableFuture.completedFuture(jon);
Supplier<CompletableFuture<Person>> valueLoaderSupplier = () -> {
loaded.set(true);
return valueLoader;
};
CompletableFuture<Person> value = cache.retrieve(this.key, valueLoaderSupplier);
assertThat(value).isNotNull();
assertThat(loaded.get()).isFalse();
assertThat(value.get()).isEqualTo(jon);
assertThat(loaded.get()).isTrue();
assertThat(value).isDone();
}
@ParameterizedRedisTest // GH-2650
void retrieveReturnsNull() throws Exception {
// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.LETTUCE) does not work!
assumeThat(this.connectionFactory instanceof LettuceConnectionFactory).isTrue();
doWithConnection(connection -> connection.stringCommands().set(this.binaryCacheKey, this.binaryNullValue));
RedisCache cache = new RedisCache("cache", usingLockingRedisCacheWriter(), usingRedisCacheConfiguration());
CompletableFuture<?> value = cache.retrieve(this.key);
assertThat(value).isNotNull();
assertThat(value.get()).isNull();
assertThat(value).isDone();
}
private RedisCacheConfiguration usingRedisCacheConfiguration() {
@@ -587,6 +708,28 @@ public class RedisCacheTests {
.serializeValuesWith(SerializationPair.fromSerializer(this.serializer)));
}
private RedisCacheWriter usingRedisCacheWriter() {
return usingNonLockingRedisCacheWriter();
}
private RedisCacheWriter usingLockingRedisCacheWriter() {
return RedisCacheWriter.lockingRedisCacheWriter(this.connectionFactory);
}
private RedisCacheWriter usingLockingRedisCacheWriter(Duration sleepTime) {
return RedisCacheWriter.lockingRedisCacheWriter(this.connectionFactory, sleepTime,
RedisCacheWriter.TtlFunction.persistent(), BatchStrategies.keys());
}
private RedisCacheWriter usingNonLockingRedisCacheWriter() {
return RedisCacheWriter.nonLockingRedisCacheWriter(this.connectionFactory);
}
@Nullable
private Object unwrap(@Nullable Object value) {
return value instanceof ValueWrapper wrapper ? wrapper.get() : value;
}
private Function<RedisCacheConfiguration, RedisCacheConfiguration> withTtiExpiration() {
Function<RedisCacheConfiguration, RedisCacheConfiguration> entryTtlFunction =

121
src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java vendored Normal file
View File

@@ -0,0 +1,121 @@
/*
* Copyright 2017-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.data.redis.cache;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import java.util.concurrent.CompletableFuture;
import org.junit.jupiter.api.Test;
import org.springframework.cache.support.NullValue;
import org.springframework.data.redis.util.ByteUtils;
/**
* Unit tests for {@link RedisCache}.
*
* @author John Blum
*/
class RedisCacheUnitTests {
@Test // GH-2650
void cacheRetrieveValueCallsCacheWriterRetrieveCorrectly() throws Exception {
RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
doReturn(CompletableFuture.completedFuture("TEST".getBytes()))
.when(mockCacheWriter).retrieve(anyString(), any(byte[].class));
RedisCache cache = new RedisCache("TestCache", mockCacheWriter,
RedisCacheConfiguration.defaultCacheConfig());
CompletableFuture<byte[]> value = cache.retrieveValue("TestKey");
assertThat(value).isNotNull();
assertThat(new String(value.get())).isEqualTo("TEST");
verify(mockCacheWriter, times(1)).retrieve(eq("TestCache"), isA(byte[].class));
verifyNoMoreInteractions(mockCacheWriter);
}
@Test // GH-2650
void nullSafeDeserializedStoreValueWithNullValueIsNullSafe() {
RedisCacheConfiguration cacheConfiguration = RedisCacheConfiguration.defaultCacheConfig();
RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
assertThat(cache.nullSafeDeserializedStoreValue(null)).isNull();
verifyNoInteractions(mockCacheWriter);
}
@Test // GH-2650
void nullSafeDeserializedStoreValueWithBinaryNullValueAllowingNullValues() {
RedisCacheConfiguration cacheConfiguration = RedisCacheConfiguration.defaultCacheConfig();
RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
assertThat(cacheConfiguration.getAllowCacheNullValues()).isTrue();
assertThat(cache.nullSafeDeserializedStoreValue(RedisCache.BINARY_NULL_VALUE)).isNull();
verifyNoInteractions(mockCacheWriter);
}
@Test // GH-2650
void nullSafeDeserializedStoreValueWithBinaryNullValueDisablingNullValues() {
RedisCacheConfiguration cacheConfiguration =
RedisCacheConfiguration.defaultCacheConfig().disableCachingNullValues();
RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
assertThat(cacheConfiguration.getAllowCacheNullValues()).isFalse();
assertThat(cache.nullSafeDeserializedStoreValue(RedisCache.BINARY_NULL_VALUE)).isEqualTo(NullValue.INSTANCE);
verifyNoInteractions(mockCacheWriter);
}
@Test // GH-2650
void nullSafeDeserializedStoreValueWithNonNullValue() {
RedisCacheConfiguration cacheConfiguration = RedisCacheConfiguration.defaultCacheConfig();
byte[] serializedValue = ByteUtils.getBytes(cacheConfiguration.getValueSerializationPair()
.write("TestValue"));
RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
assertThat(cache.nullSafeDeserializedStoreValue(serializedValue)).isEqualTo("TestValue");
verifyNoInteractions(mockCacheWriter);
}
}

21
src/test/java/org/springframework/data/redis/cache/RedisCacheWriterUnitTests.java vendored
View File

@@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.doCallRealMethod;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -27,6 +28,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import org.junit.jupiter.api.Test;
@@ -56,4 +58,23 @@ class RedisCacheWriterUnitTests {
verify(cacheWriter, times(1)).get(eq("TestCacheName"), eq(key));
verifyNoMoreInteractions(cacheWriter);
}
@Test // GH-2650
void defaultRetrieveWithNameAndKeyCallsRetrieveWithNameKeyAndTtl() throws Exception {
byte[] key = "TestKey".getBytes();
byte[] value = "TestValue".getBytes();
RedisCacheWriter cacheWriter = mock(RedisCacheWriter.class);
doCallRealMethod().when(cacheWriter).retrieve(anyString(), any());
doReturn(CompletableFuture.completedFuture(value)).when(cacheWriter).retrieve(anyString(), any(), any());
assertThat(cacheWriter.retrieve("TestCacheName", key).thenApply(String::new).get())
.isEqualTo("TestValue");
verify(cacheWriter, times(1)).retrieve(eq("TestCacheName"), eq(key));
verify(cacheWriter, times(1)).retrieve(eq("TestCacheName"), eq(key), isNull());
verifyNoMoreInteractions(cacheWriter);
}
}