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 267a28538..5c5ed1e89 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 @@ -22,6 +22,7 @@ import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.SlotHash; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.api.sync.RedisClusterCommands; +import io.lettuce.core.cluster.models.partitions.Partitions; import lombok.RequiredArgsConstructor; import java.time.Duration; @@ -63,6 +64,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau new LettuceExceptionConverter()); private final Log log = LogFactory.getLog(getClass()); + private final RedisClusterClient clusterClient; private ClusterCommandExecutor clusterCommandExecutor; private ClusterTopologyProvider topologyProvider; @@ -114,10 +116,14 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau super(null, connectionProvider, RedisURI.DEFAULT_TIMEOUT_DURATION.toMillis(), 0); - topologyProvider = new LettuceClusterTopologyProvider(getClient()); - clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, + Assert.isTrue(connectionProvider instanceof ClusterConnectionProvider, + "LettuceConnectionProvider must be a ClusterConnectionProvider."); + + this.clusterClient = ((ClusterConnectionProvider) connectionProvider).getRedisClient(); + this.topologyProvider = new LettuceClusterTopologyProvider(this.clusterClient); + this.clusterCommandExecutor = new ClusterCommandExecutor(this.topologyProvider, new LettuceClusterNodeResourceProvider(getConnectionProvider()), exceptionConverter); - disposeClusterCommandExecutorOnClose = true; + this.disposeClusterCommandExecutorOnClose = true; } /** @@ -147,30 +153,37 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau super(null, connectionProvider, timeout.toMillis(), 0); Assert.notNull(executor, "ClusterCommandExecutor must not be null."); + Assert.isTrue(connectionProvider instanceof ClusterConnectionProvider, + "LettuceConnectionProvider must be a ClusterConnectionProvider."); - this.topologyProvider = new LettuceClusterTopologyProvider(getClient()); + this.clusterClient = ((ClusterConnectionProvider) connectionProvider).getRedisClient(); + this.topologyProvider = new LettuceClusterTopologyProvider(this.clusterClient); this.clusterCommandExecutor = executor; this.disposeClusterCommandExecutorOnClose = false; } /** - * Creates new {@link LettuceClusterConnection} given a shared {@link StatefulRedisClusterConnection} - * and{@link LettuceConnectionProvider} running commands across the cluster via given {@link ClusterCommandExecutor}. + * 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 sharedConnection may be {@literal null} if no shared connection used. * @param connectionProvider must not be {@literal null}. + * @param clusterClient 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) { + LettuceClusterConnection(@Nullable StatefulRedisClusterConnection sharedConnection, + LettuceConnectionProvider connectionProvider, RedisClusterClient clusterClient, ClusterCommandExecutor executor, + Duration timeout) { super(sharedConnection, connectionProvider, timeout.toMillis(), 0); Assert.notNull(executor, "ClusterCommandExecutor must not be null."); + Assert.notNull(clusterClient, "RedisClusterClient must not be null."); - this.topologyProvider = new LettuceClusterTopologyProvider(getClient()); + this.clusterClient = clusterClient; + this.topologyProvider = new LettuceClusterTopologyProvider(clusterClient); this.clusterCommandExecutor = executor; this.disposeClusterCommandExecutorOnClose = false; } @@ -190,6 +203,14 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau connectionProvider.getClass().getName())); } + /** + * @return access to {@link RedisClusterClient} for non-connection access. + */ + private Partitions getPartitions() { + return clusterClient.getPartitions(); + } + + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#geoCommands() @@ -308,7 +329,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau @Override public RedisClusterNode clusterGetNodeForSlot(int slot) { - return LettuceConverters.toRedisClusterNode(getClient().getPartitions().getPartitionBySlot(slot)); + return LettuceConverters.toRedisClusterNode(getPartitions().getPartitionBySlot(slot)); } /* @@ -515,7 +536,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau return doGetClusterKeyCommands().keys(node, pattern); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisClusterConnection#scan(org.springframework.data.redis.connection.RedisClusterNode, org.springframework.data.redis.core.ScanOptions) */ @@ -524,7 +545,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau return doGetClusterKeyCommands().scan(node, options); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisClusterConnection#randomKey(org.springframework.data.redis.connection.RedisClusterNode) */ @@ -552,7 +573,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau */ @Override public List clusterGetNodes() { - return LettuceConverters.partitionsToClusterNodes(getClient().getPartitions()); + return LettuceConverters.partitionsToClusterNodes(getPartitions()); } /* 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 751c0fc48..3a21fd030 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 @@ -241,7 +241,7 @@ public class LettuceConnection extends AbstractRedisConnection { * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. * @since 2.0 */ - public LettuceConnection(StatefulRedisConnection sharedConnection, + public LettuceConnection(@Nullable StatefulRedisConnection sharedConnection, LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) { this((StatefulConnection) sharedConnection, connectionProvider, timeout, defaultDbIndex); } @@ -254,7 +254,8 @@ public class LettuceConnection extends AbstractRedisConnection { * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. * @since 2.1 */ - LettuceConnection(StatefulConnection sharedConnection, LettuceConnectionProvider connectionProvider, + LettuceConnection(@Nullable 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 12e806cab..a2b2fca1a 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,12 +345,14 @@ public class LettuceConnectionFactory throw new InvalidDataAccessApiUsageException("Cluster is not configured!"); } + RedisClusterClient clusterClient = (RedisClusterClient) client; + return getShareNativeConnection() ? new LettuceClusterConnection( (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection(), - connectionProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout()) - : new LettuceClusterConnection(connectionProvider, clusterCommandExecutor, - clientConfiguration.getCommandTimeout()); + connectionProvider, clusterClient, clusterCommandExecutor, clientConfiguration.getCommandTimeout()) + : new LettuceClusterConnection(null, connectionProvider, clusterClient, clusterCommandExecutor, + clientConfiguration.getCommandTimeout()); } /* 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 bd797e25f..3fa670474 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 @@ -68,7 +68,7 @@ public class LettuceClusterConnectionUnitTests { @Mock RedisClusterClient clusterMock; - @Mock ClusterConnectionProvider connectionProviderMock; + @Mock LettuceConnectionProvider connectionProviderMock; @Mock ClusterCommandExecutor executorMock; @Mock ClusterNodeResourceProvider resourceProvider; @Mock StatefulRedisClusterConnection sharedConnectionMock; @@ -355,41 +355,37 @@ public class LettuceClusterConnectionUnitTests { verify(clusterConnection1Mock, never()).configResetstat(); } - @Test // DATAREDIS-731 + @Test // DATAREDIS-731, DATAREDIS-545 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); + clusterMock, executorMock, Duration.ZERO); connection.keyCommands().del(KEY_1_BYTES); verify(sync).del(KEY_1_BYTES); - verify(connectionProviderMock).getClient(); verifyNoMoreInteractions(connectionProviderMock); } - @Test // DATAREDIS-731 + @Test // DATAREDIS-731, DATAREDIS-545 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); + clusterMock, 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/core/RedisClusterTemplateTests.java b/src/test/java/org/springframework/data/redis/core/RedisClusterTemplateTests.java index e42826e60..1c4f30f4a 100644 --- a/src/test/java/org/springframework/data/redis/core/RedisClusterTemplateTests.java +++ b/src/test/java/org/springframework/data/redis/core/RedisClusterTemplateTests.java @@ -33,6 +33,7 @@ import org.springframework.data.redis.StringObjectFactory; import org.springframework.data.redis.connection.RedisClusterConfiguration; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration; import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources; import org.springframework.data.redis.serializer.GenericToStringSerializer; import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; @@ -203,6 +204,12 @@ public class RedisClusterTemplateTests extends RedisTemplateTests { lettuceConnectionFactory.afterPropertiesSet(); + LettuceConnectionFactory pooledLettuceConnectionFactory = new LettuceConnectionFactory( + new RedisClusterConfiguration(CLUSTER_NODES), LettucePoolingClientConfiguration.builder() + .clientResources(LettuceTestClientResources.getSharedClientResources()).build()); + + pooledLettuceConnectionFactory.afterPropertiesSet(); + RedisTemplate lettuceStringTemplate = new RedisTemplate<>(); lettuceStringTemplate.setDefaultSerializer(StringRedisSerializer.UTF_8); lettuceStringTemplate.setConnectionFactory(lettuceConnectionFactory); @@ -233,6 +240,11 @@ public class RedisClusterTemplateTests extends RedisTemplateTests { lettuceJackson2JsonPersonTemplate.setValueSerializer(jackson2JsonSerializer); lettuceJackson2JsonPersonTemplate.afterPropertiesSet(); + RedisTemplate pooledLettuceStringTemplate = new RedisTemplate<>(); + pooledLettuceStringTemplate.setDefaultSerializer(StringRedisSerializer.UTF_8); + pooledLettuceStringTemplate.setConnectionFactory(pooledLettuceConnectionFactory); + pooledLettuceStringTemplate.afterPropertiesSet(); + return Arrays.asList(new Object[][] { // // JEDIS @@ -249,7 +261,8 @@ public class RedisClusterTemplateTests extends RedisTemplateTests { { lettuceRawTemplate, rawFactory, rawFactory }, // { lettucePersonTemplate, stringFactory, personFactory }, // { lettuceXstreamStringTemplate, stringFactory, stringFactory }, // - { lettuceJackson2JsonPersonTemplate, stringFactory, personFactory } // + { lettuceJackson2JsonPersonTemplate, stringFactory, personFactory }, // + { pooledLettuceStringTemplate, stringFactory, stringFactory } // }); }