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,18 @@
  */
 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.Consumer;
 import java.util.function.Function;
 
 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 +119,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 +133,32 @@ public byte[] get(String name, byte[] key, @Nullable Duration ttl) {
 		return result;
 	}
 
+	@Override
+	public Mono<ByteBuffer> retrieve(String name, byte[] key, @Nullable Duration ttl) {
+
+		assertReactiveRedisConnectionFactory();
+
+		Assert.notNull(name, "Name must not be null");
+		Assert.notNull(key, "Key must not be null");
+
+		ByteBuffer wrappedKey = ByteBuffer.wrap(key);
+
+		Mono<ByteBuffer> result = shouldExpireWithin(ttl)
+				? executeReactively(name, connection -> connection.stringCommands().getEx(wrappedKey, Expiration.from(ttl)))
+				: executeReactively(name, connection -> connection.stringCommands().get(wrappedKey));
+
+		result = result.doOnSuccess(byteBuffer -> {
+			if (byteBuffer != null) {
+				statistics.incHits(name);
+			}
+			else {
+				statistics.incMisses(name);
+			}
+		}).doFirst(() -> statistics.incGets(name));
+
+		return result;
+	}
+
 	@Override
 	public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
 
@@ -308,6 +339,18 @@ private void executeLockFree(Consumer<RedisConnection> callback) {
 		}
 	}
 
+	private <T> T executeReactively(String name, 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 +376,21 @@ private void checkAndPotentiallyWaitUntilUnlocked(String name, RedisConnection c
 		}
 	}
 
+	private void assertReactiveRedisConnectionFactory() {
+
+		Assert.state(this.connectionFactory instanceof ReactiveRedisConnectionFactory,
+			() -> "Cache.retrieve(key) is only supported in Reactive Redis");
+	}
+
+	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,12 @@ default byte[] get(String name, byte[] key, @Nullable Duration ttl) {
 		return get(name, key);
 	}
 
+	default Mono<ByteBuffer> retrieve(String name, byte[] key) {
+		return retrieve(name, key, null);
+	}
+
+	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.
 	 *