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

Original pull request: spring-projects#450 & spring-projects#457

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;
+}
 }