Implement retrieve(key) and retrieve(key, :Supplier<CompletableFuture<T>>) operations in RedisCache.

Closes spring-projects#2650

Author: jxblum

src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java

@@ -15,13 +15,20 @@
  */
 package org.springframework.data.redis.cache;
 
+import reactor.core.publisher.Mono;
+
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
 import java.util.function.Consumer;
 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;
@@ -114,8 +121,8 @@ public byte[] get(String name, byte[] key, @Nullable Duration ttl) {
 		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 +135,66 @@ public byte[] get(String name, byte[] key, @Nullable Duration ttl) {
 		return result;
 	}
 
+	@Override
+	public Mono<ByteBuffer> retrieve(String name, byte[] key, @Nullable Duration ttl) {
+
+		Assert.notNull(name, "Name must not be null");
+		Assert.notNull(key, "Key must not be null");
+
+		Mono<ByteBuffer> result = nonBlockingExecutionStrategy(name).apply(key, ttl);
+
+		result = result.doOnSuccess(byteBuffer -> {
+			if (byteBuffer != null) {
+				statistics.incHits(name);
+			}
+			else {
+				statistics.incMisses(name);
+			}
+		}).doFirst(() -> statistics.incGets(name));
+
+		return result;
+	}
+
+	private BiFunction<byte[], Duration, Mono<ByteBuffer>> nonBlockingExecutionStrategy(String cacheName) {
+
+		boolean isReactive = this.connectionFactory instanceof ReactiveRedisConnectionFactory;
+
+		// Execution strategy that will be applied when Lettuce is used
+		if (isReactive) {
+
+			return (key, ttl) -> {
+
+				ByteBuffer wrappedKey = ByteBuffer.wrap(key);
+
+				Mono<ByteBuffer> result = shouldExpireWithin(ttl)
+						? executeReactively(connection -> connection.stringCommands().getEx(wrappedKey, Expiration.from(ttl)))
+						: executeReactively(connection -> connection.stringCommands().get(wrappedKey));
+
+				result.doFirst(() -> executeLockFree(connection ->
+						checkAndPotentiallyWaitUntilUnlocked(cacheName, connection)));
+
+				return result;
+			};
+		}
+
+		// Execution strategy that will be applied when Jedis is used
+		return (key, ttl) -> {
+
+			Supplier<ByteBuffer> getKey = () -> execute(cacheName, connection ->
+					nullSafeByteBufferWrap(connection.stringCommands().get(key)));
+
+			Supplier<ByteBuffer> getKeyWithExpiration = () -> execute(cacheName, connection ->
+					nullSafeByteBufferWrap(connection.stringCommands().getEx(key, Expiration.from(ttl))));
+
+			return shouldExpireWithin(ttl) ? Mono.fromSupplier(getKeyWithExpiration) : Mono.fromSupplier(getKey);
+		};
+	}
+
+	@Nullable
+	private ByteBuffer nullSafeByteBufferWrap(@Nullable byte[] value) {
+		return value != null ? ByteBuffer.wrap(value) : null;
+	}
+
 	@Override
 	public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
 
@@ -308,6 +375,18 @@ private void executeLockFree(Consumer<RedisConnection> callback) {
 		}
 	}
 
+	private <T> T executeReactively(Function<ReactiveRedisConnection, T> callback) {
+
+		ReactiveRedisConnection connection = getReactiveRedisConnectionFactory().getReactiveConnection();
+
+		try {
+			return callback.apply(connection);
+		}
+		finally {
+			connection.closeLater();
+		}
+	}
+
 	private void checkAndPotentiallyWaitUntilUnlocked(String name, RedisConnection connection) {
 
 		if (!isLockingCacheWriter()) {
@@ -333,11 +412,15 @@ private void checkAndPotentiallyWaitUntilUnlocked(String name, RedisConnection c
 		}
 	}
 
+	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);
 	}
 

src/main/java/org/springframework/data/redis/cache/RedisCache.java

@@ -45,6 +45,8 @@
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
 
+import reactor.core.publisher.Mono;
+
 /**
  * {@link org.springframework.cache.Cache} implementation using for Redis as the underlying store for cache data.
  * <p>
@@ -293,12 +295,21 @@ protected Object preProcessCacheValue(@Nullable Object value) {
 
 	@Override
 	public CompletableFuture<?> retrieve(Object key) {
-		return super.retrieve(key);
+		return retrieveMono(key).toFuture();
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public <T> CompletableFuture<T> retrieve(Object key, Supplier<CompletableFuture<T>> valueLoader) {
-		return super.retrieve(key, valueLoader);
+
+		return retrieveMono(key)
+				.map(value -> (T) deserializeCacheValue(ByteUtils.getBytes(value)))
+				.or(Mono.fromCompletionStage(valueLoader))
+				.toFuture();
+	}
+
+	private Mono<ByteBuffer> retrieveMono(Object key) {
+		return getCacheWriter().retrieve(getName(), createAndConvertCacheKey(key));
 	}
 
 	/**

src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java

@@ -15,12 +15,15 @@
  */
 package org.springframework.data.redis.cache;
 
+import java.nio.ByteBuffer;
 import java.time.Duration;
 
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Mono;
+
 /**
  * {@link RedisCacheWriter} provides low-level access to Redis commands ({@code SET, SETNX, GET, EXPIRE,...}) used for
  * caching.
@@ -135,6 +138,40 @@ default byte[] get(String name, byte[] key, @Nullable Duration ttl) {
 		return get(name, key);
 	}
 
+	/**
+	 * Returns the {@link Mono value} to which the {@link RedisCache} maps the given {@link byte[] key}.
+	 * <p>
+	 * This operation does not block.
+	 *
+	 * @param name {@link String} containing the name of the {@link RedisCache}.
+	 * @param key {@link byte[] key} mapped to the {@link Mono value} in the {@link RedisCache}.
+	 * @return the {@link Mono value} to which the {@link RedisCache} maps the given {@link byte[] key}.
+	 * @throws IllegalStateException if the Redis connection factory is not reactive.
+	 * @see reactor.core.publisher.Mono
+	 * @see java.nio.ByteBuffer
+	 * @since 3.2.0
+	 */
+	default Mono<ByteBuffer> retrieve(String name, byte[] key) {
+		return retrieve(name, key, null);
+	}
+
+	/**
+	 * Returns the {@link Mono 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 does not block.
+	 *
+	 * @param name {@link String} containing the name of the {@link RedisCache}.
+	 * @param key {@link byte[] key} mapped to the {@link Mono value} in the {@link RedisCache}.
+	 * @param ttl {@link Duration} specifying the {@literal expiration timeout} for the cache entry.
+	 * @return the {@link Mono value} to which the {@link RedisCache} maps the given {@link byte[] key}.
+	 * @throws IllegalStateException if the Redis connection factory is not reactive.
+	 * @see reactor.core.publisher.Mono
+	 * @see java.nio.ByteBuffer
+	 * @since 3.2.0
+	 */
+	Mono<ByteBuffer> retrieve(String name, byte[] key, @Nullable Duration ttl);
+
 	/**
 	 * Write the given key/value pair to Redis and set the expiration time if defined.
 	 *