Add support for SPUBLISH #2757

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

@@ -56,6 +56,7 @@
  * @author Tugdual Grall
  * @author dengliming
  * @author Andrey Shlykov
+ * @author Manish Manghwani
  */
 @SuppressWarnings("unchecked")
 public abstract class AbstractRedisAsyncCommands<K, V> implements RedisAclAsyncCommands<K, V>, RedisHashAsyncCommands<K, V>,
@@ -1536,6 +1537,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

@@ -65,6 +65,7 @@
  * @author Tugdual Grall
  * @author dengliming
  * @author Andrey Shlykov
+ * @author Manish Manghwani
  * @since 4.0
  */
 public abstract class AbstractRedisReactiveCommands<K, V>
@@ -1615,6 +1616,11 @@ public Mono<Long> publish(K channel, V message) {
         return createMono(() -> commandBuilder.publish(channel, message));
     }
 
+    @Override
+    public Mono<Long> spublish(K channel, V message) {
+        return createMono(() -> commandBuilder.spublish(channel, message));
+    }
+
     @Override
     public Flux<K> pubsubChannels() {
         return createDissolvingFlux(commandBuilder::pubsubChannels);

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

@@ -53,6 +53,7 @@
  * @author dengliming
  * @author Mikhael Sokolov
  * @author Tihomir Mateev
+ * @author Manish Manghwani
  */
 @SuppressWarnings({ "unchecked", "varargs" })
 class RedisCommandBuilder<K, V> extends BaseRedisCommandBuilder<K, V> {
@@ -2094,6 +2095,13 @@ Command<K, V, List<K>> pubsubChannels(K pattern) {
         return createCommand(PUBSUB, new KeyListOutput<>(codec), args);
     }
 
+    Command<K, V, Long> spublish(K shardChannel, V message) {
+        LettuceAssert.notNull(shardChannel, "Shard Channel " + 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, Long> pubsubNumpat() {
         CommandArgs<K, V> args = new CommandArgs<>(codec).add(NUMPAT);
         return createCommand(PUBSUB, new IntegerOutput<>(codec), args);

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 Manish Manghwani
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateAsyncApi
  */
@@ -43,6 +44,16 @@ 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 the number of clients that received the message. Note that in a Redis Cluster, only clients that are
+     * connected to the same node as the publishing client are included in the count.
+     */
+    RedisFuture<Long> spublish(K shardChannel, V message);
+
     /**
      * Lists the currently *active 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 Manish Manghwani
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateReactiveApi
  */
@@ -43,6 +44,16 @@ public interface BaseRedisReactiveCommands<K, V> {
      */
     Mono<Long> publish(K channel, V message);
 
+    /**
+     * Post a message to a shard channel.
+     *
+     * @param shardChannel the channel type: key.
+     * @param message the message type: value.
+     * @return Long the number of clients that received the message. Note that in a Redis Cluster, only clients that are
+     * connected to the same node as the publishing client are included in the count.
+     */
+    Mono<Long> spublish(K shardChannel, V message);
+
     /**
      * Lists the currently *active 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 Manish Manghwani
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateSyncApi
  */
@@ -42,6 +43,16 @@ public interface BaseRedisCommands<K, V> {
      */
     Long publish(K channel, V message);
 
+    /**
+     * Post a message to a channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long the number of clients that received the message. Note that in a Redis Cluster, only clients that are
+     * connected to the same node as the publishing client are included in the count.
+     */
+    Long spublish(K shardChannel, V message);
+
     /**
      * Lists the currently *active 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 Manish Manghwani
  * @since 4.0
  * @generated by io.lettuce.apigenerator.CreateSyncNodeSelectionClusterApi
  */
@@ -38,6 +39,16 @@ public interface BaseNodeSelectionCommands<K, V> {
      */
     Executions<Long> publish(K channel, V message);
 
+    /**
+     * Post a message to a channel.
+     *
+     * @param shardChannel the channel type: key.
+     * @param message the message type: value.
+     * @return Long the number of clients that received the message. Note that in a Redis Cluster, only clients that are
+     * connected to the same node as the publishing client are included in the count.
+     */
+    Executions<Long> spublish(K shardChannel, V message);
+
     /**
      * Lists the currently *active channels*.
      *

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

@@ -25,6 +25,7 @@
  * @author Zhang Jessey
  * @author dengliming
  * @author Mikhael Sokolov
+ * @author Manish Manghwani
  */
 public enum CommandType implements ProtocolKeyword {
 
@@ -70,7 +71,7 @@ public enum CommandType implements ProtocolKeyword {
 
     // Pub/Sub
 
-    PSUBSCRIBE, PUBLISH, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PUBSUB,
+    PSUBSCRIBE, PUBLISH, SPUBLISH, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PUBSUB,
 
     // Sets
 

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

@@ -37,6 +37,7 @@
  *
  * @author Mark Paluch
  * @author Tihomir Mateev
+ * @author Manish Manghwani
  * @since 4.2
  */
 @SuppressWarnings("varargs")
@@ -53,6 +54,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 PubSubCommandArgs<>(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/RedisPubSubAsyncCommandsImpl.java

@@ -30,6 +30,7 @@
  * @param <V> Value type.
  * @author Will Glozer
  * @author Mark Paluch
+ * @author Manish Manghwani
  */
 public class RedisPubSubAsyncCommandsImpl<K, V> extends RedisAsyncCommandsImpl<K, V> implements RedisPubSubAsyncCommands<K, V> {
 
@@ -75,6 +76,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/RedisPubSubReactiveCommandsImpl.java

@@ -32,6 +32,7 @@
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mark Paluch
+ * @author Manish Manghwani
  * @since 5.0
  */
 public class RedisPubSubReactiveCommandsImpl<K, V> extends RedisReactiveCommandsImpl<K, V>
@@ -133,6 +134,11 @@ public Mono<Long> publish(K channel, V message) {
         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(K channel) {
         return createDissolvingFlux(() -> commandBuilder.pubsubChannels(channel));

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 Manish Manghwani
  * @since 6.0
  * @generated by io.lettuce.apigenerator.CreateKotlinCoroutinesApi
  */
@@ -43,6 +44,16 @@ interface BaseRedisCoroutinesCommands<K : Any, V : Any> {
      */
     suspend fun publish(channel: K, message: V): Long?
 
+    /**
+     * Post a message to a channel.
+     *
+     * @param shardChannel the shard channel type: key.
+     * @param message the message type: value.
+     * @return Long the number of clients that received the message. Note that in a Redis Cluster, only clients that are
+     * connected to the same node as the publishing client are included in the count.
+     */
+    suspend fun spublish(shardChannel: K, message: V): Long?
+
     /**
      * Lists the currently *active channels*.
      *

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

@@ -34,13 +34,16 @@ import kotlinx.coroutines.reactive.awaitSingle
  * @param <K> Key type.
  * @param <V> Value type.
  * @author Mikhael Sokolov
+ * @author Manish Manghwani
  * @since 6.0
  */
 @ExperimentalLettuceCoroutinesApi
 internal class BaseRedisCoroutinesCommandsImpl<K : Any, V : Any>(internal val ops: BaseRedisReactiveCommands<K, V>) : BaseRedisCoroutinesCommands<K, V> {
 
     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()