From 87bbee4310f2ff5e721a14df6339c26169092d94 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 5 Feb 2020 15:12:19 +0100 Subject: [PATCH] =?UTF-8?q?DATAREDIS-1095=20-=20Use=20Flux.usingWhen(?= =?UTF-8?q?=E2=80=A6)=20in=20DefaultReactiveScriptExecutor.execute(?= =?UTF-8?q?=E2=80=A6).?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We now consistently use Flux.usingWhen(…) to obtain and release Redis connections without blocking the calling thread. Original Pull Request: #509 --- .../core/script/DefaultReactiveScriptExecutor.java | 11 +++-------- .../DefaultReactiveScriptExecutorUnitTests.java | 10 ++++++++-- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutor.java b/src/main/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutor.java index 028040dad..6a37c2ffe 100644 --- a/src/main/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutor.java +++ b/src/main/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutor.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.core.script; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import java.nio.ByteBuffer; import java.util.List; @@ -161,15 +162,9 @@ public class DefaultReactiveScriptExecutor implements ReactiveScriptExecutor< Assert.notNull(action, "Callback object must not be null"); ReactiveRedisConnectionFactory factory = getConnectionFactory(); - ReactiveRedisConnection conn = factory.getReactiveConnection(); - try { - return Flux.defer(() -> action.doInRedis(conn)).doFinally(signal -> conn.close()); - } catch (RuntimeException e) { - - conn.close(); - throw e; - } + return Flux.usingWhen(Mono.fromSupplier(factory::getReactiveConnection), action::doInRedis, + ReactiveRedisConnection::closeLater); } public ReactiveRedisConnectionFactory getConnectionFactory() { diff --git a/src/test/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutorUnitTests.java b/src/test/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutorUnitTests.java index 8bce062fb..11b7d2ffa 100644 --- a/src/test/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutorUnitTests.java +++ b/src/test/java/org/springframework/data/redis/core/script/DefaultReactiveScriptExecutorUnitTests.java @@ -18,6 +18,7 @@ package org.springframework.data.redis.core.script; import static org.mockito.Mockito.*; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.nio.ByteBuffer; @@ -37,6 +38,8 @@ import org.springframework.data.redis.connection.ReturnType; import org.springframework.data.redis.serializer.RedisSerializationContext; /** + * Unit tests for {@link DefaultReactiveScriptExecutor}. + * * @author Mark Paluch * @author Christoph Strobl */ @@ -56,6 +59,7 @@ public class DefaultReactiveScriptExecutorUnitTests { when(connectionFactoryMock.getReactiveConnection()).thenReturn(connectionMock); when(connectionMock.scriptingCommands()).thenReturn(scriptingCommandsMock); + when(connectionMock.closeLater()).thenReturn(Mono.empty()); executor = new DefaultReactiveScriptExecutor<>(connectionFactoryMock, RedisSerializationContext.string()); } @@ -108,7 +112,8 @@ public class DefaultReactiveScriptExecutorUnitTests { StepVerifier.create(execute).expectNext("FOO").verifyComplete(); - verify(connectionMock).close(); + verify(connectionMock).closeLater(); + verify(connectionMock, never()).close(); } @Test // DATAREDIS-683 @@ -119,7 +124,8 @@ public class DefaultReactiveScriptExecutorUnitTests { StepVerifier.create(executor.execute(SCRIPT)).expectError().verify(); - verify(connectionMock).close(); + verify(connectionMock).closeLater(); + verify(connectionMock, never()).close(); } @Test // DATAREDIS-683