refactoring

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

@@ -220,7 +220,7 @@ public class RedissonTopic implements RTopic {
     }
 
     @Override
-    public RFuture<Void> removeListenerAsync(final int listenerId) {
+    public RFuture<Void> removeListenerAsync(final Integer... listenerIds) {
         final RPromise<Void> promise = new RedissonPromise<Void>();
         final AsyncSemaphore semaphore = subscribeService.getSemaphore(channelName);
         semaphore.acquire(new Runnable() {
@@ -233,7 +233,9 @@ public class RedissonTopic implements RTopic {
                     return;
                 }
 
-                entry.removeListener(channelName, listenerId);
+                for (int id : listenerIds) {
+                    entry.removeListener(channelName, id);
+                }
                 if (!entry.hasListeners(channelName)) {
                     subscribeService.unsubscribe(channelName, semaphore)
                         .addListener(new TransferListener<Void>(promise));
@@ -247,7 +249,7 @@ public class RedissonTopic implements RTopic {
     }
     
     @Override
-    public void removeListener(int listenerId) {
+    public void removeListener(Integer... listenerIds) {
         AsyncSemaphore semaphore = subscribeService.getSemaphore(channelName);
         acquire(semaphore);
         
@@ -257,7 +259,9 @@ public class RedissonTopic implements RTopic {
             return;
         }
 
-        entry.removeListener(channelName, listenerId);
+        for (int id : listenerIds) {
+            entry.removeListener(channelName, id);
+        }
         if (!entry.hasListeners(channelName)) {
             subscribeService.unsubscribe(channelName, semaphore).syncUninterruptibly();
         } else {

redisson/src/main/java/org/redisson/api/RTopic.java

@@ -77,7 +77,7 @@ public interface RTopic extends RTopicAsync {
      *
      * @param listenerId - listener id
      */
-    void removeListener(int listenerId);
+    void removeListener(Integer... listenerIds);
 
     /**
      * Removes all listeners from this topic

redisson/src/main/java/org/redisson/api/RTopicAsync.java

@@ -62,7 +62,7 @@ public interface RTopicAsync {
      * @param listenerId - listener id
      * @return void
      */
-    RFuture<Void> removeListenerAsync(int listenerId);
+    RFuture<Void> removeListenerAsync(Integer... listenerIds);
 
     /**
      * Removes the listener by its instance

redisson/src/main/java/org/redisson/cache/LocalCacheListener.java

@@ -16,9 +16,11 @@
 package org.redisson.cache;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -278,12 +280,14 @@ public abstract class LocalCacheListener {
     }
     
     public void remove() {
+        List<Integer> ids = new ArrayList<Integer>(2);
         if (syncListenerId != 0) {
-            invalidationTopic.removeListener(syncListenerId);
+            ids.add(syncListenerId);
         }
         if (reconnectionListenerId != 0) {
-            invalidationTopic.removeListener(reconnectionListenerId);
+            ids.add(reconnectionListenerId);
         }
+        invalidationTopic.removeListenerAsync(ids.toArray(new Integer[ids.size()]));
     }
 
     public String getUpdatesLogName() {