implementation of PUBSUB SHARDCHANNELS (#2793)

* implementation of PUBSUB SHARDCHANNELS / issue #2756

* Polishing #2756

src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java

@@ -56,6 +56,7 @@
  * @author Tugdual Grall
  * @author dengliming
  * @author Andrey Shlykov
+ * @author Ali Takavci
  */
 @SuppressWarnings("unchecked")
 public abstract class AbstractRedisAsyncCommands<K, V> implements RedisAclAsyncCommands<K, V>, RedisHashAsyncCommands<K, V>,
@@ -1556,6 +1557,16 @@ public RedisFuture<Map<K, Long>> pubsubNumsub(K... channels) {
         return dispatch(commandBuilder.pubsubNumsub(channels));
     }
 
+    @Override
+    public RedisFuture<List<K>> pubsubShardChannels() {
+        return dispatch(commandBuilder.pubsubShardChannels());
+    }
+
+    @Override
+    public RedisFuture<List<K>> pubsubShardChannels(K pattern) {
+        return dispatch(commandBuilder.pubsubShardChannels(pattern));
+    }
+
     @Override
     public RedisFuture<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         return dispatch(commandBuilder.pubsubShardNumsub(shardChannels));

src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java

@@ -65,6 +65,7 @@
  * @author Tugdual Grall
  * @author dengliming
  * @author Andrey Shlykov
+ * @author Ali Takavci
  * @since 4.0
  */
 public abstract class AbstractRedisReactiveCommands<K, V>
@@ -1635,6 +1636,16 @@ public Mono<Map<K, Long>> pubsubNumsub(K... channels) {
         return createMono(() -> commandBuilder.pubsubNumsub(channels));
     }
 
+    @Override
+    public Flux<K> pubsubShardChannels() {
+        return createDissolvingFlux(commandBuilder::pubsubShardChannels);
+    }
+
+    @Override
+    public Flux<K> pubsubShardChannels(K pattern) {
+        return createDissolvingFlux(() -> commandBuilder.pubsubShardChannels(pattern));
+    }
+
     @Override
     public Mono<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         return createMono(() -> commandBuilder.pubsubShardNumsub(shardChannels));

src/main/java/io/lettuce/core/RedisCommandBuilder.java

@@ -53,6 +53,7 @@
  * @author dengliming
  * @author Mikhael Sokolov
  * @author Tihomir Mateev
+ * @author Ali Takavci
  */
 @SuppressWarnings({ "unchecked", "varargs" })
 class RedisCommandBuilder<K, V> extends BaseRedisCommandBuilder<K, V> {
@@ -2108,6 +2109,18 @@ Command<K, V, Map<K, Long>> pubsubNumsub(K... channels) {
         return createCommand(PUBSUB, (MapOutput) new MapOutput<K, Long>((RedisCodec) codec), args);
     }
 
+    Command<K, V, List<K>> pubsubShardChannels() {
+        CommandArgs<K, V> args = new CommandArgs<>(codec).add(SHARDCHANNELS);
+        return createCommand(PUBSUB, new KeyListOutput<>(codec), args);
+    }
+
+    Command<K, V, List<K>> pubsubShardChannels(K pattern) {
+        LettuceAssert.notNull(pattern, "Pattern " + MUST_NOT_BE_NULL);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).add(SHARDCHANNELS).addKey(pattern);
+        return createCommand(PUBSUB, new KeyListOutput<>(codec), args);
+    }
+
     Command<K, V, Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         LettuceAssert.notNull(shardChannels, "ShardChannels " + MUST_NOT_BE_NULL);
         LettuceAssert.notEmpty(shardChannels, "ShardChannels " + MUST_NOT_BE_EMPTY);

src/main/java/io/lettuce/core/api/async/BaseRedisAsyncCommands.java

@@ -29,6 +29,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Ali Takavci
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateAsyncApi
  */
@@ -66,6 +67,21 @@ public interface BaseRedisAsyncCommands<K, V> {
      */
     RedisFuture<Map<K, Long>> pubsubNumsub(K... channels);
 
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @return List&lt;K&gt; array-reply a list of active channels.
+     */
+    RedisFuture<List<K>> pubsubShardChannels();
+
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @param pattern the pattern type: patternkey (pattern).
+     * @return List&lt;K&gt; array-reply a list of active channels, optionally matching the specified pattern.
+     */
+    RedisFuture<List<K>> pubsubShardChannels(K pattern);
+
     /**
      * Returns the number of subscribers (not counting clients subscribed to patterns) for the specified shard channels.
      *

src/main/java/io/lettuce/core/api/reactive/BaseRedisReactiveCommands.java

@@ -29,6 +29,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Ali Takavci
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateReactiveApi
  */
@@ -66,6 +67,21 @@ public interface BaseRedisReactiveCommands<K, V> {
      */
     Mono<Map<K, Long>> pubsubNumsub(K... channels);
 
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @return K array-reply a list of active channels.
+     */
+    Flux<K> pubsubShardChannels();
+
+    /**
+     * Lists the currently *active shard channels*.     
+     * 
+     * @param pattern the pattern type: patternkey (pattern).
+     * @return K array-reply a list of active channels, optionally matching the specified pattern.
+     */
+    Flux<K> pubsubShardChannels(K pattern);
+
     /**
      * Returns the number of subscribers (not counting clients subscribed to patterns) for the specified shard channels.
      *

src/main/java/io/lettuce/core/api/sync/BaseRedisCommands.java

@@ -28,6 +28,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Ali Takavci
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateSyncApi
  */
@@ -65,6 +66,21 @@ public interface BaseRedisCommands<K, V> {
      */
     Map<K, Long> pubsubNumsub(K... channels);
 
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @return List&lt;K&gt; array-reply a list of active channels.
+     */
+    List<K> pubsubShardChannels();
+
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @param pattern the pattern type: patternkey (pattern).
+     * @return List&lt;K&gt; array-reply a list of active channels, optionally matching the specified pattern.
+     */
+    List<K> pubsubShardChannels(K pattern);
+
     /**
      * Returns the number of subscribers (not counting clients subscribed to patterns) for the specified shard channels.
      *

src/main/java/io/lettuce/core/cluster/api/async/BaseNodeSelectionAsyncCommands.java

@@ -29,6 +29,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Ali Takavci
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateAsyncNodeSelectionClusterApi
  */
@@ -66,6 +67,21 @@ public interface BaseNodeSelectionAsyncCommands<K, V> {
      */
     AsyncExecutions<Map<K, Long>> pubsubNumsub(K... channels);
 
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @return List&lt;K&gt; array-reply a list of active channels.
+     */
+    AsyncExecutions<List<K>> pubsubShardChannels();
+
+        /**
+     * Lists the currently *active shard channels*.
+     *
+     * @param pattern the pattern type: patternkey (pattern).
+     * @return List&lt;K&gt; array-reply a list of active channels, optionally matching the specified pattern.
+     */
+    AsyncExecutions<List<K>> pubsubShardChannels(K pattern);
+
     /**
      * Returns the number of subscribers (not counting clients subscribed to patterns) for the specified shard channels.
      *

src/main/java/io/lettuce/core/cluster/api/sync/BaseNodeSelectionCommands.java

@@ -24,6 +24,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Ali Takavci
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateSyncNodeSelectionClusterApi
  */
@@ -61,6 +62,21 @@ public interface BaseNodeSelectionCommands<K, V> {
      */
     Executions<Map<K, Long>> pubsubNumsub(K... channels);
 
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @return List&lt;K&gt; array-reply a list of active channels.
+     */
+    Executions<List<K>> pubsubShardChannels();
+
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @param pattern the pattern type: patternkey (pattern).
+     * @return List&lt;K&gt; array-reply a list of active channels, optionally matching the specified pattern.
+     */
+    Executions<List<K>> pubsubShardChannels(K pattern);
+
     /**
      * Returns the number of subscribers (not counting clients subscribed to patterns) for the specified shard channels.
      *

src/main/java/io/lettuce/core/protocol/CommandKeyword.java

@@ -25,6 +25,7 @@
  * @author Zhang Jessey
  * @author dengliming
  * @author Tihomir Mateev
+ * @author Ali Takavci
  */
 public enum CommandKeyword implements ProtocolKeyword {
 
@@ -36,7 +37,7 @@ public enum CommandKeyword implements ProtocolKeyword {
 
     IDLETIME, JUSTID, KILL, KEYSLOT, LEFT, LEN, LIMIT, LIST, LOAD, LOG, MATCH,
 
-    MAX, MAXLEN, MEET, MIN, MINID, MOVED, NO, NOACK, NOCOMMANDS, NODE, NODES, NOMKSTREAM, NOPASS, NOSAVE, NOT, NUMSUB, SHARDNUMSUB, NUMPAT, NX, OFF, ON, ONE, OR, PAUSE,
+    MAX, MAXLEN, MEET, MIN, MINID, MOVED, NO, NOACK, NOCOMMANDS, NODE, NODES, NOMKSTREAM, NOPASS, NOSAVE, NOT, NUMSUB, SHARDCHANNELS, SHARDNUMSUB, NUMPAT, NX, OFF, ON, ONE, OR, PAUSE,
 
     REFCOUNT, REMOVE, RELOAD, REPLACE, REPLICATE, REPLICAS, REV, RESET, RESETCHANNELS, RESETKEYS, RESETPASS,
 

src/main/java/io/lettuce/core/pubsub/PubSubCommandBuilder.java

@@ -37,6 +37,7 @@
  *
  * @author Mark Paluch
  * @author Tihomir Mateev
+ * @author Ali Takavci
  * @since 4.2
  */
 @SuppressWarnings("varargs")
@@ -67,6 +68,11 @@ final Command<K, V, Map<K, Long>> pubsubNumsub(K... channels) {
         return createCommand(PUBSUB, new MapOutput<>((RedisCodec) codec), args);
     }
 
+    Command<K, V, List<K>> pubsubShardChannels(K pattern) {
+        CommandArgs<K, V> args = new PubSubCommandArgs<>(codec).add(SHARDCHANNELS).addKey(pattern);
+        return createCommand(PUBSUB, new KeyListOutput<>(codec), args);
+    }
+
     @SafeVarargs
     @SuppressWarnings({ "unchecked", "rawtypes" })
     final Command<K, V, Map<K, Long>> pubsubShardNumsub(K... shardChannels) {

src/main/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImpl.java

@@ -30,6 +30,7 @@
  * @param <V> Value type.
  * @author Will Glozer
  * @author Mark Paluch
+ * @author Ali Takavci
  */
 public class RedisPubSubAsyncCommandsImpl<K, V> extends RedisAsyncCommandsImpl<K, V> implements RedisPubSubAsyncCommands<K, V> {
 
@@ -85,6 +86,11 @@ public RedisFuture<Map<K, Long>> pubsubNumsub(K... channels) {
         return dispatch(commandBuilder.pubsubNumsub(channels));
     }
 
+    @Override
+    public RedisFuture<List<K>> pubsubShardChannels(K pattern) {
+        return dispatch(commandBuilder.pubsubShardChannels(pattern));
+    }
+
     @Override
     public RedisFuture<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         return dispatch(commandBuilder.pubsubShardNumsub(shardChannels));

src/main/java/io/lettuce/core/pubsub/RedisPubSubReactiveCommandsImpl.java

@@ -32,6 +32,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Ali Takavci
  * @since 5.0
  */
 public class RedisPubSubReactiveCommandsImpl<K, V> extends RedisReactiveCommandsImpl<K, V>
@@ -143,6 +144,11 @@ public Mono<Map<K, Long>> pubsubNumsub(K... channels) {
         return createMono(() -> commandBuilder.pubsubNumsub(channels));
     }
 
+    @Override
+    public Flux<K> pubsubShardChannels(K channel) {
+        return createDissolvingFlux(() -> commandBuilder.pubsubShardChannels(channel));
+    }
+
     @Override
     public Mono<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         return createMono(() -> commandBuilder.pubsubShardNumsub(shardChannels));

src/main/kotlin/io/lettuce/core/api/coroutines/BaseRedisCoroutinesCommands.kt

@@ -28,6 +28,7 @@ import kotlinx.coroutines.flow.Flow
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mikhael Sokolov
+ * @author Ali Takavci
  * @since 6.0
  * @generated by io.lettuce.apigenerator.CreateKotlinCoroutinesApi
  */
@@ -65,6 +66,21 @@ interface BaseRedisCoroutinesCommands<K : Any, V : Any> {
      * @return array-reply a list of channels and number of subscribers for every channel.
      */
     suspend fun pubsubNumsub(vararg channels: K): Map<K, Long>
+
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @return List<K> array-reply a list of active channels.
+     */
+    suspend fun pubsubShardChannels(): List<K>
+
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @param pattern the pattern type: patternkey (pattern).
+     * @return List<K> array-reply a list of active channels, optionally matching the specified pattern.
+     */
+    suspend fun pubsubShardChannels(pattern: K): List<K>
 
     /**
      * Returns the number of subscribers (not counting clients subscribed to patterns) for the specified shard channels.

src/main/kotlin/io/lettuce/core/api/coroutines/BaseRedisCoroutinesCommandsImpl.kt

@@ -34,6 +34,7 @@ import kotlinx.coroutines.reactive.awaitSingle
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mikhael Sokolov
+ * @author Ali Takavci
  * @since 6.0
  */
 @ExperimentalLettuceCoroutinesApi
@@ -47,6 +48,10 @@ internal class BaseRedisCoroutinesCommandsImpl<K : Any, V : Any>(internal val op
 
     override suspend fun pubsubNumsub(vararg channels: K): Map<K, Long> = ops.pubsubNumsub(*channels).awaitSingle()
 
+    override suspend fun pubsubShardChannels(): List<K> = ops.pubsubShardChannels().asFlow().toList()
+
+    override suspend fun pubsubShardChannels(pattern: K): List<K> = ops.pubsubShardChannels(pattern).asFlow().toList()
+
     override suspend fun pubsubShardNumsub(vararg shardChannels: K): Map<K, Long> = ops.pubsubShardNumsub(*shardChannels).awaitSingle()
 
     override suspend fun pubsubNumpat(): Long = ops.pubsubNumpat().awaitSingle()

src/main/templates/io/lettuce/core/api/BaseRedisCommands.java

@@ -30,6 +30,7 @@
  * @param <V> Value type.
  * @author Mark Paluch
  * @author Tihomir Mateev
+ * @author Ali Takavci
  * @since 4.0
  */
 public interface BaseRedisCommands<K, V> {
@@ -66,6 +67,21 @@ public interface BaseRedisCommands<K, V> {
      */
     Map<K, Long> pubsubNumsub(K... channels);
 
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @return List&lt;K&gt; array-reply a list of active channels.
+     */
+    List<K> pubsubShardChannels();
+
+    /**
+     * Lists the currently *active shard channels*.
+     *
+     * @param pattern the pattern type: patternkey (pattern).
+     * @return List&lt;K&gt; array-reply a list of active channels, optionally matching the specified pattern.
+     */
+    List<K> pubsubShardChannels(K pattern);
+
     /**
      * Returns the number of subscribers (not counting clients subscribed to patterns) for the specified shard channels.
      *