DATAREDIS-508 - Support lettuce Pub/Sub when using Redis Cluster.

We now use Lettuce's Redis Cluster Pub/Sub connection to create Pub/Sub connections in a cluster.

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<byte[], byte[]> switchToPubSub() {
+
+		close();
+
+		// open a pubsub one
+		return clusterClient.connectPubSub(CODEC);
+	}
+
 	public ClusterCommandExecutor getClusterCommandExecutor() {
 		return clusterCommandExecutor;
 	}

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -798,11 +798,10 @@ public class LettuceConnection extends AbstractRedisConnection {
 		}
 	}
 
-	private StatefulRedisPubSubConnection<byte[], byte[]> switchToPubSub() {
+	protected StatefulRedisPubSubConnection<byte[], byte[]> 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<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
 				}
 			}
-
 		}
 
 		if (asyncDedicatedConn instanceof StatefulRedisConnection) {

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