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 046f558ca..811433aca 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,9 +22,9 @@ 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 io.lettuce.core.codec.ByteArrayCodec; import io.lettuce.core.codec.RedisCodec; +import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import java.util.ArrayList; import java.util.Collection; @@ -35,7 +35,6 @@ import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.DisposableBean; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; @@ -101,9 +100,9 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau Assert.notNull(executor, "ClusterCommandExecutor must not be null."); this.clusterClient = clusterClient; - topologyProvider = new LettuceClusterTopologyProvider(clusterClient); - clusterCommandExecutor = executor; - disposeClusterCommandExecutorOnClose = false; + this.topologyProvider = new LettuceClusterTopologyProvider(clusterClient); + this.clusterCommandExecutor = executor; + this.disposeClusterCommandExecutorOnClose = false; } /* @@ -224,9 +223,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau @Override public RedisClusterNode clusterGetNodeForSlot(int slot) { - DirectFieldAccessor accessor = new DirectFieldAccessor(clusterClient); - return LettuceConverters - .toRedisClusterNode(((Partitions) accessor.getPropertyValue("partitions")).getPartitionBySlot(slot)); + return LettuceConverters.toRedisClusterNode(clusterClient.getPartitions().getPartitionBySlot(slot)); } /* @@ -518,6 +515,19 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau return result; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#switchToPubSub() + */ + @Override + protected StatefulRedisPubSubConnection switchToPubSub() { + + close(); + + // open a pubsub one + return clusterClient.connectPubSub(CODEC); + } + public ClusterCommandExecutor getClusterCommandExecutor() { return clusterCommandExecutor; } 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 057edff4d..8d7ce5f07 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 @@ -798,11 +798,10 @@ public class LettuceConnection extends AbstractRedisConnection { } } - private StatefulRedisPubSubConnection switchToPubSub() { + protected StatefulRedisPubSubConnection switchToPubSub() { + close(); - // open a pubsub one return ((RedisClient) client).connectPubSub(CODEC); - // return ((RedisClient) client).connectPubSub(CODEC); } void pipeline(LettuceResult result) { @@ -892,7 +891,6 @@ public class LettuceConnection extends AbstractRedisConnection { ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); } } - } if (asyncDedicatedConn instanceof StatefulRedisConnection) { diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java index 4e2ca2125..c1820cd07 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2016 the original author or authors. + * Copyright 2011-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.redis.listener; import static org.junit.Assert.*; @@ -45,6 +44,8 @@ import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.RedisTestProfileValueSource; import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.TestCondition; +import org.springframework.data.redis.connection.ClusterTestVariables; +import org.springframework.data.redis.connection.RedisClusterConfiguration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; @@ -113,7 +114,16 @@ public class PubSubResubscribeTests { lettuceConnFactory.setValidateConnection(true); lettuceConnFactory.afterPropertiesSet(); - return Arrays.asList(new Object[][] { { jedisConnFactory }, { lettuceConnFactory } }); + LettuceConnectionFactory lettuceClusterConnFactory = new LettuceConnectionFactory( + new RedisClusterConfiguration().clusterNode(ClusterTestVariables.CLUSTER_NODE_1)); + lettuceClusterConnFactory.setClientResources(LettuceTestClientResources.getSharedClientResources()); + lettuceClusterConnFactory.setPort(port); + lettuceClusterConnFactory.setHostName(host); + lettuceClusterConnFactory.setValidateConnection(true); + lettuceClusterConnFactory.afterPropertiesSet(); + + return Arrays + .asList(new Object[][] { { jedisConnFactory }, { lettuceConnFactory }, { lettuceClusterConnFactory } }); } @Before @@ -251,7 +261,7 @@ public class PubSubResubscribeTests { /** * Validates the behavior of {@link RedisMessageListenerContainer} when it needs to spin up a thread executing its * PatternSubscriptionTask - * + * * @throws Exception */ @Test