diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java index c10a33e3f..8c8eff434 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java @@ -152,6 +152,28 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau this.disposeClusterCommandExecutorOnClose = false; } + /** + * Creates new {@link LettuceClusterConnection} given a shared {@link StatefulRedisClusterConnection} + * and{@link LettuceConnectionProvider} running commands across the cluster via given {@link ClusterCommandExecutor}. + * + * @param sharedConnection must not be {@literal null}. + * @param connectionProvider must not be {@literal null}. + * @param executor must not be {@literal null}. + * @param timeout must not be {@literal null}. + * @since 2.1 + */ + public LettuceClusterConnection(StatefulRedisClusterConnection sharedConnection, + LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor, Duration timeout) { + + super(sharedConnection, connectionProvider, timeout.toMillis(), 0); + + Assert.notNull(executor, "ClusterCommandExecutor must not be null."); + + this.topologyProvider = new LettuceClusterTopologyProvider(getClient()); + this.clusterCommandExecutor = executor; + this.disposeClusterCommandExecutorOnClose = false; + } + /** * @return access to {@link RedisClusterClient} for non-connection access. */ @@ -319,7 +341,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau clusterCommandExecutor.executeCommandOnSingleNode( (LettuceClusterCommandCallback) client -> client.clusterAddSlots(slots), node); - } /* diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index c09ff5c06..751c0fc48 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -243,6 +243,19 @@ public class LettuceConnection extends AbstractRedisConnection { */ public LettuceConnection(StatefulRedisConnection sharedConnection, LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) { + this((StatefulConnection) sharedConnection, connectionProvider, timeout, defaultDbIndex); + } + + /** + * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be + * used for transactions or blocking operations. + * @param connectionProvider connection provider to obtain and release native connections. + * @param timeout The connection timeout (in milliseconds) + * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. + * @since 2.1 + */ + LettuceConnection(StatefulConnection sharedConnection, LettuceConnectionProvider connectionProvider, + long timeout, int defaultDbIndex) { Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null."); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index 2eb685bbc..12e806cab 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -345,7 +345,11 @@ public class LettuceConnectionFactory throw new InvalidDataAccessApiUsageException("Cluster is not configured!"); } - return new LettuceClusterConnection(connectionProvider, clusterCommandExecutor, + return getShareNativeConnection() + ? new LettuceClusterConnection( + (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection(), + connectionProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout()) + : new LettuceClusterConnection(connectionProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout()); } @@ -921,7 +925,7 @@ public class LettuceConnectionFactory return new StandaloneConnectionProvider((RedisClient) client, codec, readFrom); } - private AbstractRedisClient createClient() { + protected AbstractRedisClient createClient() { if (isRedisSentinelAware()) { @@ -1065,15 +1069,10 @@ public class LettuceConnectionFactory * * @return the connection. */ - @Nullable private StatefulConnection getNativeConnection() { try { - if (isClusterAware() && !shareNativeClusterConnection) { - return null; - } - StatefulConnection connection = connectionProvider.getConnection(StatefulConnection.class); if (connection instanceof StatefulRedisConnection && getDatabase() > 0) { ((StatefulRedisConnection) connection).sync().select(getDatabase()); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java index 0f33ed07a..bd797e25f 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java @@ -24,13 +24,19 @@ import static org.springframework.data.redis.connection.ClusterTestVariables.*; import static org.springframework.data.redis.test.util.MockitoUtils.*; import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands; +import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; import io.lettuce.core.cluster.api.sync.RedisClusterCommands; import io.lettuce.core.cluster.models.partitions.Partitions; import io.lettuce.core.cluster.models.partitions.RedisClusterNode.NodeFlag; import java.io.IOException; +import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -62,7 +68,10 @@ public class LettuceClusterConnectionUnitTests { @Mock RedisClusterClient clusterMock; + @Mock ClusterConnectionProvider connectionProviderMock; + @Mock ClusterCommandExecutor executorMock; @Mock ClusterNodeResourceProvider resourceProvider; + @Mock StatefulRedisClusterConnection sharedConnectionMock; @Mock RedisClusterAsyncCommands dedicatedConnectionMock; @Mock RedisClusterCommands clusterConnection1Mock; @Mock RedisClusterCommands clusterConnection2Mock; @@ -346,4 +355,43 @@ public class LettuceClusterConnectionUnitTests { verify(clusterConnection1Mock, never()).configResetstat(); } + @Test // DATAREDIS-731 + public void shouldExecuteOnSharedConnection() { + + RedisAdvancedClusterCommands sync = mock(RedisAdvancedClusterCommands.class); + + when(connectionProviderMock.getClient()).thenReturn(clusterMock); + when(sharedConnectionMock.sync()).thenReturn(sync); + + LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnectionMock, connectionProviderMock, + executorMock, Duration.ZERO); + + connection.keyCommands().del(KEY_1_BYTES); + + verify(sync).del(KEY_1_BYTES); + verify(connectionProviderMock).getClient(); + verifyNoMoreInteractions(connectionProviderMock); + } + + @Test // DATAREDIS-731 + public void shouldExecuteOnDedicatedConnection() { + + RedisCommands sync = mock(RedisCommands.class); + StatefulRedisConnection dedicatedConnection = mock(StatefulRedisConnection.class); + + when(connectionProviderMock.getClient()).thenReturn(clusterMock); + when(connectionProviderMock.getConnection(StatefulConnection.class)).thenReturn(dedicatedConnection); + when(dedicatedConnection.sync()).thenReturn(sync); + + LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnectionMock, connectionProviderMock, + executorMock, Duration.ZERO); + + connection.listCommands().bLPop(1, KEY_1_BYTES); + + verify(sync).blpop(1, KEY_1_BYTES); + verify(connectionProviderMock).getClient(); + verify(connectionProviderMock).getConnection(StatefulConnection.class); + verifyNoMoreInteractions(connectionProviderMock); + verifyZeroInteractions(sharedConnectionMock); + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java index 70ef1bf3d..7f218d050 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java @@ -20,6 +20,7 @@ import static org.hamcrest.core.IsEqual.*; import static org.hamcrest.core.IsInstanceOf.*; import static org.hamcrest.core.IsNull.*; import static org.junit.Assert.*; +import static org.mockito.Mockito.*; import static org.springframework.data.redis.connection.ClusterTestVariables.*; import static org.springframework.data.redis.connection.lettuce.LettuceTestClientResources.*; import static org.springframework.test.util.ReflectionTestUtils.*; @@ -29,6 +30,9 @@ import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.codec.RedisCodec; import io.lettuce.core.resource.ClientResources; import java.security.NoSuchAlgorithmException; @@ -38,6 +42,8 @@ import java.util.Collections; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentMatchers; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.connection.RedisClusterConfiguration; import org.springframework.data.redis.connection.RedisClusterConnection; @@ -552,6 +558,7 @@ public class LettuceConnectionFactoryUnitTests { LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig); connectionFactory.setShutdownTimeout(0); connectionFactory.setTimeout(2000); + connectionFactory.setShareNativeConnection(false); connectionFactory.afterPropertiesSet(); ConnectionFactoryTracker.add(connectionFactory); @@ -566,6 +573,7 @@ public class LettuceConnectionFactoryUnitTests { LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, LettuceClientConfiguration .builder().commandTimeout(Duration.ofSeconds(2)).shutdownTimeout(Duration.ZERO).build()); + connectionFactory.setShareNativeConnection(false); connectionFactory.afterPropertiesSet(); ConnectionFactoryTracker.add(connectionFactory); @@ -575,4 +583,30 @@ public class LettuceConnectionFactoryUnitTests { clusterConnection.close(); } + + @Test // DATAREDIS-731 + public void shouldShareNativeConnectionWithCluster() { + + RedisClusterClient clientMock = mock(RedisClusterClient.class); + StatefulRedisClusterConnection connectionMock = mock(StatefulRedisClusterConnection.class); + when(clientMock.connect(ByteArrayCodec.INSTANCE)).thenReturn(connectionMock); + + LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, + LettuceClientConfiguration.defaultConfiguration()) { + + @Override + protected AbstractRedisClient createClient() { + return clientMock; + } + }; + + connectionFactory.afterPropertiesSet(); + + new DirectFieldAccessor(connectionFactory).setPropertyValue("client", clientMock); + + connectionFactory.getClusterConnection().close(); + connectionFactory.getClusterConnection().close(); + + verify(clientMock).connect(ArgumentMatchers.any(RedisCodec.class)); + } }