Fixed - RTopic.removeAllListeners leads to PubSub connections leak. #1696

redisson/src/main/java/org/redisson/RedissonPatternTopic.java

@@ -147,8 +147,7 @@ public class RedissonPatternTopic implements RPatternTopic {
             return;
         }
 
-        entry.removeAllListeners(channelName);
-        if (!entry.hasListeners(channelName)) {
+        if (entry.removeAllListeners(channelName)) {
             subscribeService.punsubscribe(channelName, semaphore);
         } else {
             semaphore.release();

redisson/src/main/java/org/redisson/RedissonTopic.java

@@ -146,7 +146,20 @@ public class RedissonTopic implements RTopic {
 
     @Override
     public void removeAllListeners() {
-        subscribeService.unsubscribe(channelName, PubSubType.UNSUBSCRIBE);
+        AsyncSemaphore semaphore = subscribeService.getSemaphore(channelName);
+        acquire(semaphore);
+        
+        PubSubConnectionEntry entry = subscribeService.getPubSubEntry(channelName);
+        if (entry == null) {
+            semaphore.release();
+            return;
+        }
+
+        if (entry.removeAllListeners(channelName)) {
+            subscribeService.unsubscribe(channelName, semaphore);
+        } else {
+            semaphore.release();
+        }
     }
 
     protected void acquire(AsyncSemaphore semaphore) {

redisson/src/main/java/org/redisson/pubsub/PubSubConnectionEntry.java

@@ -100,7 +100,7 @@ public class PubSubConnectionEntry {
         for (RedisPubSubListener<?> listener : listeners) {
             removeListener(channelName, listener);
         }
-        return !listeners.isEmpty();
+        return listeners.isEmpty();
     }
     
     // TODO optimize

redisson/src/test/java/org/redisson/RedissonTopicTest.java

@@ -369,9 +369,11 @@ public class RedissonTopicTest {
     public void testRemoveAllListeners() throws InterruptedException {
         RedissonClient redisson = BaseTest.createInstance();
         RTopic topic1 = redisson.getTopic("topic1");
+        AtomicInteger counter = new AtomicInteger();
+        
         for (int i = 0; i < 10; i++) {
             topic1.addListener(Message.class, (channel, msg) -> {
-                Assert.fail();
+                counter.incrementAndGet();
             });
         }
 
@@ -379,6 +381,33 @@ public class RedissonTopicTest {
         topic1.removeAllListeners();
         topic1.publish(new Message("123"));
 
+        Thread.sleep(1000);
+        assertThat(counter.get()).isZero();
+        
+        redisson.shutdown();
+    }
+
+    @Test
+    public void testRemoveAllListeners2() throws InterruptedException {
+        RedissonClient redisson = BaseTest.createInstance();
+        RTopic topic1 = redisson.getTopic("topic1");
+        AtomicInteger counter = new AtomicInteger();
+        
+        for (int j = 0; j < 100; j++) {
+            for (int i = 0; i < 10; i++) {
+                topic1.addListener(Message.class, (channel, msg) -> {
+                    counter.incrementAndGet();
+                });
+            }
+            
+            topic1 = redisson.getTopic("topic1");
+            topic1.removeAllListeners();
+            topic1.publish(new Message("123"));
+        }
+
+        Thread.sleep(1000);
+        assertThat(counter.get()).isZero();
+        
         redisson.shutdown();
     }