Add support for SPUBLISH

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

@@ -1582,6 +1582,11 @@ public RedisFuture<Long> publish(K channel, V message) {
         return dispatch(commandBuilder.publish(channel, message));
     }
 
+    @Override
+    public RedisFuture<Long> spublish(K shardChannel, V message) {
+        return dispatch(commandBuilder.spublish(shardChannel, message));
+    }
+
     @Override
     public RedisFuture<List<K>> pubsubChannels() {
         return dispatch(commandBuilder.pubsubChannels());

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

@@ -1661,6 +1661,11 @@
         return createMono(() -> commandBuilder.publish(channel, message));
     }
 
+    @Override
+    public Mono<Long> spublish(K shardChannel, V message) {
+        return createMono(() -> commandBuilder.spublish(shardChannel, message));
+    }
+
     @Override
     public Flux<K> pubsubChannels() {
         return createDissolvingFlux(commandBuilder::pubsubChannels);

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

@@ -2156,6 +2156,13 @@ Command<K, V, Long> publish(K channel, V message) {
         return createCommand(PUBLISH, new IntegerOutput<>(codec), args);
     }
 
+    Command<K, V, Long> spublish(K shardChannel, V message) {
+        LettuceAssert.notNull(shardChannel, "ShardChannel " + MUST_NOT_BE_NULL);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).addKey(shardChannel).addValue(message);
+        return createCommand(SPUBLISH, new IntegerOutput<>(codec), args);
+    }
+
     Command<K, V, List<K>> pubsubChannels() {
         CommandArgs<K, V> args = new CommandArgs<>(codec).add(CHANNELS);
         return createCommand(PUBSUB, new KeyListOutput<>(codec), args);

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

@@ -48,6 +48,17 @@ public interface BaseRedisAsyncCommands<K, V> {
      */
     RedisFuture<Long> publish(K channel, V message);
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long integer-reply the number of clients that received the message.
+     * @since 7.0
+     */
+    RedisFuture<Long> spublish(K shardChannel, V message);
+
+
     /**
      * Lists the currently *active channels*.
      *

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

@@ -48,6 +48,17 @@ public interface BaseRedisReactiveCommands<K, V> {
      */
     Mono<Long> publish(K channel, V message);
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long integer-reply the number of clients that received the message.
+     * @since 7.0
+     */
+    Mono<Long> spublish(K shardChannel, V message);
+
+
     /**
      * Lists the currently *active channels*.
      *

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

@@ -47,6 +47,16 @@ public interface BaseRedisCommands<K, V> {
      */
     Long publish(K channel, V message);
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long integer-reply the number of clients that received the message.
+     * @since 7.0
+     */
+    Long spublish(K shardChannel, V message);
+
     /**
      * Lists the currently *active channels*.
      *

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 smessage(K shardChannel, V message) {
+            notifications.smessage(getNode(), shardChannel, message);
+        }
+
         @Override
         public void ssubscribed(K channel, long count) {
             notifications.ssubscribed(getNode(), channel, count);

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

@@ -88,6 +88,9 @@
             case unsubscribe:
                 multicast.unsubscribed(clusterNode, output.channel(), output.count());
                 break;
+            case smessage:
+                multicast.smessage(clusterNode, output.channel(), output.body());
+                break;
             case ssubscribe:
                 multicast.ssubscribed(clusterNode, output.channel(), output.count());
                 break;
@@ -192,6 +195,12 @@
             clusterListeners.forEach(listener -> listener.punsubscribed(node, pattern, count));
         }
 
+        @Override
+        public void smessage(RedisClusterNode node, K shardChannel, V message) {
+            getListeners().forEach(listener -> listener.smessage(shardChannel, message));
+            clusterListeners.forEach(listener -> listener.smessage(node, shardChannel, message));
+        }
+
         @Override
         public void ssubscribed(RedisClusterNode node, K channel, long count) {
             getListeners().forEach(listener -> listener.ssubscribed(channel, count));

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

@@ -48,6 +48,17 @@ public interface BaseNodeSelectionAsyncCommands<K, V> {
      */
     AsyncExecutions<Long> publish(K channel, V message);
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long integer-reply the number of clients that received the message.
+     * @since 7.0
+     */
+    AsyncExecutions<Long> spublish(K shardChannel, V message);
+
+
     /**
      * Lists the currently *active channels*.
      *

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

@@ -43,6 +43,17 @@ public interface BaseNodeSelectionCommands<K, V> {
      */
     Executions<Long> publish(K channel, V message);
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long integer-reply the number of clients that received the message.
+     * @since 7.0
+     */
+    Executions<Long> spublish(K shardChannel, V message);
+
+
     /**
      * Lists the currently *active channels*.
      *

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

@@ -42,6 +42,11 @@
         // empty adapter method
     }
 
+    @Override
+    public void smessage(RedisClusterNode node, K shardChannel, V message) {
+        // 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,6 +67,18 @@
      */
     void punsubscribed(RedisClusterNode node, K pattern, long count);
 
+    /**
+     * Message received from a shard channel subscription.
+     *
+     * @param node the {@link RedisClusterNode} from which the {@code message} originates.
+     * @param shardChannel shard channel.
+     * @param message Message.
+     * @since 7.0
+     */
+    default void smessage(RedisClusterNode node, K shardChannel, V message){
+        message(node, shardChannel, message);
+    }
+
     /**
      * Subscribed to a shard channel.
      *

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, SSUBSCRIBE,
+    PSUBSCRIBE, PUBLISH, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PUBSUB, SSUBSCRIBE, SPUBLISH,
 
     // Sets
 

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

@@ -58,6 +58,11 @@ Command<K, V, Long> publish(K channel, V message) {
         return createCommand(PUBLISH, new IntegerOutput<>(codec), args);
     }
 
+    Command<K, V, Long> spublish(K shardChannel, V message) {
+        CommandArgs<K, V> args = new CommandArgs<>(codec).addKey(shardChannel).addValue(message);
+        return createCommand(SPUBLISH, new IntegerOutput<>(codec), args);
+    }
+
     Command<K, V, List<K>> pubsubChannels(K pattern) {
         CommandArgs<K, V> args = new PubSubCommandArgs<>(codec).add(CHANNELS).addKey(pattern);
         return createCommand(PUBSUB, new KeyListOutput<>(codec), args);

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

@@ -263,6 +263,9 @@ protected void notifyListeners(PubSubMessage<K, V> message) {
                 case unsubscribe:
                     listener.unsubscribed(message.channel(), message.count());
                     break;
+                case smessage:
+                    listener.smessage(message.channel(), message.body());
+                    break;
                 case ssubscribe:
                     listener.ssubscribed(message.channel(), message.count());
                     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, ssubscribe;
+        message, pmessage, psubscribe, punsubscribe, subscribe, unsubscribe, ssubscribe, smessage;
 
         private final static Set<String> names = new HashSet<>();
 
@@ -108,6 +108,7 @@ private void handleOutput(ByteBuffer bytes) {
                     pattern = codec.decodeKey(bytes);
                     break;
                 }
+            case smessage:
             case message:
                 if (channel == null) {
                     channel = codec.decodeKey(bytes);

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

@@ -59,6 +59,11 @@
         // empty adapter method
     }
 
+    @Override
+    public void smessage(K shardChannel, V message) {  
+        // empty adapter method
+    }
+
     @Override
     public void ssubscribed(K shardChannel, long count) {
         // empty adapter method

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

@@ -80,6 +80,11 @@ public RedisFuture<Long> publish(K channel, V message) {
         return dispatch(commandBuilder.publish(channel, message));
     }
 
+    @Override
+    public RedisFuture<Long> spublish(K shardChannel, V message) {
+        return dispatch(commandBuilder.spublish(shardChannel, message));
+    }
+
     @Override
     public RedisFuture<List<K>> pubsubChannels(K channel) {
         return dispatch(commandBuilder.pubsubChannels(channel));

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

@@ -88,4 +88,15 @@
         subscribed(shardChannel, count);
     }
 
+    /**
+     * Message received from a shard channel subscription.
+     *
+     * @param shardChannel shard channel.
+     * @param message Message.
+     * @since 7.0
+     */
+    default void smessage(K shardChannel, V message) {
+        message(shardChannel, message);
+    }
+
 }

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

@@ -139,6 +139,11 @@
         return createMono(() -> commandBuilder.publish(channel, message));
     }
 
