Add support for SSUBSCRIBE

src/main/java/io/lettuce/core/cluster/ClusterPubSubConnectionProvider.java

@@ -167,6 +167,11 @@ public void punsubscribed(K pattern, long count) {
             notifications.punsubscribed(getNode(), pattern, count);
         }
 
+        @Override
+        public void ssubscribed(K channel, long count) {
+            notifications.ssubscribed(getNode(), channel, count);
+        }
+
         private RedisClusterNode getNode() {
             return nodeId != null ? getPartitions().getPartitionByNodeId(nodeId) : getPartitions().getPartition(host, port);
         }

src/main/java/io/lettuce/core/cluster/ClusterReadOnlyCommands.java

@@ -58,7 +58,7 @@ public static ReadOnlyCommands.ReadOnlyPredicate asPredicate() {
     enum CommandName {
 
         // Pub/Sub commands are no key-space commands so they are safe to execute on replica nodes
-        PUBLISH, PUBSUB, PSUBSCRIBE, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE
+        PUBLISH, PUBSUB, PSUBSCRIBE, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, SSUBSCRIBE
     }
 
 }

src/main/java/io/lettuce/core/cluster/PubSubClusterEndpoint.java

@@ -88,6 +88,9 @@
             case unsubscribe:
                 multicast.unsubscribed(clusterNode, output.channel(), output.count());
                 break;
+            case ssubscribe:
+                multicast.ssubscribed(clusterNode, output.channel(), output.count());
+                break;
             default:
                 throw new UnsupportedOperationException("Operation " + output.type() + " not supported");
         }
@@ -189,6 +192,12 @@
             clusterListeners.forEach(listener -> listener.punsubscribed(node, pattern, count));
         }
 
+        @Override
+        public void ssubscribed(RedisClusterNode node, K channel, long count) {
+            getListeners().forEach(listener -> listener.ssubscribed(channel, count));
+            clusterListeners.forEach(listener -> listener.ssubscribed(node, channel, count));
+        }
+
     }
 
 }

src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubAdapter.java

@@ -42,4 +42,9 @@
         // empty adapter method
     }
 
+    @Override
+    public void ssubscribed(RedisClusterNode node, K channel, long count) {
+        // empty adapter method
+    }
+
 }

src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubListener.java

@@ -67,4 +67,16 @@
      */
     void punsubscribed(RedisClusterNode node, K pattern, long count);
 
+    /**
+     * Subscribed to a shard channel.
+     *
+     * @param node the {@link RedisClusterNode} from which the {@code message} originates.
+     * @param shardChannel Shard channel
+     * @param count Subscription count.
+     */
+    default void ssubscribed(RedisClusterNode node, K shardChannel, long count){
+        throw new UnsupportedOperationException(
+                "This operation is not available by default! Use one of available pub/sub adapters or override this method in your class.");
+    }
+
 }

src/main/java/io/lettuce/core/cluster/pubsub/api/async/NodeSelectionPubSubAsyncCommands.java

@@ -42,4 +42,11 @@ public interface NodeSelectionPubSubAsyncCommands<K, V> {
      */
     AsyncExecutions<Void> unsubscribe(K... channels);
 
+    /**
+     * Listen for messages published to the given shard channels.
+     *
+     * @param shardChannels the channels
+     * @return RedisFuture&lt;Void&gt; Future to synchronize {@code subscribe} completion
+     */
+    AsyncExecutions<Void> ssubscribe(K... shardChannels);
 }

src/main/java/io/lettuce/core/cluster/pubsub/api/reactive/NodeSelectionPubSubReactiveCommands.java

@@ -42,4 +42,12 @@ public interface NodeSelectionPubSubReactiveCommands<K, V> {
      */
     ReactiveExecutions<Void> unsubscribe(K... channels);
 
+        /**
+     * Listen for messages published to the given shard channels.
+     *
+     * @param shardCchannels the channels
+     * @return RedisFuture&lt;Void&gt; Future to synchronize {@code subscribe} completion
+     */
+    ReactiveExecutions<Void> ssubscribe(K... shardCchannels);
+
 }

src/main/java/io/lettuce/core/cluster/pubsub/api/sync/NodeSelectionPubSubCommands.java

@@ -42,4 +42,12 @@ public interface NodeSelectionPubSubCommands<K, V> {
      */
     Executions<Void> unsubscribe(K... channels);
 
+        /**
+     * Listen for messages published to the given shard channels.
+     *
+     * @param shardChannels the channels
+     * @return Executions Future to synchronize {@code subscribe} completion
+     */
+    Executions<Void> ssubscribe(K... shardChannels);
+
 }

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

