[Commands API interface consistency #3] Fix node-selection aggregate interface wiring

pom.xml

@@ -495,6 +495,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-reflect</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <!-- Logging -->
 
         <dependency>

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

@@ -478,7 +478,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param hGetExArgs hgetex arguments.
      * @param fields fields to retrieve.
      * @return Long the number of fields that were removed from the hash.
+     * @deprecated since 7.7 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hgetex}.
      */
+    @Deprecated
     Mono<Long> hgetex(KeyValueStreamingChannel<K, V> channel, K key, HGetExArgs hGetExArgs, K... fields);
 
     /**
@@ -497,7 +500,10 @@ public interface RedisHashReactiveCommands<K, V> {
      * @param key the key.
      * @param fields fields to retrieve and delete.
      * @return Long the number of fields that were removed from the hash.
+     * @deprecated since 7.7 in favor of consuming large results through the {@link org.reactivestreams.Publisher} returned by
+     *             {@link #hgetdel}.
      */
+    @Deprecated
     Mono<Long> hgetdel(KeyValueStreamingChannel<K, V> channel, K key, K... fields);
 
     /**

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

@@ -1,7 +1,6 @@
 package io.lettuce.core.cluster.api.async;
 
 import io.lettuce.core.cluster.api.NodeSelectionSupport;
-import io.lettuce.core.cluster.api.sync.NodeSelectionStreamCommands;
 
 /**
  * Asynchronous and thread-safe Redis API to execute commands on a {@link NodeSelectionSupport}.
@@ -10,12 +9,13 @@
  * @author Tihomir Mateev
  * @author Yordan Tsintsov
  */
