Polishing #2758

Introduce convenience method to clear listener state.

Original pull request: #2813

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

@@ -39,6 +39,8 @@
 import io.lettuce.test.condition.EnabledOnCommand;
 
 /**
+ * Integration tests for Cluster Pub/Sub.
+ *
  * @author Mark Paluch
  */
 @ExtendWith(LettuceExtension.class)

src/test/java/io/lettuce/core/pubsub/PubSubCommandTest.java

@@ -36,7 +36,12 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import io.lettuce.core.*;
+import io.lettuce.core.AbstractRedisClientTest;
+import io.lettuce.core.ClientOptions;
+import io.lettuce.core.KillArgs;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisFuture;
+import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.api.push.PushMessage;
 import io.lettuce.core.internal.LettuceFactories;
@@ -91,10 +96,7 @@ void openPubSubConnection() {
             pubsub = client.connectPubSub().async();
             pubsub.getStatefulConnection().addListener(listener);
         } finally {
-            channels.clear();
-            patterns.clear();
-            messages.clear();
-            counts.clear();
+            listener.clear();
         }
     }
 
@@ -378,7 +380,7 @@ void pubsubMultipleShardChannels() {
 
     @Test
     @EnabledOnCommand("SPUBLISH")
-    void pubsubShardChannelsWithArg() {        
+    void pubsubShardChannelsWithArg() {
         /// TODO : uncomment after SSUBSCRIBE is implemented
         // TestFutures.awaitOrTimeout(pubsub.ssubscribe(channel));
         List<String> result = redis.pubsubShardChannels(pattern);

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

@@ -1,3 +1,22 @@
+/*
+ * Copyright 2011-Present, Redis Ltd. and Contributors
+ * All rights reserved.
+ *
+ * Licensed under the MIT License.
+ *
+ * This file contains contributions from third-party contributors
+ * licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.lettuce.core.support;
 
 import java.util.concurrent.BlockingQueue;
@@ -69,4 +88,14 @@ public BlockingQueue<String> getMessages() {
     public BlockingQueue<Long> getCounts() {
         return counts;
     }
+
+    /**
+     * Clear listener state (i.e. channels, patterns, messages, counts).
+     */
+    public void clear() {
+        channels.clear();
+        patterns.clear();
+        messages.clear();
+        counts.clear();
+    }
 }