Add support for SPUBLISH (#2838)

* implementation of SPUBLISH

* sort methods by name

* test cluster spublish with no redirects

* use injected cluster client in RedisClusterPubSubConnectionIntegrationTests

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

@@ -1295,7 +1295,7 @@ public RedisFuture<StreamScanCursor> hscan(KeyValueStreamingChannel<K, V> channe
 
     @Override
     public RedisFuture<StreamScanCursor> hscanNovalues(KeyStreamingChannel<K> channel, K key, ScanCursor scanCursor,
-                                                       ScanArgs scanArgs) {
+            ScanArgs scanArgs) {
         return dispatch(commandBuilder.hscanNoValuesStreaming(channel, key, scanCursor, scanArgs));
     }
 
@@ -2069,6 +2069,11 @@ public RedisFuture<Set<V>> spop(K key, long count) {
         return dispatch(commandBuilder.spop(key, count));
     }
 
+    @Override
+    public RedisFuture<Long> spublish(K shardChannel, V message) {
+        return dispatch(commandBuilder.spublish(shardChannel, message));
+    }
+
     @Override
     public RedisFuture<V> srandmember(K key) {
         return dispatch(commandBuilder.srandmember(key));

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

@@ -2147,6 +2147,11 @@ public Flux<V> spop(K key, long count) {
         return createDissolvingFlux(() -> commandBuilder.spop(key, count));
     }
 
+    @Override
+    public Mono<Long> spublish(K shardChannel, V message) {
+        return createMono(() -> commandBuilder.spublish(shardChannel, message));
+    }
+
     @Override
     public Mono<V> srandmember(K key) {
         return createMono(() -> commandBuilder.srandmember(key));

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

@@ -2754,6 +2754,13 @@ Command<K, V, Set<V>> spop(K key, long count) {
         return createCommand(SPOP, new ValueSetOutput<>(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, V> srandmember(K key) {
         notNullKey(key);
 

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

@@ -102,6 +102,16 @@ public interface BaseRedisAsyncCommands<K, V> {
      */
     RedisFuture<Long> pubsubNumpat();
 
+    /**
+     * 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);
+
     /**
      * Echo the given string.
      *

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

@@ -102,6 +102,16 @@ public interface BaseRedisReactiveCommands<K, V> {
      */
     Mono<Long> pubsubNumpat();
 
+    /**
+     * 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);
+
     /**
      * Echo the given string.
      *

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

@@ -101,6 +101,16 @@ public interface BaseRedisCommands<K, V> {
      */
     Long pubsubNumpat();
 
+    /**
+     * 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);
+
     /**
      * Echo the given string.
      *

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 @@ protected void notifyListeners(PubSubMessage<K, V> output) {
             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 @@ public void punsubscribed(RedisClusterNode node, K pattern, long count) {
             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

@@ -102,6 +102,16 @@ public interface BaseNodeSelectionAsyncCommands<K, V> {
      */
     AsyncExecutions<Long> pubsubNumpat();
 
+    /**
+     * 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);
+
     /**
      * Echo the given string.
      *

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

@@ -97,6 +97,16 @@ public interface BaseNodeSelectionCommands<K, V> {
      */
     Executions<Long> pubsubNumpat();
 
+    /**
+     * 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);
+
     /**
      * Echo the given string.
      *

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

@@ -42,6 +42,11 @@ public void punsubscribed(RedisClusterNode node, K pattern, long count) {
         // 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 @@ public interface RedisClusterPubSubListener<K, V> {
      */
     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

@@ -98,6 +98,19 @@ final Command<K, V, V> punsubscribe(K... patterns) {
         return pubSubCommand(PUNSUBSCRIBE, new PubSubOutput<>(codec), patterns);
     }
 
+    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);
+    }
+
+    @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 Command<K, V, V> subscribe(K... channels) {
         LettuceAssert.notEmpty(channels, "Channels " + MUST_NOT_BE_EMPTY);
@@ -110,14 +123,6 @@ 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/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 @@ public void punsubscribed(K pattern, long count) {
         // 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

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

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

@@ -88,4 +88,15 @@ default void ssubscribed(K shardChannel, long count) {
         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

@@ -159,6 +159,11 @@ public Mono<Map<K, Long>> pubsubShardNumsub(K... shardChannels) {
         return createMono(() -> commandBuilder.pubsubShardNumsub(shardChannels));
     }
 
+    @Override
+    public Mono<Long> spublish(K shardChannel, V message) {
+        return createMono(() -> commandBuilder.publish(shardChannel, message));
+    }
+
     @Override
     public Mono<Void> ssubscribe(K... shardChannels) {
         return createFlux(() -> commandBuilder.ssubscribe(shardChannels)).then();

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

@@ -102,6 +102,16 @@ interface BaseRedisCoroutinesCommands<K : Any, V : Any> {
      */
     suspend fun pubsubNumpat(): 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?
+
     /**
      * Echo the given string.
      *

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

@@ -60,6 +60,8 @@ internal class BaseRedisCoroutinesCommandsImpl<K : Any, V : Any>(internal val op
 
     override suspend fun pubsubNumpat(): Long = ops.pubsubNumpat().awaitSingle()
 
+    override suspend fun spublish(shardChannel: K, message: V): Long? = ops.spublish(shardChannel, message).awaitFirstOrNull()
+
     override suspend fun echo(msg: V): V = ops.echo(msg).awaitSingle()
 
     override suspend fun role(): List<Any> = ops.role().asFlow().toList()

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

@@ -102,6 +102,16 @@ public interface BaseRedisCommands<K, V> {
      */
     Long pubsubNumpat();
 
+    /**
+     * 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);
+
     /**
      * Echo the given string.
      *