-public interface NodeSelectionAsyncCommands<K, V> extends BaseNodeSelectionAsyncCommands<K, V>,
-        NodeSelectionFunctionAsyncCommands<K, V>, NodeSelectionGeoAsyncCommands<K, V>, NodeSelectionHashAsyncCommands<K, V>,
-        NodeSelectionHLLAsyncCommands<K, V>, NodeSelectionKeyAsyncCommands<K, V>, NodeSelectionListAsyncCommands<K, V>,
-        NodeSelectionScriptingAsyncCommands<K, V>, NodeSelectionServerAsyncCommands<K, V>, NodeSelectionSetAsyncCommands<K, V>,
-        NodeSelectionSortedSetAsyncCommands<K, V>, NodeSelectionStreamCommands<K, V>, NodeSelectionStringAsyncCommands<K, V>,
-        NodeSelectionJsonAsyncCommands<K, V>, NodeSelectionVectorSetAsyncCommands<K, V>, NodeSelectionSearchAsyncCommands<K, V>,
-        NodeSelectionBloomFilterAsyncCommands<K, V>, NodeSelectionCuckooFilterAsyncCommands<K, V>,
-        NodeSelectionTopKAsyncCommands<K, V> {
+public interface NodeSelectionAsyncCommands<K, V>
+        extends BaseNodeSelectionAsyncCommands<K, V>, NodeSelectionAclAsyncCommands<K, V>,
+        NodeSelectionArrayAsyncCommands<K, V>, NodeSelectionFunctionAsyncCommands<K, V>, NodeSelectionGeoAsyncCommands<K, V>,
+        NodeSelectionHashAsyncCommands<K, V>, NodeSelectionHLLAsyncCommands<K, V>, NodeSelectionKeyAsyncCommands<K, V>,
+        NodeSelectionListAsyncCommands<K, V>, NodeSelectionScriptingAsyncCommands<K, V>, NodeSelectionServerAsyncCommands<K, V>,
+        NodeSelectionSetAsyncCommands<K, V>, NodeSelectionSortedSetAsyncCommands<K, V>, NodeSelectionStreamAsyncCommands<K, V>,
+        NodeSelectionStringAsyncCommands<K, V>, NodeSelectionJsonAsyncCommands<K, V>, NodeSelectionVectorSetAsyncCommands<K, V>,
+        NodeSelectionSearchAsyncCommands<K, V>, NodeSelectionBloomFilterAsyncCommands<K, V>,
+        NodeSelectionCuckooFilterAsyncCommands<K, V>, NodeSelectionTopKAsyncCommands<K, V> {
 }

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

@@ -86,6 +86,7 @@ public interface RedisAdvancedClusterAsyncCommands<K, V> extends RedisClusterAsy
      * @return API with asynchronous executed commands on a selection of upstream cluster nodes.
      * @deprecated since 6.0 in favor of {@link #upstream()}.
      */
+    @Deprecated
     default AsyncNodeSelection<K, V> masters() {
         return nodes(redisClusterNode -> redisClusterNode.is(RedisClusterNode.NodeFlag.UPSTREAM));
     }

src/main/java/io/lettuce/core/cluster/api/reactive/RedisAdvancedClusterReactiveCommands.java

@@ -206,6 +206,7 @@ public interface RedisAdvancedClusterReactiveCommands<K, V> extends RedisCluster
      * @return K array-reply list of keys matching {@code pattern}.
      * @deprecated Use {@link #keys(String)} instead. This legacy overload will be removed in a later version.
      */
+    @Deprecated
     Flux<K> keysLegacy(K pattern);
 
     /**

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

@@ -8,11 +8,12 @@
  * @author Mark Paluch
  * @author Tihomir Mateev
  */
-public interface NodeSelectionCommands<K, V> extends BaseNodeSelectionCommands<K, V>, NodeSelectionFunctionCommands<K, V>,
-        NodeSelectionGeoCommands<K, V>, NodeSelectionHashCommands<K, V>, NodeSelectionHLLCommands<K, V>,
-        NodeSelectionKeyCommands<K, V>, NodeSelectionListCommands<K, V>, NodeSelectionScriptingCommands<K, V>,
-        NodeSelectionServerCommands<K, V>, NodeSelectionSetCommands<K, V>, NodeSelectionSortedSetCommands<K, V>,
-        NodeSelectionStreamCommands<K, V>, NodeSelectionStringCommands<K, V>, NodeSelectionJsonCommands<K, V>,
-        NodeSelectionVectorSetCommands<K, V>, NodeSelectionSearchCommands<K, V>, NodeSelectionBloomFilterCommands<K, V>,
-        NodeSelectionCuckooFilterCommands<K, V>, NodeSelectionTopKCommands<K, V> {
+public interface NodeSelectionCommands<K, V>
+        extends BaseNodeSelectionCommands<K, V>, NodeSelectionAclCommands<K, V>, NodeSelectionArrayCommands<K, V>,
+        NodeSelectionFunctionCommands<K, V>, NodeSelectionGeoCommands<K, V>, NodeSelectionHashCommands<K, V>,
+        NodeSelectionHLLCommands<K, V>, NodeSelectionKeyCommands<K, V>, NodeSelectionListCommands<K, V>,
+        NodeSelectionScriptingCommands<K, V>, NodeSelectionServerCommands<K, V>, NodeSelectionSetCommands<K, V>,
+        NodeSelectionSortedSetCommands<K, V>, NodeSelectionStreamCommands<K, V>, NodeSelectionStringCommands<K, V>,
+        NodeSelectionJsonCommands<K, V>, NodeSelectionVectorSetCommands<K, V>, NodeSelectionSearchCommands<K, V>,
+        NodeSelectionBloomFilterCommands<K, V>, NodeSelectionCuckooFilterCommands<K, V>, NodeSelectionTopKCommands<K, V> {
 }

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

@@ -54,7 +54,8 @@ interface RedisCoroutinesCommands<K : Any, V : Any> :
     RedisArrayCoroutinesCommands<K, V>,
     RedisBloomFilterCoroutinesCommands<K, V>,
     RedisCuckooFilterCoroutinesCommands<K, V>,
-    RedisTopKCoroutinesCommands<K, V> {
+    RedisTopKCoroutinesCommands<K, V>,
+    RediSearchCoroutinesCommands<K, V> {
 
     /**
      * Authenticate to the server.

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

@@ -280,6 +280,26 @@ interface RedisHashCoroutinesCommands<K : Any, V : Any> {
      */
     suspend fun hsetex(key: K, hSetExArgs: HSetExArgs, map: Map<K, V>): Long?
 
+    /**
+     * Set the value of one or more fields of a given hash key.
+     *
+     * @param key the key of the hash.
+     * @param map the field/value pairs to update.
+     * @return Long long-reply: 0 if no fields were set, 1 if all the fields were set
+     * @since 7.7
+     */
+    suspend fun hsetex(key: K, map: Map<K, V>): Long?
+
+    /**
+     * Get the value of one or more fields of a given hash key.
+     *
+     * @param key the key of the hash.
+     * @param fields fields to retrieve.
+     * @return List<KeyValue<K, V>> array-reply list of fields and their values.
+     * @since 7.7
+     */
+    fun hgetex(key: K, vararg fields: K): Flow<KeyValue<K, V>>
+
     /**
      * Get the value of one or more fields of a given hash key, and optionally set their expiration
      *

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

@@ -65,6 +65,10 @@ internal class RedisHashCoroutinesCommandsImpl<K : Any, V : Any>(internal val op
     override suspend fun hsetex(key: K, hSetExArgs: HSetExArgs, map: Map<K, V>): Long? =
         ops.hsetex(key, hSetExArgs, map).awaitFirstOrNull()
 
+    override suspend fun hsetex(key: K, map: Map<K, V>): Long? = ops.hsetex(key, map).awaitFirstOrNull()
+
+    override fun hgetex(key: K, vararg fields: K): Flow<KeyValue<K, V>> = ops.hgetex(key, *fields).asFlow()
+
     override fun hgetex(key: K, hGetExArgs: HGetExArgs, vararg fields: K): Flow<KeyValue<K, V>> = ops.hgetex(key, hGetExArgs, *fields).asFlow()
 
     override fun hgetdel(key: K, vararg fields: K): Flow<KeyValue<K, V>> = ops.hgetdel(key, *fields).asFlow()

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

@@ -25,6 +25,7 @@ import io.lettuce.core.XReadArgs.StreamOffset
 import io.lettuce.core.models.stream.ClaimedMessages
 import io.lettuce.core.models.stream.PendingMessage
 import io.lettuce.core.models.stream.PendingMessages
+import io.lettuce.core.models.stream.StreamEntryDeletionResult
 import kotlinx.coroutines.flow.Flow
 
 /**
@@ -173,6 +174,50 @@ interface RedisStreamCoroutinesCommands<K : Any, V : Any> {
      */
     suspend fun xdel(key: K, vararg messageIds: String): Long?
 
+    /**
+     * Acknowledge and delete one or multiple messages for a consumer group.
+     *
+     * @param key the stream key.
+     * @param group name of the consumer group.
+     * @param messageIds message Ids to acknowledge and delete.
+     * @return simple-reply an array of deletion results, one per message id.
+     * @since 7.7
+     */
+    fun xackdel(key: K, group: K, vararg messageIds: String): Flow<StreamEntryDeletionResult>
+
+    /**
+     * Acknowledge and delete one or multiple messages for a consumer group applying the given deletion policy.
+     *
+     * @param key the stream key.
+     * @param group name of the consumer group.
+     * @param policy the deletion policy to apply.
+     * @param messageIds message Ids to acknowledge and delete.
+     * @return simple-reply an array of deletion results, one per message id.
+     * @since 7.7
+     */
+    fun xackdel(key: K, group: K, policy: StreamDeletionPolicy, vararg messageIds: String): Flow<StreamEntryDeletionResult>
+
+    /**
+     * Removes the specified entries from the stream.
+     *
+     * @param key the stream key.
+     * @param messageIds stream entry IDs to delete.
+     * @return simple-reply an array of deletion results, one per message id.
+     * @since 7.7
+     */
+    fun xdelex(key: K, vararg messageIds: String): Flow<StreamEntryDeletionResult>
+
+    /**
+     * Removes the specified entries from the stream applying the given deletion policy.
+     *
+     * @param key the stream key.
+     * @param policy the deletion policy to apply.
+     * @param messageIds stream entry IDs to delete.
+     * @return simple-reply an array of deletion results, one per message id.
+     * @since 7.7
+     */
+    fun xdelex(key: K, policy: StreamDeletionPolicy, vararg messageIds: String): Flow<StreamEntryDeletionResult>
+
     /**
      * Create a consumer group.
      *

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

@@ -26,6 +26,7 @@ import io.lettuce.core.api.reactive.RedisStreamReactiveCommands
 import io.lettuce.core.models.stream.ClaimedMessages
 import io.lettuce.core.models.stream.PendingMessage
 import io.lettuce.core.models.stream.PendingMessages
+import io.lettuce.core.models.stream.StreamEntryDeletionResult
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.reactive.asFlow
@@ -66,6 +67,22 @@ internal class RedisStreamCoroutinesCommandsImpl<K : Any, V : Any>(internal val
 
     override suspend fun xdel(key: K, vararg messageIds: String): Long? = ops.xdel(key, *messageIds).awaitFirstOrNull()
 
+    override fun xackdel(key: K, group: K, vararg messageIds: String): Flow<StreamEntryDeletionResult> =
+        ops.xackdel(key, group, *messageIds).asFlow()
+
+    override fun xackdel(
+        key: K,
+        group: K,
+        policy: StreamDeletionPolicy,
+        vararg messageIds: String
+    ): Flow<StreamEntryDeletionResult> = ops.xackdel(key, group, policy, *messageIds).asFlow()
+
+    override fun xdelex(key: K, vararg messageIds: String): Flow<StreamEntryDeletionResult> =
+        ops.xdelex(key, *messageIds).asFlow()
+
+    override fun xdelex(key: K, policy: StreamDeletionPolicy, vararg messageIds: String): Flow<StreamEntryDeletionResult> =
+        ops.xdelex(key, policy, *messageIds).asFlow()
+
     override suspend fun xgroupCreate(streamOffset: StreamOffset<K>, group: K): String? = ops.xgroupCreate(streamOffset, group).awaitFirstOrNull()
 
     override suspend fun xgroupCreate(streamOffset: StreamOffset<K>, group: K, args: XGroupCreateArgs): String? = ops.xgroupCreate(streamOffset, group, args).awaitFirstOrNull()

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

@@ -176,6 +176,57 @@ interface RedisStringCoroutinesCommands<K : Any, V : Any> {
      */
     suspend fun bitopXor(destination: K, vararg keys: K): Long?
 
+    /**
+     * Perform bitwise DIFF between strings. Members of the source key that are not members of any of the other keys.
+     * Equivalent to: X ∧ ¬(Y1 ∨ Y2 ∨ …)
+     *
+     * @param destination result key of the operation.
+     * @param sourceKey the source key (X) for comparison.
+     * @param keys one or more additional keys (Y1, Y2, ...). At least one key is required.
+     * @return Long integer-reply The size of the string stored in the destination key, that is equal to the size of the
+     *         longest input string.
+     * @since 7.7
+     */
+    suspend fun bitopDiff(destination: K, sourceKey: K, vararg keys: K): Long?
+
+    /**
+     * Perform bitwise DIFF1 between strings. Members of one or more of the keys that are not members of the source key.
+     * Equivalent to: ¬X ∧ (Y1 ∨ Y2 ∨ …)
+     *
+     * @param destination result key of the operation.
+     * @param sourceKey the source key (X) for comparison.
+     * @param keys one or more additional keys (Y1, Y2, ...). At least one key is required.
+     * @return Long integer-reply The size of the string stored in the destination key, that is equal to the size of the
+     *         longest input string.
+     * @since 7.7
+     */
+    suspend fun bitopDiff1(destination: K, sourceKey: K, vararg keys: K): Long?
+
+    /**
+     * Perform bitwise ANDOR between strings. Members of the source key that are also members of one or more of the other
+     * keys. Equivalent to: X ∧ (Y1 ∨ Y2 ∨ …)
+     *
+     * @param destination result key of the operation.
+     * @param sourceKey the source key (X) for comparison.
+     * @param keys one or more additional keys (Y1, Y2, ...). At least one key is required.
+     * @return Long integer-reply The size of the string stored in the destination key, that is equal to the size of the
+     *         longest input string.
+     * @since 7.7
+     */
+    suspend fun bitopAndor(destination: K, sourceKey: K, vararg keys: K): Long?
+
+    /**
+     * Perform bitwise ONE between strings. Members of exactly one of the given keys. For two keys this is equivalent to XOR.
+     * For more than two keys, returns members that appear in exactly one key.
+     *
+     * @param destination result key of the operation.
+     * @param keys operation input key names.
+     * @return Long integer-reply The size of the string stored in the destination key, that is equal to the size of the
+     *         longest input string.
+     * @since 7.7
+     */
+    suspend fun bitopOne(destination: K, vararg keys: K): Long?
+
     /**
      * Decrement the integer value of a key by one.
      *

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

@@ -63,6 +63,18 @@ internal class RedisStringCoroutinesCommandsImpl<K : Any, V : Any>(internal val
     override suspend fun bitopXor(destination: K, vararg keys: K): Long? =
         ops.bitopXor(destination, *keys).awaitFirstOrNull()
 
+    override suspend fun bitopDiff(destination: K, sourceKey: K, vararg keys: K): Long? =
+        ops.bitopDiff(destination, sourceKey, *keys).awaitFirstOrNull()
+
+    override suspend fun bitopDiff1(destination: K, sourceKey: K, vararg keys: K): Long? =
+        ops.bitopDiff1(destination, sourceKey, *keys).awaitFirstOrNull()
+
+    override suspend fun bitopAndor(destination: K, sourceKey: K, vararg keys: K): Long? =
+        ops.bitopAndor(destination, sourceKey, *keys).awaitFirstOrNull()
+
+    override suspend fun bitopOne(destination: K, vararg keys: K): Long? =
+        ops.bitopOne(destination, *keys).awaitFirstOrNull()
+
     override suspend fun decr(key: K): Long? = ops.decr(key).awaitFirstOrNull()
 
     override suspend fun decrby(key: K, amount: Long): Long? =

src/main/kotlin/io/lettuce/core/cluster/api/coroutines/RedisClusterCoroutinesCommands.kt

@@ -45,6 +45,11 @@ interface RedisClusterCoroutinesCommands<K : Any, V : Any> :
     RedisSortedSetCoroutinesCommands<K, V>,
     RedisStreamCoroutinesCommands<K, V>,
     RedisStringCoroutinesCommands<K, V>,
+    RedisFunctionCoroutinesCommands<K, V>,
+    RedisJsonCoroutinesCommands<K, V>,
+    RedisVectorSetCoroutinesCommands<K, V>,
+    RedisArrayCoroutinesCommands<K, V>,
+    RediSearchCoroutinesCommands<K, V>,
     RedisBloomFilterCoroutinesCommands<K, V>,
     RedisCuckooFilterCoroutinesCommands<K, V>,
     RedisTopKCoroutinesCommands<K, V> {

src/main/kotlin/io/lettuce/core/cluster/api/coroutines/RedisClusterCoroutinesCommandsImpl.kt

@@ -53,6 +53,11 @@ internal class RedisClusterCoroutinesCommandsImpl<K : Any, V : Any>(
     RedisSortedSetCoroutinesCommands<K, V> by RedisSortedSetCoroutinesCommandsImpl(ops),
     RedisStreamCoroutinesCommands<K, V> by RedisStreamCoroutinesCommandsImpl(ops),
     RedisStringCoroutinesCommands<K, V> by RedisStringCoroutinesCommandsImpl(ops),
+    RedisFunctionCoroutinesCommands<K, V> by RedisFunctionCoroutinesCommandsImpl(ops),
+    RedisJsonCoroutinesCommands<K, V> by RedisJsonCoroutinesCommandsImpl(ops),
+    RedisVectorSetCoroutinesCommands<K, V> by RedisVectorSetCoroutinesCommandsImpl(ops),
+    RedisArrayCoroutinesCommands<K, V> by RedisArrayCoroutinesCommandsImpl(ops),
+    RediSearchCoroutinesCommands<K, V> by RediSearchCoroutinesCommandsImpl(ops),
     RedisBloomFilterCoroutinesCommands<K, V> by RedisBloomFilterCoroutinesCommandsImpl(ops),
     RedisCuckooFilterCoroutinesCommands<K, V> by RedisCuckooFilterCoroutinesCommandsImpl(ops),
     RedisTopKCoroutinesCommands<K, V> by RedisTopKCoroutinesCommandsImpl(ops) {

src/test/java/io/lettuce/apigenerator/KotlinCompilationUnitFactory.java

@@ -70,12 +70,12 @@ class KotlinCompilationUnitFactory {
             "getStatefulConnection");
 
     private static final Set<String> FLOW_METHODS = LettuceSets.unmodifiableSet("aclList", "aclLog", "dispatch", "geohash",
-            "georadius", "georadiusbymember", "geosearch", "hgetall", "hkeys", "hmget", "hvals", "keys", "mget", "sdiff",
-            "sinter", "smembers", "smismember", "sort", "sortReadOnly", "srandmember", "sunion", "xclaim", "xpending", "xrange",
-            "xread", "xreadgroup", "xrevrange", "zdiff", "zdiffWithScores", "zinter", "zinterWithScores", "zpopmax", "zpopmin",
-            "zrange", "zrangeWithScores", "zrangebylex", "zrangebyscore", "zrangebyscoreWithScores", "zrevrange",
-            "zrevrangeWithScores", "zrevrangebylex", "zrevrangebyscore", "zrevrangebyscore", "zrevrangebyscoreWithScores",
-            "zunion", "zunionWithScores");
+            "georadius", "georadiusbymember", "geosearch", "hgetall", "hgetdel", "hgetex", "hkeys", "hmget", "hvals", "keys",
+            "mget", "sdiff", "sinter", "smembers", "smismember", "sort", "sortReadOnly", "srandmember", "sunion", "xackdel",
+            "xclaim", "xdelex", "xpending", "xrange", "xread", "xreadgroup", "xrevrange", "zdiff", "zdiffWithScores", "zinter",
+            "zinterWithScores", "zpopmax", "zpopmin", "zrange", "zrangeWithScores", "zrangebylex", "zrangebyscore",
+            "zrangebyscoreWithScores", "zrevrange", "zrevrangeWithScores", "zrevrangebylex", "zrevrangebyscore",
+            "zrevrangebyscore", "zrevrangebyscoreWithScores", "zunion", "zunionWithScores");
 
     private static final Set<String> NON_NULLABLE_RESULT_METHODS = LettuceSets.unmodifiableSet("discard", "multi", "exec",
             "watch", "unwatch", "getMasterAddrByName", "master", "reset", "failover", "monitor",