Attempt to stabilize integration tests

Makefile

@@ -388,11 +388,11 @@ stop:
 	pkill redis-sentinel && sleep 1 || true
 
 test-coverage: start
-	mvn -B -DskipITs=false clean compile verify jacoco:report -P$(PROFILE)
+	mvn -DskipITs=false clean compile verify jacoco:report -P$(PROFILE)
 	$(MAKE) stop
 
 test: start
-	mvn -B -DskipITs=false clean compile verify -P$(PROFILE)
+	mvn -DskipITs=false clean compile verify -P$(PROFILE)
 	$(MAKE) stop
 
 prepare: stop

pom.xml

@@ -862,6 +862,7 @@
                         <phase>integration-test</phase>
                         <goals>
                             <goal>integration-test</goal>
+                            <goal>verify</goal>
                         </goals>
                     </execution>
                 </executions>

src/test/java/io/lettuce/core/SslIntegrationTests.java

@@ -36,6 +36,7 @@
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -383,20 +384,15 @@ void pubSubSsl() {
         RedisPubSubCommands<String, String> connection = redisClient.connectPubSub(URI_NO_VERIFY).sync();
         connection.subscribe("c1");
         connection.subscribe("c2");
-        Delay.delay(Duration.ofMillis(200));
 
         RedisPubSubCommands<String, String> connection2 = redisClient.connectPubSub(URI_NO_VERIFY).sync();
 
         assertThat(connection2.pubsubChannels()).contains("c1", "c2");
         connection.quit();
-        Delay.delay(Duration.ofMillis(200));
-        Wait.untilTrue(connection::isOpen).waitOrTimeout();
-        Wait.untilEquals(2, () -> connection2.pubsubChannels().size()).waitOrTimeout();
+        Wait.untilTrue(connection.getStatefulConnection()::isOpen).waitOrTimeout();
+        Wait.untilEquals(2, () -> connection2.pubsubChannels().size()).during(Duration.ofSeconds(120)).waitOrTimeout();
 
         assertThat(connection2.pubsubChannels()).contains("c1", "c2");
-
-        connection.getStatefulConnection().close();
-        connection2.getStatefulConnection().close();
     }
 
     private static RedisURI.Builder sslURIBuilder(int portOffset) {

src/test/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubConnectionIntegrationTests.java

@@ -5,10 +5,10 @@
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
 
 import javax.inject.Inject;
 
+import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -22,16 +22,13 @@
 import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
 import io.lettuce.core.cluster.pubsub.api.async.NodeSelectionPubSubAsyncCommands;
 import io.lettuce.core.cluster.pubsub.api.async.PubSubAsyncNodeSelection;
-import io.lettuce.core.cluster.pubsub.api.async.RedisClusterPubSubAsyncCommands;
 import io.lettuce.core.cluster.pubsub.api.reactive.NodeSelectionPubSubReactiveCommands;
 import io.lettuce.core.cluster.pubsub.api.reactive.PubSubReactiveNodeSelection;
 import io.lettuce.core.cluster.pubsub.api.sync.NodeSelectionPubSubCommands;
 import io.lettuce.core.cluster.pubsub.api.sync.PubSubNodeSelection;
 import io.lettuce.core.event.command.CommandFailedEvent;
 import io.lettuce.core.event.command.CommandListener;
-import io.lettuce.core.internal.LettuceFactories;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
-import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
 import io.lettuce.core.support.PubSubTestListener;
 import io.lettuce.test.LettuceExtension;
 import io.lettuce.test.TestFutures;
@@ -58,32 +55,28 @@ class RedisClusterPubSubConnectionIntegrationTests extends TestSupport {
 
     private StatefulRedisClusterPubSubConnection<String, String> pubSubConnection2;
 
-    BlockingQueue<String> shardChannels;
-
     String shardChannel = "shard-channel";
 
-    String shardTestChannel = "shard-test-channel";
-
     @Inject
     RedisClusterPubSubConnectionIntegrationTests(RedisClusterClient clusterClient) {
         this.clusterClient = clusterClient;
-        shardChannels = LettuceFactories.newBlockingQueue();
-
     }
 
     @BeforeEach
     void openPubSubConnection() {
         connection = clusterClient.connect();
         pubSubConnection = clusterClient.connectPubSub();
         pubSubConnection2 = clusterClient.connectPubSub();
-
+        pubSubConnection.addListener(connectionListener);
     }
 
     @AfterEach
     void closePubSubConnection() {
         connection.close();
         pubSubConnection.close();
         pubSubConnection2.close();
+        connectionListener.clear();
+        pubSubConnection.removeListener(connectionListener);
     }
 
     @Test
@@ -103,36 +96,41 @@ void testRegularClientPubSubChannels() {
     @Test
     void testRegularClientPubSubShardChannels() {
 
-        String nodeId = pubSubConnection.sync().clusterMyId();
-        RedisClusterNode otherNode = getOtherThan(nodeId);
-        /// TODO : uncomment after SSUBSCRIBE is implemented
-        // pubSubConnection.sync().ssubscribe(key);
+        pubSubConnection.sync().ssubscribe(shardChannel);
+
+        Integer clusterKeyslot = connection.sync().clusterKeyslot(shardChannel).intValue();
+        RedisCommands<String, String> rightSlot =
+                connection.sync().nodes(node -> node.getSlots().contains(clusterKeyslot)).commands(0);
+        RedisCommands<String, String> wrongSlot =
+                connection.sync().nodes(node -> !node.getSlots().contains(clusterKeyslot)).commands(0);
 
-        List<String> channelsOnSubscribedNode = connection.getConnection(nodeId).sync().pubsubShardChannels();
-        // assertThat(channelsOnSubscribedNode).hasSize(1);
+        List<String> channelsOnSubscribedNode = rightSlot.pubsubShardChannels();
+        assertThat(channelsOnSubscribedNode).hasSize(1);
 
-        List<String> channelsOnOtherNode = connection.getConnection(otherNode.getNodeId()).sync().pubsubShardChannels();
+        List<String> channelsOnOtherNode = wrongSlot.pubsubShardChannels();
         assertThat(channelsOnOtherNode).isEmpty();
     }
 
     @Test
     @EnabledOnCommand("SSUBSCRIBE")
-    void subscribeToShardChannel() throws Exception {
-        pubSubConnection.addListener(connectionListener);
-        pubSubConnection.async().ssubscribe(shardChannel);
-        assertThat(connectionListener.getChannels().poll(3, TimeUnit.SECONDS)).isEqualTo(shardChannel);
+    void subscribeToShardChannel(){
+        pubSubConnection.sync().ssubscribe(shardChannel);
+
+        Wait.untilEquals(1L, connectionListener.getShardCounts()::poll).waitOrTimeout();
+        Wait.untilEquals(shardChannel, connectionListener.getShardChannels()::poll).waitOrTimeout();
     }
 
     @Test
     @EnabledOnCommand("SSUBSCRIBE")
-    void subscribeToShardChannelViaOtherEndpoint() throws Exception {
-        pubSubConnection.addListener(connectionListener);
-        RedisClusterPubSubAsyncCommands<String, String> pubSub = pubSubConnection.async();
-        String nodeId = pubSub.clusterMyId().get();
-        RedisPubSubAsyncCommands<String, String> other = pubSub
-                .nodes(node -> node.getRole().isUpstream() && !node.getNodeId().equals(nodeId)).commands(0);
-        other.ssubscribe(shardChannel);
-        assertThat(connectionListener.getChannels().poll(3, TimeUnit.SECONDS)).isEqualTo(shardChannel);
+    void subscribeToShardChannelViaReplica() {
+        int clusterKeyslot = connection.sync().clusterKeyslot(shardChannel).intValue();
+        String thisNode = connection.getPartitions().getPartitionBySlot(clusterKeyslot).getNodeId();
+
+        RedisPubSubAsyncCommands<String, String> replica =
+                pubSubConnection.async().nodes(node -> thisNode.equals(node.getSlaveOf())).commands(0);
+        replica.ssubscribe(shardChannel);
+
+        Wait.untilEquals(shardChannel, connectionListener.getShardChannels()::poll).waitOrTimeout();
     }
 
     @Test
@@ -169,7 +167,6 @@ void testRegularClientPublish() throws Exception {
         String nodeId = pubSubConnection.sync().clusterMyId();
         RedisClusterNode otherNode = getOtherThan(nodeId);
         pubSubConnection.sync().subscribe(key);
-        pubSubConnection.addListener(connectionListener);
 
         connection.getConnection(nodeId).sync().publish(key, value);
         assertThat(connectionListener.getMessages().take()).isEqualTo(value);
@@ -183,7 +180,6 @@ void testPubSubClientPublish() throws Exception {
 
         String nodeId = pubSubConnection.sync().clusterMyId();
         pubSubConnection.sync().subscribe(key);
-        pubSubConnection.addListener(connectionListener);
 
         assertThat(pubSubConnection2.sync().clusterMyId()).isEqualTo(nodeId);
 
@@ -210,7 +206,6 @@ void testRegularClientPubSubPublish() throws Exception {
         String nodeId = pubSubConnection.sync().clusterMyId();
         RedisClusterNode otherNode = getOtherThan(nodeId);
         pubSubConnection.sync().subscribe(key);
-        pubSubConnection.addListener(connectionListener);
 
         List<String> channelsOnSubscribedNode = connection.getConnection(nodeId).sync().pubsubChannels();
         assertThat(channelsOnSubscribedNode).hasSize(1);
@@ -264,7 +259,6 @@ void testNodeMessagePropagationSubscription() throws Exception {
 
         RedisClusterNode partition = pubSubConnection.getPartitions().getPartition(0);
         pubSubConnection.setNodeMessagePropagation(true);
-        pubSubConnection.addListener(connectionListener);
 
         StatefulRedisPubSubConnection<String, String> node = pubSubConnection.getConnection(partition.getNodeId());
         node.sync().subscribe("channel");
@@ -279,7 +273,6 @@ void testNodeHostAndPortMessagePropagationSubscription() throws Exception {
 
         RedisClusterNode partition = pubSubConnection.getPartitions().getPartition(0);
         pubSubConnection.setNodeMessagePropagation(true);
-        pubSubConnection.addListener(connectionListener);
 
         RedisURI uri = partition.getUri();
         StatefulRedisPubSubConnection<String, String> node = pubSubConnection.getConnection(uri.getHost(), uri.getPort());
@@ -294,7 +287,6 @@ void testNodeHostAndPortMessagePropagationSubscription() throws Exception {
     void testAsyncSubscription() throws Exception {
 
         pubSubConnection.setNodeMessagePropagation(true);
-        pubSubConnection.addListener(connectionListener);
 
         PubSubAsyncNodeSelection<String, String> masters = pubSubConnection.async().masters();
         NodeSelectionPubSubAsyncCommands<String, String> commands = masters.commands();
@@ -312,7 +304,6 @@ void testAsyncSubscription() throws Exception {
     void testSyncSubscription() throws Exception {
 
         pubSubConnection.setNodeMessagePropagation(true);
-        pubSubConnection.addListener(connectionListener);
 
         PubSubNodeSelection<String, String> masters = pubSubConnection.sync().masters();
         NodeSelectionPubSubCommands<String, String> commands = masters.commands();
@@ -330,7 +321,6 @@ void testSyncSubscription() throws Exception {
     void testReactiveSubscription() throws Exception {
 
         pubSubConnection.setNodeMessagePropagation(true);
-        pubSubConnection.addListener(connectionListener);
 
         PubSubReactiveNodeSelection<String, String> masters = pubSubConnection.reactive().masters();
         NodeSelectionPubSubReactiveCommands<String, String> commands = masters.commands();
@@ -350,7 +340,6 @@ void testClusterListener() throws Exception {
 
         BlockingQueue<RedisClusterNode> nodes = new LinkedBlockingQueue<>();
         pubSubConnection.setNodeMessagePropagation(true);
-        pubSubConnection.addListener(connectionListener);
         pubSubConnection.addListener(new RedisClusterPubSubAdapter<String, String>() {
 
             @Override

src/test/java/io/lettuce/core/support/PubSubTestListener.java

@@ -33,6 +33,8 @@ public class PubSubTestListener implements RedisPubSubListener<String, String> {
     private BlockingQueue<String> patterns = LettuceFactories.newBlockingQueue();
     private BlockingQueue<String> messages = LettuceFactories.newBlockingQueue();
     private BlockingQueue<Long> counts = LettuceFactories.newBlockingQueue();
+    private BlockingQueue<String> shardChannels = LettuceFactories.newBlockingQueue();
+    private BlockingQueue<Long> shardCounts = LettuceFactories.newBlockingQueue();
 
     // RedisPubSubListener implementation
 
@@ -55,6 +57,12 @@ public void subscribed(String channel, long count) {
         counts.add(count);
     }
 
+    @Override
+    public void ssubscribed(String shardChannel, long count) {
+        shardChannels.add(shardChannel);
+        shardCounts.add(count);
+    }
+
     @Override
     public void psubscribed(String pattern, long count) {
         patterns.add(pattern);
@@ -77,6 +85,10 @@ public BlockingQueue<String> getChannels() {
         return channels;
     }
 
+    public BlockingQueue<String> getShardChannels() {
+        return shardChannels;
+    }
+
     public BlockingQueue<String> getPatterns() {
         return patterns;
     }
@@ -89,13 +101,19 @@ public BlockingQueue<Long> getCounts() {
         return counts;
     }
 
+    public BlockingQueue<Long> getShardCounts() {
+        return shardCounts;
+    }
+
     /**
      * Clear listener state (i.e. channels, patterns, messages, counts).
      */
     public void clear() {
         channels.clear();
+        shardChannels.clear();
         patterns.clear();
         messages.clear();
         counts.clear();
+        shardCounts.clear();
     }
 }