Feature - RTopic.countSubscribers method added. #1472

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

@@ -27,6 +27,8 @@ import org.redisson.client.ChannelName;
 import org.redisson.client.RedisPubSubListener;
 import org.redisson.client.RedisTimeoutException;
 import org.redisson.client.codec.Codec;
+import org.redisson.client.codec.LongCodec;
+import org.redisson.client.codec.StringCodec;
 import org.redisson.client.protocol.RedisCommands;
 import org.redisson.command.CommandAsyncExecutor;
 import org.redisson.config.MasterSlaveServersConfig;
@@ -80,7 +82,7 @@ public class RedissonTopic implements RTopic {
 
     @Override
     public RFuture<Long> publishAsync(Object message) {
-        return commandExecutor.writeAsync(name, codec, RedisCommands.PUBLISH, name, encode(message));
+        return commandExecutor.writeAsync(name, StringCodec.INSTANCE, RedisCommands.PUBLISH, name, encode(message));
     }
 
     protected ByteBuf encode(Object value) {
@@ -273,4 +275,14 @@ public class RedissonTopic implements RTopic {
         return 0;
     }
 
+    @Override
+    public RFuture<Long> countSubscribersAsync() {
+        return commandExecutor.writeAsync(name, LongCodec.INSTANCE, RedisCommands.PUBSUB_NUMSUB, name);
+    }
+
+    @Override
+    public long countSubscribers() {
+        return commandExecutor.get(countSubscribersAsync());
+    }
+
 }

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

@@ -85,9 +85,17 @@ public interface RTopic extends RTopicAsync {
     void removeAllListeners();
 
     /**
-     * Returns amount of registered listeners
+     * Returns amount of registered listeners to this topic
      * 
      * @return amount of listeners
      */
     int countListeners();
+    
+    /**
+     * Returns amount of subscribers to this topic across all Redisson instances.
+     * Each subscriber may have multiple listeners.
+     * 
+     * @return amount of subscribers
+     */
+    long countSubscribers();
 }

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

@@ -72,4 +72,12 @@ public interface RTopicAsync {
      */
     RFuture<Void> removeListenerAsync(MessageListener<?> listener);
     
+    /**
+     * Returns amount of subscribers to this topic across all Redisson instances.
+     * Each subscriber may have multiple listeners.
+     * 
+     * @return amount of subscribers
+     */
+    RFuture<Long> countSubscribersAsync();
+    
 }

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

@@ -72,4 +72,13 @@ public interface RTopicReactive {
      * @param listenerId - listener id
      */
     void removeListener(int listenerId);
+    
+    /**
+     * Returns amount of subscribers to this topic across all Redisson instances.
+     * Each subscriber may have multiple listeners.
+     * 
+     * @return amount of subscribers
+     */
+    Publisher<Long> countSubscribers();
+    
 }

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

@@ -83,4 +83,12 @@ public interface RTopicRx {
      */
     <M> Flowable<M> getMessages(Class<M> type);
 
+    /**
+     * Returns amount of subscribers to this topic across all Redisson instances.
+     * Each subscriber may have multiple listeners.
+     * 
+     * @return amount of subscribers
+     */
+    Flowable<Long> countSubscribers();
+    
 }

redisson/src/main/java/org/redisson/client/protocol/RedisCommands.java

@@ -426,6 +426,7 @@ public interface RedisCommands {
     RedisStrictCommand<Void> QUIT = new RedisStrictCommand<Void>("QUIT", new VoidReplayConvertor());
 
     RedisStrictCommand<Long> PUBLISH = new RedisStrictCommand<Long>("PUBLISH");
+    RedisCommand<Long> PUBSUB_NUMSUB = new RedisCommand<Long>("PUBSUB", "NUMSUB", new ListObjectDecoder<Long>(1));
 
     RedisCommand<Object> SUBSCRIBE = new RedisCommand<Object>("SUBSCRIBE", new PubSubStatusDecoder());
     RedisCommand<Object> UNSUBSCRIBE = new RedisCommand<Object>("UNSUBSCRIBE", new PubSubStatusDecoder());

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

@@ -107,6 +107,20 @@ public class RedissonTopicTest {
 
     }
     
+    @Test
+    public void testCountSubscribers() {
+        RedissonClient redisson = BaseTest.createInstance();
+        RTopic topic1 = redisson.getTopic("topic", LongCodec.INSTANCE);
+        assertThat(topic1.countSubscribers()).isZero();
+        int id = topic1.addListener(Long.class, (channel, msg) -> {
+        });
+        assertThat(topic1.countSubscribers()).isOne();
+        topic1.removeListener(id);
+        assertThat(topic1.countSubscribers()).isZero();
+
+        redisson.shutdown();
+    }
+    
     @Test
     public void testCountListeners() {
         RedissonClient redisson = BaseTest.createInstance();