+    @Override
+    public Mono<Long> spublish(K shardChannel, V message) {
+        return createMono(() -> commandBuilder.publish(shardChannel, message));
+    }
+
     @Override
     public Flux<K> pubsubChannels(K channel) {
         return createDissolvingFlux(() -> commandBuilder.pubsubChannels(channel));

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

@@ -48,6 +48,17 @@ interface BaseRedisCoroutinesCommands<K : Any, V : Any> {
      */
     suspend fun publish(channel: K, message: V): Long?
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long integer-reply the number of clients that received the message.
+     * @since 7.0
+     */
+    suspend fun spublish(shardChannel: K, message: V): Long?
+
+
     /**
      * Lists the currently *active channels*.
      *

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

@@ -46,6 +46,8 @@
 
     override suspend fun publish(channel: K, message: V): Long? = ops.publish(channel, message).awaitFirstOrNull()
 
+    override suspend fun spublish(shardChannel: K, message: V): Long? = ops.spublish(shardChannel, message).awaitFirstOrNull()
+
     override suspend fun pubsubChannels(): List<K> = ops.pubsubChannels().asFlow().toList()
 
     override suspend fun pubsubChannels(channel: K): List<K> = ops.pubsubChannels(channel).asFlow().toList()

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

@@ -48,6 +48,16 @@ public interface BaseRedisCommands<K, V> {
      */
     Long publish(K channel, V message);
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long integer-reply the number of clients that received the message.
+     * @since 7.0
+     */
+    Long spublish(K shardChannel, V message);
+
     /**
      * Lists the currently *active channels*.
      *