diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java index a534e4539..ab7bc1103 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java @@ -58,6 +58,7 @@ import org.springframework.util.ClassUtils; * * @author Mark Paluch * @author Christoph Strobl + * @author Petromir Dzhunev * @since 2.0 * @param the Redis key type against which the template works (usually a String) * @param the Redis value type against which the template works @@ -192,7 +193,14 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations { + Mono connection = getConnection(); + + if (!exposeConnection) { + connection = connection.map(this::createRedisConnectionProxy); + } + + return Flux.usingWhen(connection, conn -> { + Publisher result = action.doInRedis(conn); return postProcessResult(result, conn, false); @@ -201,18 +209,16 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations getConnection(boolean exposeConnection) { - ReactiveRedisConnectionFactory factory = getConnectionFactory(); - ReactiveRedisConnection conn = factory.getReactiveConnection(); - ReactiveRedisConnection connToUse = preProcessConnection(conn, false); + protected Mono getConnection() { - return Mono.just(exposeConnection ? connToUse : createRedisConnectionProxy(connToUse)); + ReactiveRedisConnectionFactory factory = getConnectionFactory(); + + return Mono.fromSupplier(() -> preProcessConnection(factory.getReactiveConnection(), false)); } /*