@@ -74,7 +74,7 @@ public enum CommandType implements ProtocolKeyword {
 
     // Pub/Sub
 
-    PSUBSCRIBE, PUBLISH, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PUBSUB,
+    PSUBSCRIBE, PUBLISH, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PUBSUB, SSUBSCRIBE,
 
     // Sets
 

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

@@ -110,6 +110,14 @@ final Command<K, V, V> unsubscribe(K... channels) {
         return pubSubCommand(UNSUBSCRIBE, new PubSubOutput<>(codec), channels);
     }
 
+    @SafeVarargs
+    final Command<K, V, V> ssubscribe(K... shardChannels) {
+        LettuceAssert.notEmpty(shardChannels, "Shard channels " + MUST_NOT_BE_EMPTY);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).addKeys(shardChannels);
+        return createCommand(SSUBSCRIBE, new PubSubOutput<>(codec), args);
+    }
+
     @SafeVarargs
     final <T> Command<K, V, T> pubSubCommand(CommandType type, CommandOutput<K, V, T> output, K... keys) {
         return new Command<>(type, output, new PubSubCommandArgs<>(codec).addKeys(keys));

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

@@ -218,6 +218,9 @@ private boolean shouldCompleteCommand(PubSubOutput.Type type, RedisCommand<?, ?,
             case subscribe:
                 return commandType.equalsIgnoreCase("SUBSCRIBE");
 
+            case ssubscribe:
+                return commandType.equalsIgnoreCase("SSUBSCRIBE");
+
             case psubscribe:
                 return commandType.equalsIgnoreCase("PSUBSCRIBE");
 

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

@@ -27,6 +27,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
 
 import io.lettuce.core.ClientOptions;
 import io.lettuce.core.ConnectionState;
@@ -56,6 +57,8 @@ public class PubSubEndpoint<K, V> extends DefaultEndpoint {
 
     private final Set<Wrapper<K>> channels;
 
+    private final Set<Wrapper<K>> shardChannels;
+
     private final Set<Wrapper<K>> patterns;
 
     private volatile boolean subscribeWritten = false;
@@ -70,13 +73,15 @@ public class PubSubEndpoint<K, V> extends DefaultEndpoint {
         ALLOWED_COMMANDS_SUBSCRIBED.add(CommandType.PSUBSCRIBE.name());
         ALLOWED_COMMANDS_SUBSCRIBED.add(CommandType.UNSUBSCRIBE.name());
         ALLOWED_COMMANDS_SUBSCRIBED.add(CommandType.PUNSUBSCRIBE.name());
+        ALLOWED_COMMANDS_SUBSCRIBED.add(CommandType.SSUBSCRIBE.name());
         ALLOWED_COMMANDS_SUBSCRIBED.add(CommandType.QUIT.name());
         ALLOWED_COMMANDS_SUBSCRIBED.add(CommandType.PING.name());
 
         SUBSCRIBE_COMMANDS = new HashSet<>(2, 1);
 
         SUBSCRIBE_COMMANDS.add(CommandType.SUBSCRIBE.name());
         SUBSCRIBE_COMMANDS.add(CommandType.PSUBSCRIBE.name());
+        SUBSCRIBE_COMMANDS.add(CommandType.SSUBSCRIBE.name());
     }
 
     /**
@@ -91,6 +96,7 @@ public PubSubEndpoint(ClientOptions clientOptions, ClientResources clientResourc
 
         this.channels = ConcurrentHashMap.newKeySet();
         this.patterns = ConcurrentHashMap.newKeySet();
+        this.shardChannels = ConcurrentHashMap.newKeySet();
     }
 
     /**
@@ -257,6 +263,9 @@ protected void notifyListeners(PubSubMessage<K, V> message) {
                 case unsubscribe:
                     listener.unsubscribed(message.channel(), message.count());
                     break;
+                case ssubscribe:
+                    listener.ssubscribed(message.channel(), message.count());
+                    break;
                 default:
                     throw new UnsupportedOperationException("Operation " + message.type() + " not supported");
             }
@@ -278,6 +287,9 @@ private void updateInternalState(PubSubMessage<K, V> message) {
             case unsubscribe:
                 channels.remove(new Wrapper<>(message.channel()));
                 break;
+            case ssubscribe:
+                shardChannels.add(new Wrapper<>(message.channel()));
+                break;
             default:
                 break;
         }

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

@@ -38,7 +38,7 @@ public class PubSubOutput<K, V> extends CommandOutput<K, V, V> implements PubSub
 
     public enum Type {
 
-        message, pmessage, psubscribe, punsubscribe, subscribe, unsubscribe;
+        message, pmessage, psubscribe, punsubscribe, subscribe, unsubscribe, ssubscribe;
 
         private final static Set<String> names = new HashSet<>();
 
@@ -122,6 +122,7 @@ private void handleOutput(ByteBuffer bytes) {
                 break;
             case subscribe:
             case unsubscribe:
+            case ssubscribe:
                 channel = codec.decodeKey(bytes);
                 break;
             default:

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

@@ -59,4 +59,9 @@
         // empty adapter method
     }
 
+    @Override
+    public void ssubscribed(K shardChannel, long count) {
+        // empty adapter method
+    }
+
 }

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

@@ -100,6 +100,12 @@ public RedisFuture<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         return dispatch(commandBuilder.pubsubShardNumsub(shardChannels));
     }
 
+    @Override
+    @SuppressWarnings("unchecked")
+    public RedisFuture<Void> ssubscribe(K... channels) {
+        return (RedisFuture<Void>) dispatch(commandBuilder.ssubscribe(channels));
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     public StatefulRedisPubSubConnection<K, V> getStatefulConnection() {

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

@@ -77,4 +77,14 @@
      */
     void punsubscribed(K pattern, long count);
 
+    /**
+     * Subscribed to a Shard channel.
+     *
+     * @param shardChannel Shard channel
+     * @param count Subscription count.
+     */
+    default void ssubscribed(K chashardChannelnnel, long count) {
+        throw new UnsupportedOperationException(
+                "This operation is not available by default. Use one of available pub/sub adapters or override this method in your class!");
+    }
 }

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

@@ -26,6 +26,7 @@
 import reactor.core.publisher.Mono;
 import io.lettuce.core.RedisReactiveCommandsImpl;
 import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.protocol.Command;
 import io.lettuce.core.pubsub.api.reactive.ChannelMessage;
 import io.lettuce.core.pubsub.api.reactive.PatternMessage;
 import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands;
@@ -158,6 +159,15 @@ public Mono<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         return createMono(() -> commandBuilder.pubsubShardNumsub(shardChannels));
     }
 
+    @Override
+    public Mono<Void> ssubscribe(K... shardChannels) {
+        return createFlux(() -> {
+            Command<K, V, V> c = commandBuilder.ssubscribe(shardChannels);
+            System.out.println("");
+            return c;
+        }).then();
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     public StatefulRedisPubSubConnection<K, V> getStatefulConnection() {

src/main/java/io/lettuce/core/pubsub/api/async/RedisPubSubAsyncCommands.java

@@ -51,4 +51,12 @@ public interface RedisPubSubAsyncCommands<K, V> extends RedisAsyncCommands<K, V>
      */
     StatefulRedisPubSubConnection<K, V> getStatefulConnection();
 
+    /**
+     * Listen for messages published to the given shard channels.
+     *
+     * @param shardChannels the shard channels
+     * @return RedisFuture&lt;Void&gt; Future to synchronize {@code subscribe} completion
+     */
+    RedisFuture<Void> ssubscribe(K... shardChannels);
+
 }

src/main/java/io/lettuce/core/pubsub/api/reactive/RedisPubSubReactiveCommands.java

@@ -97,6 +97,15 @@ public interface RedisPubSubReactiveCommands<K, V> extends RedisReactiveCommands
      */
     Mono<Void> unsubscribe(K... channels);
 
+    /**
+     * Listen for messages published to the given shard channels. The {@link Mono} completes without a result as soon as the *
+     * subscription is registered.
+     *
+     * @param shardChannels the channels.
+     * @return Mono&lt;Void&gt; Mono for {@code subscribe} command.
+     */
+    Mono<Void> ssubscribe(K... shardChannels);
+
     /**
      * @return the underlying connection.
      * @since 6.2, will be removed with Lettuce 7 to avoid exposing the underlying connection.

src/main/java/io/lettuce/core/pubsub/api/sync/RedisPubSubCommands.java

@@ -41,6 +41,13 @@ public interface RedisPubSubCommands<K, V> extends RedisCommands<K, V> {
      */
     void unsubscribe(K... channels);
 
+    /**
+     * Listen for messages published to the given shard channels.
+     *
+     * @param shardChannels the channels
+     */
+    void ssubscribe(K... shardChannels);
+
     /**
      * @return the underlying connection.
      */

src/test/java/io/lettuce/core/cluster/ClusterReadOnlyCommandsUnitTests.java

@@ -16,7 +16,7 @@ class ClusterReadOnlyCommandsUnitTests {
 
     @Test
     void testCount() {
-        assertThat(ClusterReadOnlyCommands.getReadOnlyCommands()).hasSize(85);
+        assertThat(ClusterReadOnlyCommands.getReadOnlyCommands()).hasSize(86);
     }
 
     @Test
@@ -26,4 +26,5 @@ void testResolvableCommandNames() {
             assertThat(readOnlyCommand.name()).isEqualTo(CommandType.valueOf(readOnlyCommand.name()).name());
         }
     }
+
 }