DATAREDIS-976 - Add Pub/Sub tests for Redis Cluster.

Original pull request: #450.

Author: mp911de

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

@@ -22,8 +22,6 @@
 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.cluster.pubsub.StatefulRedisClusterPubSubConnection;
 import lombok.RequiredArgsConstructor;
 
 import java.time.Duration;
@@ -55,6 +53,9 @@
 import org.springframework.util.ObjectUtils;
 
 /**
+ * {@code RedisClusterConnection} implementation on top of <a href="https://github.com/mp911de/lettuce">Lettuce</a>
+ * Redis client.
+ *
  * @author Christoph Strobl
  * @author Mark Paluch
  * @since 1.7

src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java

@@ -15,20 +15,24 @@
  */
 package org.springframework.data.redis.listener;
 
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Collection;
 
+import org.junit.runners.model.Statement;
+
 import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.Person;
 import org.springframework.data.redis.PersonObjectFactory;
 import org.springframework.data.redis.RawObjectFactory;
 import org.springframework.data.redis.SettingsUtils;
 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.LettuceTestClientResources;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.test.util.RedisClusterRule;
 
 /**
  * @author Costin Leau
@@ -76,8 +80,49 @@ public static Collection<Object[]> testParams() {
 		rawTemplateLtc.setConnectionFactory(lettuceConnFactory);
 		rawTemplateLtc.afterPropertiesSet();
 
-		return Arrays.asList(new Object[][] { { stringFactory, stringTemplate }, { personFactory, personTemplate },
-				{ rawFactory, rawTemplate }, { stringFactory, stringTemplateLtc }, { personFactory, personTemplateLtc },
-				{ rawFactory, rawTemplateLtc } });
+		Collection<Object[]> parameters = new ArrayList<>();
+		parameters.add(new Object[] { stringFactory, stringTemplate });
+		parameters.add(new Object[] { personFactory, personTemplate });
+		parameters.add(new Object[] { stringFactory, stringTemplateLtc });
+		parameters.add(new Object[] { personFactory, personTemplateLtc });
+		parameters.add(new Object[] { rawFactory, rawTemplateLtc });
+
+		if (clusterAvailable()) {
+
+			RedisClusterConfiguration configuration = new RedisClusterConfiguration().clusterNode("127.0.0.1", 7379);
+
+			// add Jedis
+			JedisConnectionFactory jedisClusterFactory = new JedisConnectionFactory(configuration);
+			jedisClusterFactory.afterPropertiesSet();
+
+			RedisTemplate<String, String> jedisClusterStringTemplate = new StringRedisTemplate(jedisClusterFactory);
+
+			// add Lettuce
+			LettuceConnectionFactory lettuceClusterFactory = new LettuceConnectionFactory(configuration);
+			lettuceClusterFactory.setClientResources(LettuceTestClientResources.getSharedClientResources());
+			lettuceClusterFactory.afterPropertiesSet();
+
+			RedisTemplate<String, String> lettuceClusterStringTemplate = new StringRedisTemplate(lettuceClusterFactory);
+
+			parameters.add(new Object[] { stringFactory, jedisClusterStringTemplate });
+			parameters.add(new Object[] { stringFactory, lettuceClusterStringTemplate });
+		}
+
+		return parameters;
+	}
+
+	private static boolean clusterAvailable() {
+
+		try {
+			new RedisClusterRule().apply(new Statement() {
+				@Override
+				public void evaluate() {
+
+				}
+			}, null).evaluate();
+		} catch (Throwable throwable) {
+			return false;
+		}
+		return true;
 	}
 }

src/test/java/org/springframework/data/redis/listener/PubSubTests.java

@@ -30,32 +30,31 @@
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
+
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.ObjectFactory;
-import org.springframework.data.redis.RedisTestProfileValueSource;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
-import org.springframework.data.redis.test.util.RedisSentinelRule;
 
 /**
  * Base test class for PubSub integration tests
  *
  * @author Costin Leau
  * @author Jennifer Hickey
+ * @author Mark Paluch
  */
 @RunWith(Parameterized.class)
 public class PubSubTests<T> {
 
-	public @Rule RedisSentinelRule sentinelRule = RedisSentinelRule.withDefaultConfig().sentinelsDisabled();
-
 	private static final String CHANNEL = "pubsub::test";
 
 	protected RedisMessageListenerContainer container;
@@ -73,11 +72,6 @@ public void handleMessage(Object message) {
 
 	private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler);
 
-	@BeforeClass
-	public static void shouldRun() {
-		assumeTrue(RedisTestProfileValueSource.matches("runLongTests", "true"));
-	}
-
 	@Before
 	public void setUp() throws Exception {
 		bag.clear();
@@ -178,6 +172,9 @@ public void testStartNoListeners() {
 	@SuppressWarnings("unchecked")
 	@Test // DATAREDIS-251
 	public void testStartListenersToNoSpecificChannelTest() throws InterruptedException {
+
+		assumeTrue(isClusterAware(template.getConnectionFactory()));
+
 		container.removeMessageListener(adapter, new ChannelTopic(CHANNEL));
 		container.addMessageListener(adapter, Arrays.asList(new PatternTopic("*")));
 		container.start();
@@ -193,4 +190,14 @@ public void testStartListenersToNoSpecificChannelTest() throws InterruptedExcept
 
 		assertThat(set, hasItems(payload));
 	}
+
+	private static boolean isClusterAware(RedisConnectionFactory connectionFactory) {
+
+		if (connectionFactory instanceof LettuceConnectionFactory) {
+			return ((LettuceConnectionFactory) connectionFactory).isClusterAware();
+		} else if (connectionFactory instanceof JedisConnectionFactory) {
+			return ((JedisConnectionFactory) connectionFactory).isRedisClusterAware();
+		}
+		return false;
+	}
 }