From 38d3576052f55446b8d8df7098e44ba3c69b6bc6 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 24 Jun 2021 14:55:12 +0200 Subject: [PATCH] Add reactive support for ZDIFF, ZDIFFSTORE, ZINTER, and ZUNION commands. See: #2041 & #2042 Original Pull Request: #2097 --- .../connection/ReactiveZSetCommands.java | 848 ++++++++++++++---- .../redis/connection/RedisZSetCommands.java | 56 -- .../LettuceReactiveClusterZSetCommands.java | 12 +- .../lettuce/LettuceReactiveZSetCommands.java | 231 ++++- .../core/DefaultReactiveZSetOperations.java | 239 +++-- .../redis/core/ReactiveZSetOperations.java | 329 +++++-- ...eReactiveZSetCommandsIntegrationTests.java | 105 ++- ...eactiveZSetOperationsIntegrationTests.java | 220 +++-- 8 files changed, 1615 insertions(+), 425 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java index bb3f01ced..e73dcaa8d 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java @@ -15,6 +15,8 @@ */ package org.springframework.data.redis.connection; +import static org.springframework.data.redis.connection.ReactiveRedisConnection.*; + import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -31,11 +33,6 @@ import java.util.function.Function; import org.reactivestreams.Publisher; import org.springframework.data.domain.Range; import org.springframework.data.domain.Sort.Direction; -import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; -import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; -import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyScanCommand; -import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse; -import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; @@ -1252,9 +1249,9 @@ public interface ReactiveZSetCommands { * {@code ZPOPMIN}/{@literal ZPOPMAX} command parameters. * * @author Mark Paluch + * @since 2.6 * @see Redis Documentation: ZPOPMIN * @see Redis Documentation: ZPOPMAX - * @since 2.6 */ class ZPopCommand extends KeyCommand { @@ -1326,9 +1323,9 @@ public interface ReactiveZSetCommands { * {@code BZPOPMIN}/{@literal BZPOPMAX} command parameters. * * @author Mark Paluch + * @since 2.6 * @see Redis Documentation: BZPOPMIN * @see Redis Documentation: BZPOPMAX - * @since 2.6 */ class BZPopCommand extends KeyCommand { @@ -1444,8 +1441,8 @@ public interface ReactiveZSetCommands { * * @param key must not be {@literal null}. * @return - * @see Redis Documentation: ZPOPMIN * @since 2.6 + * @see Redis Documentation: ZPOPMIN */ default Mono zPopMin(ByteBuffer key) { return zPop(Mono.just(ZPopCommand.min().from(key))).map(CommandResponse::getOutput).flatMap(Flux::next).next(); @@ -1457,8 +1454,8 @@ public interface ReactiveZSetCommands { * @param key must not be {@literal null}. * @param count number of elements to pop. * @return - * @see Redis Documentation: ZPOPMIN * @since 2.6 + * @see Redis Documentation: ZPOPMIN */ default Flux zPopMin(ByteBuffer key, long count) { return zPop(Mono.just(ZPopCommand.min().from(key).count(count))).map(CommandResponse::getOutput) @@ -1473,8 +1470,8 @@ public interface ReactiveZSetCommands { * @param timeout must not be {@literal null}. * @return * @throws IllegalArgumentException if the timeout is {@literal null} or negative. - * @see Redis Documentation: BZPOPMIN * @since 2.6 + * @see Redis Documentation: BZPOPMIN */ default Mono bZPopMin(ByteBuffer key, Duration timeout) { @@ -1490,8 +1487,8 @@ public interface ReactiveZSetCommands { * * @param key must not be {@literal null}. * @return - * @see Redis Documentation: ZPOPMAX * @since 2.6 + * @see Redis Documentation: ZPOPMAX */ default Mono zPopMax(ByteBuffer key) { return zPop(Mono.just(ZPopCommand.max().from(key))).map(CommandResponse::getOutput).flatMap(Flux::next).next(); @@ -1503,8 +1500,8 @@ public interface ReactiveZSetCommands { * @param key must not be {@literal null}. * @param count number of elements to pop. * @return - * @see Redis Documentation: ZPOPMAX * @since 2.6 + * @see Redis Documentation: ZPOPMAX */ default Flux zPopMax(ByteBuffer key, long count) { return zPop(Mono.just(ZPopCommand.max().from(key).count(count))).map(CommandResponse::getOutput) @@ -1519,8 +1516,8 @@ public interface ReactiveZSetCommands { * @param timeout must not be {@literal null}. * @return * @throws IllegalArgumentException if the timeout is {@literal null} or negative. - * @see Redis Documentation: BZPOPMAX * @since 2.6 + * @see Redis Documentation: BZPOPMAX */ default Mono bZPopMax(ByteBuffer key, Duration timeout) { @@ -1654,8 +1651,8 @@ public interface ReactiveZSetCommands { * {@code ZMSCORE} command parameters. * * @author Mark Paluch - * @see Redis Documentation: ZMSCORE * @since 2.6 + * @see Redis Documentation: ZMSCORE */ class ZMScoreCommand extends KeyCommand { @@ -1721,8 +1718,8 @@ public interface ReactiveZSetCommands { * @param key must not be {@literal null}. * @param values must not be {@literal null}. * @return - * @see Redis Documentation: ZMSCORE * @since 2.6 + * @see Redis Documentation: ZMSCORE */ default Mono> zMScore(ByteBuffer key, Collection values) { @@ -1738,8 +1735,8 @@ public interface ReactiveZSetCommands { * * @param commands must not be {@literal null}. * @return - * @see Redis Documentation: ZMSCORE * @since 2.6 + * @see Redis Documentation: ZMSCORE */ Flux> zMScore(Publisher commands); @@ -1968,57 +1965,229 @@ public interface ReactiveZSetCommands { Flux> zRemRangeByLex(Publisher commands); /** - * {@code ZUNIONSTORE} command parameters. + * {@code ZDIFF} command parameters. + * + * @author Mark Paluch + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + class ZDiffCommand implements Command { + + private final List keys; + + private ZDiffCommand(List keys) { + this.keys = keys; + } + + /** + * Creates a new {@link ZDiffCommand} given a {@link Collection} of keys. + * + * @param keys must not be {@literal null}. + * @return a new {@link ZDiffCommand} for a {@link Collection} of values. + */ + public static ZDiffCommand sets(Collection keys) { + + Assert.notNull(keys, "Keys must not be null!"); + + return new ZDiffCommand(new ArrayList<>(keys)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveRedisConnection.Command#getKey() + */ + @Override + @Nullable + public ByteBuffer getKey() { + return null; + } + + /** + * @return + */ + public List getKeys() { + return keys; + } + } + + /** + * Diff sorted {@literal sets}. + * + * @param sets must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + default Flux zDiff(List sets) { + return zDiff(Mono.just(ZDiffCommand.sets(sets))).flatMap(CommandResponse::getOutput); + } + + /** + * Diff sorted {@literal sets}. + * + * @param commands + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + Flux>> zDiff(Publisher commands); + + /** + * Diff sorted {@literal sets}. + * + * @param sets must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + default Flux zDiffWithScores(List sets) { + return zDiffWithScores(Mono.just(ZDiffCommand.sets(sets))).flatMap(CommandResponse::getOutput); + } + + /** + * Diff sorted {@literal sets}. + * + * @param commands + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + Flux>> zDiffWithScores(Publisher commands); + + /** + * {@code ZDIFFSTORE} command parameters. + * + * @author Mark Paluch + * @since 2.6 + * @see Redis Documentation: ZDIFFSTORE + */ + class ZDiffStoreCommand extends KeyCommand { + + private final List sourceKeys; + + private ZDiffStoreCommand(@Nullable ByteBuffer key, List sourceKeys) { + + super(key); + + this.sourceKeys = sourceKeys; + } + + /** + * Creates a new {@link ZDiffStoreCommand} given a {@link Collection} of keys. + * + * @param keys must not be {@literal null}. + * @return a new {@link ZDiffStoreCommand} for a {@link Collection} of values. + */ + public static ZDiffStoreCommand sourceKeys(Collection keys) { + + Assert.notNull(keys, "Keys must not be null!"); + + return new ZDiffStoreCommand(null, new ArrayList<>(keys)); + } + + /** + * Applies the {@literal key} at which the result is stored. Constructs a new command instance with all previously + * configured properties. + * + * @param key must not be {@literal null}. + * @return a new {@link ZDiffStoreCommand} with {@literal key} applied. + */ + public ZDiffStoreCommand storeAs(ByteBuffer key) { + + Assert.notNull(key, "Key must not be null!"); + + return new ZDiffStoreCommand(key, sourceKeys); + } + + /** + * @return + */ + public List getSourceKeys() { + return sourceKeys; + } + } + + /** + * Diff sorted {@literal sets} and store result in destination {@literal destinationKey}. + * + * @param destinationKey must not be {@literal null}. + * @param sets must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFFSTORE + */ + default Mono zDiffStore(ByteBuffer destinationKey, List sets) { + + Assert.notNull(destinationKey, "DestinationKey must not be null!"); + Assert.notNull(sets, "Sets must not be null!"); + + return zDiffStore(Mono.just(ZDiffStoreCommand.sourceKeys(sets).storeAs(destinationKey))).next() + .map(NumericResponse::getOutput); + } + + /** + * Diff sorted {@literal sets} and store result in destination {@literal destinationKey}. + * + * @param commands + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFFSTORE + */ + Flux> zDiffStore(Publisher commands); + + /** + * {@code ZINTER}/{@code ZUNION} command parameters. * * @author Christoph Strobl - * @see Redis Documentation: ZUNIONSTORE + * @author Mark Paluch + * @since 2.6 + * @see Redis Documentation: ZINTER + * @see Redis Documentation: ZUNION */ - class ZUnionStoreCommand extends KeyCommand { + class ZAggregateCommand implements Command { private final List sourceKeys; private final List weights; private final @Nullable Aggregate aggregateFunction; - private ZUnionStoreCommand(@Nullable ByteBuffer key, List sourceKeys, List weights, - @Nullable Aggregate aggregate) { + private ZAggregateCommand(List sourceKeys, List weights, @Nullable Aggregate aggregate) { - super(key); this.sourceKeys = sourceKeys; this.weights = weights; this.aggregateFunction = aggregate; } /** - * Creates a new {@link ZUnionStoreCommand} given a {@link List} of keys. + * Creates a new {@link ZAggregateCommand} given a {@link List} of keys. * * @param keys must not be {@literal null}. - * @return a new {@link ZUnionStoreCommand} for {@link Range}. + * @return a new {@link ZAggregateCommand} for {@link Range}. */ - public static ZUnionStoreCommand sets(List keys) { + public static ZAggregateCommand sets(List keys) { Assert.notNull(keys, "Keys must not be null!"); - return new ZUnionStoreCommand(null, new ArrayList<>(keys), Collections.emptyList(), null); + return new ZAggregateCommand(new ArrayList<>(keys), Collections.emptyList(), null); } /** * Applies the {@link List} of weights. Constructs a new command instance with all previously configured properties. * * @param weights must not be {@literal null}. - * @return a new {@link ZUnionStoreCommand} with {@literal weights} applied. + * @return a new {@link ZAggregateCommand} with {@literal weights} applied. */ - public ZUnionStoreCommand applyWeights(List weights) { - return new ZUnionStoreCommand(getKey(), sourceKeys, weights, aggregateFunction); + public ZAggregateCommand applyWeights(List weights) { + return new ZAggregateCommand(sourceKeys, weights, aggregateFunction); } /** * Applies the {@link Weights}. Constructs a new command instance with all previously configured properties. * * @param weights must not be {@literal null}. - * @return a new {@link ZUnionStoreCommand} with {@literal weights} applied. + * @return a new {@link ZAggregateCommand} with {@literal weights} applied. * @since 2.1 */ - public ZUnionStoreCommand applyWeights(Weights weights) { + public ZAggregateCommand applyWeights(Weights weights) { return applyWeights(weights.toList()); } @@ -2027,25 +2196,17 @@ public interface ReactiveZSetCommands { * properties. * * @param aggregateFunction can be {@literal null}. - * @return a new {@link ZUnionStoreCommand} with {@link Aggregate} applied. + * @return a new {@link ZAggregateStoreCommand} with {@link Aggregate} applied. */ - public ZUnionStoreCommand aggregateUsing(@Nullable Aggregate aggregateFunction) { + public ZAggregateCommand aggregateUsing(@Nullable Aggregate aggregateFunction) { - return new ZUnionStoreCommand(getKey(), sourceKeys, weights, aggregateFunction); + return new ZAggregateCommand(sourceKeys, weights, aggregateFunction); } - /** - * Applies the {@literal key} at which the result is stored. Constructs a new command instance with all previously - * configured properties. - * - * @param key must not be {@literal null}. - * @return a new {@link ZUnionStoreCommand} with {@literal key} applied. - */ - public ZUnionStoreCommand storeAs(ByteBuffer key) { - - Assert.notNull(key, "Key must not be null!"); - - return new ZUnionStoreCommand(key, sourceKeys, weights, aggregateFunction); + @Nullable + @Override + public ByteBuffer getKey() { + return null; } /** @@ -2071,100 +2232,223 @@ public interface ReactiveZSetCommands { } /** - * Union sorted {@literal sets} and store result in destination {@literal destinationKey}. + * {@code ZINTERSTORE}/{@code ZUNIONSTORE} command parameters. * - * @param destinationKey must not be {@literal null}. - * @param sets must not be {@literal null}. - * @return + * @author Christoph Strobl + * @author Mark Paluch + * @since 2.6 + * @see Redis Documentation: ZINTERSTORE * @see Redis Documentation: ZUNIONSTORE */ - default Mono zUnionStore(ByteBuffer destinationKey, List sets) { - return zUnionStore(destinationKey, sets, Collections.emptyList()); + class ZAggregateStoreCommand extends KeyCommand { + + private final List sourceKeys; + private final List weights; + private final @Nullable Aggregate aggregateFunction; + + private ZAggregateStoreCommand(@Nullable ByteBuffer key, List sourceKeys, List weights, + @Nullable Aggregate aggregate) { + + super(key); + this.sourceKeys = sourceKeys; + this.weights = weights; + this.aggregateFunction = aggregate; + } + + /** + * Creates a new {@link ZAggregateStoreCommand} given a {@link List} of keys. + * + * @param keys must not be {@literal null}. + * @return a new {@link ZAggregateStoreCommand} for {@link Range}. + */ + public static ZAggregateStoreCommand sets(List keys) { + + Assert.notNull(keys, "Keys must not be null!"); + + return new ZAggregateStoreCommand(null, new ArrayList<>(keys), Collections.emptyList(), null); + } + + /** + * Applies the {@link List} of weights. Constructs a new command instance with all previously configured properties. + * + * @param weights must not be {@literal null}. + * @return a new {@link ZAggregateStoreCommand} with {@literal weights} applied. + */ + public ZAggregateStoreCommand applyWeights(List weights) { + return new ZAggregateStoreCommand(getKey(), sourceKeys, weights, aggregateFunction); + } + + /** + * Applies the {@link Weights}. Constructs a new command instance with all previously configured properties. + * + * @param weights must not be {@literal null}. + * @return a new {@link ZAggregateStoreCommand} with {@literal weights} applied. + * @since 2.1 + */ + public ZAggregateStoreCommand applyWeights(Weights weights) { + return applyWeights(weights.toList()); + } + + /** + * Applies a specific {@link Aggregate} function. Constructs a new command instance with all previously configured + * properties. + * + * @param aggregateFunction can be {@literal null}. + * @return a new {@link ZAggregateStoreCommand} with {@link Aggregate} applied. + */ + public ZAggregateStoreCommand aggregateUsing(@Nullable Aggregate aggregateFunction) { + + return new ZAggregateStoreCommand(getKey(), sourceKeys, weights, aggregateFunction); + } + + /** + * Applies the {@literal key} at which the result is stored. Constructs a new command instance with all previously + * configured properties. + * + * @param key must not be {@literal null}. + * @return a new {@link ZAggregateStoreCommand} with {@literal key} applied. + */ + public ZAggregateStoreCommand storeAs(ByteBuffer key) { + + Assert.notNull(key, "Key must not be null!"); + + return new ZAggregateStoreCommand(key, sourceKeys, weights, aggregateFunction); + } + + /** + * @return never {@literal null}. + */ + public List getSourceKeys() { + return sourceKeys; + } + + /** + * @return never {@literal null}. + */ + public List getWeights() { + return weights; + } + + /** + * @return never {@literal null}. + */ + public Optional getAggregateFunction() { + return Optional.ofNullable(aggregateFunction); + } } /** - * Union sorted {@literal sets} and store result in destination {@literal destinationKey} and apply weights to - * individual sets. + * Intersect sorted {@literal sets}. * - * @param destinationKey must not be {@literal null}. * @param sets must not be {@literal null}. - * @param weights must not be {@literal null}. * @return - * @see Redis Documentation: ZUNIONSTORE + * @since 2.6 + * @see Redis Documentation: ZINTER */ - default Mono zUnionStore(ByteBuffer destinationKey, List sets, List weights) { - return zUnionStore(destinationKey, sets, weights, null); - } + default Flux zInter(List sets) { - /** - * Union sorted {@literal sets} and store result in destination {@literal destinationKey} and apply weights to - * individual sets. - * - * @param destinationKey must not be {@literal null}. - * @param sets must not be {@literal null}. - * @param weights must not be {@literal null}. - * @return - * @since 2.1 - * @see Redis Documentation: ZUNIONSTORE - */ - default Mono zUnionStore(ByteBuffer destinationKey, List sets, Weights weights) { - return zUnionStore(destinationKey, sets, weights, null); - } - - /** - * Union sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination - * {@literal destinationKey} and apply weights to individual sets. - * - * @param destinationKey must not be {@literal null}. - * @param sets must not be {@literal null}. - * @param weights can be {@literal null}. - * @param aggregateFunction can be {@literal null}. - * @return - * @see Redis Documentation: ZUNIONSTORE - */ - default Mono zUnionStore(ByteBuffer destinationKey, List sets, List weights, - @Nullable Aggregate aggregateFunction) { - - Assert.notNull(destinationKey, "DestinationKey must not be null!"); Assert.notNull(sets, "Sets must not be null!"); - return zUnionStore(Mono.just( - ZUnionStoreCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights).storeAs(destinationKey))) - .next().map(NumericResponse::getOutput); + return zInter(Mono.just(ZAggregateCommand.sets(sets))).flatMap(CommandResponse::getOutput); } /** - * Union sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination - * {@literal destinationKey} and apply weights to individual sets. - * - * @param destinationKey must not be {@literal null}. - * @param sets must not be {@literal null}. - * @param weights can be {@literal null}. - * @param aggregateFunction can be {@literal null}. - * @return - * @since 2.1 - * @see Redis Documentation: ZUNIONSTORE - */ - default Mono zUnionStore(ByteBuffer destinationKey, List sets, Weights weights, - @Nullable Aggregate aggregateFunction) { - - Assert.notNull(destinationKey, "DestinationKey must not be null!"); - Assert.notNull(sets, "Sets must not be null!"); - - return zUnionStore(Mono.just( - ZUnionStoreCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights).storeAs(destinationKey))) - .next().map(NumericResponse::getOutput); - } - - /** - * Union sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination - * {@literal destinationKey} and apply weights to individual sets. + * Intersect sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. * * @param commands * @return - * @see Redis Documentation: ZUNIONSTORE + * @since 2.6 + * @see Redis Documentation: ZINTER */ - Flux> zUnionStore(Publisher commands); + Flux>> zInter(Publisher commands); + + /** + * Intersect sorted {@literal sets}. + * + * @param sets must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux zInterWithScores(List sets) { + return zInterWithScores(sets, Collections.emptyList()); + } + + /** + * Intersect sorted {@literal sets} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux zInterWithScores(List sets, List weights) { + return zInterWithScores(sets, weights, null); + } + + /** + * Intersect sorted {@literal sets} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux zInterWithScores(List sets, Weights weights) { + return zInterWithScores(sets, weights, null); + } + + /** + * Intersect sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @param aggregateFunction can be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux zInterWithScores(List sets, List weights, + @Nullable Aggregate aggregateFunction) { + + Assert.notNull(sets, "Sets must not be null!"); + + return zInterWithScores( + Mono.just(ZAggregateCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights))) + .flatMap(CommandResponse::getOutput); + } + + /** + * Intersect sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @param aggregateFunction can be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux zInterWithScores(List sets, Weights weights, @Nullable Aggregate aggregateFunction) { + + Assert.notNull(sets, "Sets must not be null!"); + + return zInterWithScores( + Mono.just(ZAggregateCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights))) + .flatMap(CommandResponse::getOutput); + } + + /** + * Intersect sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. + * + * @param commands + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + Flux>> zInterWithScores( + Publisher commands); /** * {@code ZINTERSTORE} command parameters. @@ -2172,19 +2456,11 @@ public interface ReactiveZSetCommands { * @author Christoph Strobl * @see Redis Documentation: ZINTERSTORE */ - class ZInterStoreCommand extends KeyCommand { - - private final List sourceKeys; - private final List weights; - private final @Nullable Aggregate aggregateFunction; + class ZInterStoreCommand extends ZAggregateStoreCommand { private ZInterStoreCommand(ByteBuffer key, List sourceKeys, List weights, @Nullable Aggregate aggregate) { - - super(key); - this.sourceKeys = sourceKeys; - this.weights = weights; - this.aggregateFunction = aggregate; + super(key, sourceKeys, weights, aggregate); } /** @@ -2208,7 +2484,7 @@ public interface ReactiveZSetCommands { * @return a new {@link ZInterStoreCommand} with {@literal weights} applied. */ public ZInterStoreCommand applyWeights(List weights) { - return new ZInterStoreCommand(getKey(), sourceKeys, weights, aggregateFunction); + return new ZInterStoreCommand(getKey(), getSourceKeys(), weights, getAggregateFunction().orElse(null)); } /** @@ -2231,7 +2507,7 @@ public interface ReactiveZSetCommands { */ public ZInterStoreCommand aggregateUsing(@Nullable Aggregate aggregateFunction) { - return new ZInterStoreCommand(getKey(), sourceKeys, weights, aggregateFunction); + return new ZInterStoreCommand(getKey(), getSourceKeys(), getWeights(), aggregateFunction); } /** @@ -2245,29 +2521,9 @@ public interface ReactiveZSetCommands { Assert.notNull(key, "Key must not be null!"); - return new ZInterStoreCommand(key, sourceKeys, weights, aggregateFunction); + return new ZInterStoreCommand(key, getSourceKeys(), getWeights(), getAggregateFunction().orElse(null)); } - /** - * @return never {@literal null}. - */ - public List getSourceKeys() { - return sourceKeys; - } - - /** - * @return never {@literal null}. - */ - public List getWeights() { - return weights; - } - - /** - * @return never {@literal null}.ΓΈ - */ - public Optional getAggregateFunction() { - return Optional.ofNullable(aggregateFunction); - } } /** @@ -2364,7 +2620,291 @@ public interface ReactiveZSetCommands { * @return * @see Redis Documentation: ZINTERSTORE */ - Flux> zInterStore(Publisher commands); + Flux> zInterStore(Publisher commands); + + /** + * Union sorted {@literal sets}. + * + * @param sets must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux zUnion(List sets) { + + Assert.notNull(sets, "Sets must not be null!"); + + return zUnion(Mono.just(ZAggregateCommand.sets(sets))).flatMap(CommandResponse::getOutput); + } + + /** + * Union sorted {@literal sets}. + * + * @param sets must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux zUnionWithScores(List sets) { + return zUnionWithScores(sets, Collections.emptyList()); + } + + /** + * Union sorted {@literal sets} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux zUnionWithScores(List sets, List weights) { + return zUnionWithScores(sets, weights, null); + } + + /** + * Union sorted {@literal sets} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux zUnionWithScores(List sets, Weights weights) { + return zUnionWithScores(sets, weights, null); + } + + /** + * Union sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights can be {@literal null}. + * @param aggregateFunction can be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux zUnionWithScores(List sets, List weights, + @Nullable Aggregate aggregateFunction) { + + Assert.notNull(sets, "Sets must not be null!"); + + return zUnionWithScores( + Mono.just(ZAggregateCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights))) + .flatMap(CommandResponse::getOutput); + } + + /** + * Union sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. + * + * @param sets must not be {@literal null}. + * @param weights can be {@literal null}. + * @param aggregateFunction can be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux zUnionWithScores(List sets, Weights weights, @Nullable Aggregate aggregateFunction) { + + Assert.notNull(sets, "Sets must not be null!"); + + return zUnionWithScores( + Mono.just(ZAggregateCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights))) + .flatMap(CommandResponse::getOutput); + } + + /** + * Union sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. + * + * @param commands + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + Flux>> zUnionWithScores( + Publisher commands); + + /** + * Union sorted {@literal sets} by applying {@literal aggregateFunction} and apply weights to individual sets. + * + * @param commands + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + Flux>> zUnion(Publisher commands); + + /** + * {@code ZUNIONSTORE} command parameters. + * + * @author Christoph Strobl + * @see Redis Documentation: ZUNIONSTORE + */ + class ZUnionStoreCommand extends ZAggregateStoreCommand { + + private ZUnionStoreCommand(@Nullable ByteBuffer key, List sourceKeys, List weights, + @Nullable Aggregate aggregate) { + + super(key, sourceKeys, weights, aggregate); + } + + /** + * Creates a new {@link ZUnionStoreCommand} given a {@link List} of keys. + * + * @param keys must not be {@literal null}. + * @return a new {@link ZUnionStoreCommand} for {@link Range}. + */ + public static ZUnionStoreCommand sets(List keys) { + + Assert.notNull(keys, "Keys must not be null!"); + + return new ZUnionStoreCommand(null, new ArrayList<>(keys), Collections.emptyList(), null); + } + + /** + * Applies the {@link List} of weights. Constructs a new command instance with all previously configured properties. + * + * @param weights must not be {@literal null}. + * @return a new {@link ZUnionStoreCommand} with {@literal weights} applied. + */ + public ZUnionStoreCommand applyWeights(List weights) { + return new ZUnionStoreCommand(getKey(), getSourceKeys(), weights, getAggregateFunction().orElse(null)); + } + + /** + * Applies the {@link Weights}. Constructs a new command instance with all previously configured properties. + * + * @param weights must not be {@literal null}. + * @return a new {@link ZUnionStoreCommand} with {@literal weights} applied. + * @since 2.1 + */ + public ZUnionStoreCommand applyWeights(Weights weights) { + return applyWeights(weights.toList()); + } + + /** + * Applies a specific {@link Aggregate} function. Constructs a new command instance with all previously configured + * properties. + * + * @param aggregateFunction can be {@literal null}. + * @return a new {@link ZUnionStoreCommand} with {@link Aggregate} applied. + */ + public ZUnionStoreCommand aggregateUsing(@Nullable Aggregate aggregateFunction) { + + return new ZUnionStoreCommand(getKey(), getSourceKeys(), getWeights(), aggregateFunction); + } + + /** + * Applies the {@literal key} at which the result is stored. Constructs a new command instance with all previously + * configured properties. + * + * @param key must not be {@literal null}. + * @return a new {@link ZUnionStoreCommand} with {@literal key} applied. + */ + public ZUnionStoreCommand storeAs(ByteBuffer key) { + + Assert.notNull(key, "Key must not be null!"); + + return new ZUnionStoreCommand(key, getSourceKeys(), getWeights(), getAggregateFunction().orElse(null)); + } + + } + + /** + * Union sorted {@literal sets} and store result in destination {@literal destinationKey}. + * + * @param destinationKey must not be {@literal null}. + * @param sets must not be {@literal null}. + * @return + * @see Redis Documentation: ZUNIONSTORE + */ + default Mono zUnionStore(ByteBuffer destinationKey, List sets) { + return zUnionStore(destinationKey, sets, Collections.emptyList()); + } + + /** + * Union sorted {@literal sets} and store result in destination {@literal destinationKey} and apply weights to + * individual sets. + * + * @param destinationKey must not be {@literal null}. + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @see Redis Documentation: ZUNIONSTORE + */ + default Mono zUnionStore(ByteBuffer destinationKey, List sets, List weights) { + return zUnionStore(destinationKey, sets, weights, null); + } + + /** + * Union sorted {@literal sets} and store result in destination {@literal destinationKey} and apply weights to + * individual sets. + * + * @param destinationKey must not be {@literal null}. + * @param sets must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.1 + * @see Redis Documentation: ZUNIONSTORE + */ + default Mono zUnionStore(ByteBuffer destinationKey, List sets, Weights weights) { + return zUnionStore(destinationKey, sets, weights, null); + } + + /** + * Union sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination + * {@literal destinationKey} and apply weights to individual sets. + * + * @param destinationKey must not be {@literal null}. + * @param sets must not be {@literal null}. + * @param weights can be {@literal null}. + * @param aggregateFunction can be {@literal null}. + * @return + * @see Redis Documentation: ZUNIONSTORE + */ + default Mono zUnionStore(ByteBuffer destinationKey, List sets, List weights, + @Nullable Aggregate aggregateFunction) { + + Assert.notNull(destinationKey, "DestinationKey must not be null!"); + Assert.notNull(sets, "Sets must not be null!"); + + return zUnionStore(Mono.just(ZAggregateStoreCommand.sets(sets).aggregateUsing(aggregateFunction) + .applyWeights(weights).storeAs(destinationKey))).next().map(NumericResponse::getOutput); + } + + /** + * Union sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination + * {@literal destinationKey} and apply weights to individual sets. + * + * @param destinationKey must not be {@literal null}. + * @param sets must not be {@literal null}. + * @param weights can be {@literal null}. + * @param aggregateFunction can be {@literal null}. + * @return + * @since 2.1 + * @see Redis Documentation: ZUNIONSTORE + */ + default Mono zUnionStore(ByteBuffer destinationKey, List sets, Weights weights, + @Nullable Aggregate aggregateFunction) { + + Assert.notNull(destinationKey, "DestinationKey must not be null!"); + Assert.notNull(sets, "Sets must not be null!"); + + return zUnionStore(Mono.just( + ZUnionStoreCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights).storeAs(destinationKey))) + .next().map(NumericResponse::getOutput); + } + + /** + * Union sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination + * {@literal destinationKey} and apply weights to individual sets. + * + * @param commands + * @return + * @see Redis Documentation: ZUNIONSTORE + */ + Flux> zUnionStore(Publisher commands); /** * {@code ZRANGEBYLEX}/{@literal ZREVRANGEBYLEX} command parameters. diff --git a/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java index 1c65a9708..07b37c720 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java @@ -1174,34 +1174,6 @@ public interface RedisZSetCommands { @Nullable Set zInter(byte[]... sets); - /** - * Intersect sorted {@code sets}. - * - * @param aggregate must not be {@literal null}. - * @param weights must not be {@literal null}. - * @param sets must not be {@literal null}. - * @return {@literal null} when used in pipeline / transaction. - * @since 2.6 - * @see Redis Documentation: ZINTER - */ - @Nullable - default Set zInter(Aggregate aggregate, int[] weights, byte[]... sets) { - return zInter(aggregate, Weights.of(weights), sets); - } - - /** - * Intersect sorted {@code sets}. - * - * @param aggregate must not be {@literal null}. - * @param weights must not be {@literal null}. - * @param sets must not be {@literal null}. - * @return {@literal null} when used in pipeline / transaction. - * @since 2.6 - * @see Redis Documentation: ZINTER - */ - @Nullable - Set zInter(Aggregate aggregate, Weights weights, byte[]... sets); - /** * Intersect sorted {@code sets}. * @@ -1293,34 +1265,6 @@ public interface RedisZSetCommands { @Nullable Set zUnion(byte[]... sets); - /** - * Union sorted {@code sets}. - * - * @param aggregate must not be {@literal null}. - * @param weights must not be {@literal null}. - * @param sets must not be {@literal null}. - * @return {@literal null} when used in pipeline / transaction. - * @since 2.6 - * @see Redis Documentation: ZUNION - */ - @Nullable - default Set zUnion(Aggregate aggregate, int[] weights, byte[]... sets) { - return zUnion(aggregate, Weights.of(weights), sets); - } - - /** - * Union sorted {@code sets}. - * - * @param aggregate must not be {@literal null}. - * @param weights must not be {@literal null}. - * @param sets must not be {@literal null}. - * @return {@literal null} when used in pipeline / transaction. - * @since 2.6 - * @see Redis Documentation: ZUNION - */ - @Nullable - Set zUnion(Aggregate aggregate, Weights weights, byte[]... sets); - /** * Union sorted {@code sets}. * diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterZSetCommands.java index 511ab1766..0d79e7727 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterZSetCommands.java @@ -41,11 +41,13 @@ class LettuceReactiveClusterZSetCommands extends LettuceReactiveZSetCommands imp super(connection); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveZSetCommands#zUnionStore(org.reactivestreams.Publisher) */ @Override - public Flux> zUnionStore(Publisher commands) { + public Flux> zUnionStore( + Publisher commands) { return getConnection().execute(cmd -> Flux.from(commands).concatMap(command -> { @@ -60,11 +62,13 @@ class LettuceReactiveClusterZSetCommands extends LettuceReactiveZSetCommands imp })); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveZSetCommands#zInterStore(org.reactivestreams.Publisher) */ @Override - public Flux> zInterStore(Publisher commands) { + public Flux> zInterStore( + Publisher commands) { return getConnection().execute(cmd -> Flux.from(commands).concatMap(command -> { Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty."); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java index e9f07fee7..9ce78ad3d 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java @@ -200,22 +200,18 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) { if (command.isWithScores()) { - result = cmd.zrangeWithScores(command.getKey(), start, stop) - .map(this::toTuple); + result = cmd.zrangeWithScores(command.getKey(), start, stop).map(this::toTuple); } else { - result = cmd.zrange(command.getKey(), start, stop) - .map(value -> toTuple(value, Double.NaN)); + result = cmd.zrange(command.getKey(), start, stop).map(value -> toTuple(value, Double.NaN)); } } else { if (command.isWithScores()) { - result = cmd.zrevrangeWithScores(command.getKey(), start, stop) - .map(this::toTuple); + result = cmd.zrevrangeWithScores(command.getKey(), start, stop).map(this::toTuple); } else { - result = cmd.zrevrange(command.getKey(), start, stop) - .map(value -> toTuple(value, Double.NaN)); + result = cmd.zrevrange(command.getKey(), start, stop).map(value -> toTuple(value, Double.NaN)); } } @@ -247,8 +243,7 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { if (command.isWithScores()) { if (!isLimited) { - result = cmd.zrangebyscoreWithScores(command.getKey(), range) - .map(this::toTuple); + result = cmd.zrangebyscoreWithScores(command.getKey(), range).map(this::toTuple); } else { result = cmd .zrangebyscoreWithScores(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get())) @@ -257,8 +252,7 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { } else { if (!isLimited) { - result = cmd.zrangebyscore(command.getKey(), range) - .map(value -> toTuple(value, Double.NaN)); + result = cmd.zrangebyscore(command.getKey(), range).map(value -> toTuple(value, Double.NaN)); } else { result = cmd.zrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get())) @@ -272,20 +266,16 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { if (command.isWithScores()) { if (!isLimited) { - result = cmd.zrevrangebyscoreWithScores(command.getKey(), range) - .map(this::toTuple); + result = cmd.zrevrangebyscoreWithScores(command.getKey(), range).map(this::toTuple); } else { - result = cmd - .zrevrangebyscoreWithScores(command.getKey(), range, - LettuceConverters.toLimit(command.getLimit().get())) - .map(this::toTuple); + result = cmd.zrevrangebyscoreWithScores(command.getKey(), range, + LettuceConverters.toLimit(command.getLimit().get())).map(this::toTuple); } } else { if (!isLimited) { - result = cmd.zrevrangebyscore(command.getKey(), range) - .map(value -> toTuple(value, Double.NaN)); + result = cmd.zrevrangebyscore(command.getKey(), range).map(value -> toTuple(value, Double.NaN)); } else { result = cmd.zrevrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get())) @@ -518,12 +508,186 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { })); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zDiff(Publisher) + */ + @Override + public Flux>> zDiff(Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).map(command -> { + + Assert.notEmpty(command.getKeys(), "Keys must not be null or empty!"); + + ByteBuffer[] sourceKeys = command.getKeys().toArray(new ByteBuffer[0]); + return new CommandResponse<>(command, cmd.zdiff(sourceKeys)); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zDiffWithScores(Publisher) + */ + @Override + public Flux>> zDiffWithScores(Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).map(command -> { + + Assert.notEmpty(command.getKeys(), "Keys must not be null or empty!"); + + ByteBuffer[] sourceKeys = command.getKeys().toArray(new ByteBuffer[0]); + return new CommandResponse<>(command, cmd.zdiffWithScores(sourceKeys).map(this::toTuple)); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zDiffStore(Publisher) + */ + @Override + public Flux> zDiffStore(Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + + Assert.notNull(command.getKey(), "Destination key must not be null!"); + Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!"); + + ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]); + return cmd.zdiffstore(command.getKey(), sourceKeys).map(value -> new NumericResponse<>(command, value)); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zInter(Publisher) + */ + @Override + public Flux>> zInter( + Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).map(command -> { + + Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!"); + + ZStoreArgs args = null; + if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) { + args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, + command.getWeights()); + } + + ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]); + Flux result = args != null ? cmd.zinter(args, sourceKeys) : cmd.zinter(sourceKeys); + return new CommandResponse<>(command, result); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zInterWithScores(Publisher) + */ + @Override + public Flux>> zInterWithScores( + Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).map(command -> { + + Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!"); + + ZStoreArgs args = null; + if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) { + args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, + command.getWeights()); + } + + ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]); + Flux> result = args != null ? cmd.zinterWithScores(args, sourceKeys) + : cmd.zinterWithScores(sourceKeys); + return new CommandResponse<>(command, result.map(this::toTuple)); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zInterStore(org.reactivestreams.Publisher) + */ + @Override + public Flux> zInterStore( + Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + + Assert.notNull(command.getKey(), "Destination key must not be null!"); + Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!"); + + ZStoreArgs args = null; + if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) { + args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, + command.getWeights()); + } + + ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]); + Mono result = args != null ? cmd.zinterstore(command.getKey(), args, sourceKeys) + : cmd.zinterstore(command.getKey(), sourceKeys); + return result.map(value -> new NumericResponse<>(command, value)); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zUnion(org.reactivestreams.Publisher) + */ + @Override + public Flux>> zUnion( + Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).map(command -> { + + Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!"); + + ZStoreArgs args = null; + if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) { + args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, + command.getWeights()); + } + + ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new); + Flux result = args != null ? cmd.zunion(args, sourceKeys) : cmd.zunion(sourceKeys); + return new CommandResponse<>(command, result); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zUnion(org.reactivestreams.Publisher) + */ + @Override + public Flux>> zUnionWithScores( + Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).map(command -> { + + Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!"); + + ZStoreArgs args = null; + if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) { + args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, + command.getWeights()); + } + + ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new); + Flux> result = args != null ? cmd.zunionWithScores(args, sourceKeys) + : cmd.zunionWithScores(sourceKeys); + return new CommandResponse<>(command, result.map(this::toTuple)); + })); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zUnionStore(org.reactivestreams.Publisher) */ @Override - public Flux> zUnionStore(Publisher commands) { + public Flux> zUnionStore( + Publisher commands) { return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { @@ -543,31 +707,6 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { })); } - /* - * (non-Javadoc) - * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zInterStore(org.reactivestreams.Publisher) - */ - @Override - public Flux> zInterStore(Publisher commands) { - - return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { - - Assert.notNull(command.getKey(), "Destination key must not be null!"); - Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!"); - - ZStoreArgs args = null; - if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) { - args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, - command.getWeights()); - } - - ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new); - Mono result = args != null ? cmd.zinterstore(command.getKey(), args, sourceKeys) - : cmd.zinterstore(command.getKey(), sourceKeys); - return result.map(value -> new NumericResponse<>(command, value)); - })); - } - /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zRangeByLex(org.reactivestreams.Publisher) diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java index 234320b1e..5d018caa4 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java @@ -27,7 +27,6 @@ import java.util.List; import java.util.function.Function; import org.reactivestreams.Publisher; - import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.DefaultTuple; import org.springframework.data.redis.connection.ReactiveZSetCommands; @@ -501,6 +500,194 @@ class DefaultReactiveZSetOperations implements ReactiveZSetOperations connection.zRemRangeByScore(rawKey(key), range)); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#difference(K, Collection) + */ + @Override + public Flux difference(K key, Collection otherKeys) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(connection::zDiff).map(this::readValue)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#differenceWithScores(K, Collection) + */ + @Override + public Flux> differenceWithScores(K key, Collection otherKeys) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(connection::zDiffWithScores).map(this::readTypedTuple)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#differenceAndStore(K, Collection, K) + */ + @Override + public Mono differenceAndStore(K key, Collection otherKeys, K destKey) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + Assert.notNull(destKey, "Destination key must not be null!"); + + return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMap(serialized -> connection.zDiffStore(rawKey(destKey), serialized))); + + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersect(K, Collection) + */ + @Override + public Flux intersect(K key, Collection otherKeys) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(connection::zInter).map(this::readValue)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectWithScores(K, Collection) + */ + @Override + public Flux> intersectWithScores(K key, Collection otherKeys) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(connection::zInterWithScores).map(this::readTypedTuple)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectWithScores(K, Collection, Aggregate, Weights) + */ + @Override + public Flux> intersectWithScores(K key, Collection otherKeys, Aggregate aggregate, Weights weights) { + + // TODO: Inconsistent method signatures Aggregate/Weights vs Weights/Aggregate in Connection API + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + Assert.notNull(aggregate, "Aggregate must not be null!"); + Assert.notNull(weights, "Weights must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(sets -> connection.zInterWithScores(sets, weights, aggregate)).map(this::readTypedTuple)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectAndStore(java.lang.Object, java.util.Collection, java.lang.Object) + */ + @Override + public Mono intersectAndStore(K key, Collection otherKeys, K destKey) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + Assert.notNull(destKey, "Destination key must not be null!"); + + return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized))); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectAndStore(java.lang.Object, java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights) + */ + @Override + public Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + Assert.notNull(destKey, "Destination key must not be null!"); + Assert.notNull(aggregate, "Aggregate must not be null!"); + Assert.notNull(weights, "Weights must not be null!"); + + return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized, weights, aggregate))); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#union(K, Collection) + */ + @Override + public Flux union(K key, Collection otherKeys) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(connection::zUnion).map(this::readValue)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionWithScores(K, Collection) + */ + @Override + public Flux> unionWithScores(K key, Collection otherKeys) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(connection::zUnionWithScores).map(this::readTypedTuple)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionWithScores(K, Collection, Aggregate, Weights) + */ + @Override + public Flux> unionWithScores(K key, Collection otherKeys, Aggregate aggregate, Weights weights) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(otherKeys, "Other keys must not be null!"); + Assert.notNull(aggregate, "Aggregate must not be null!"); + Assert.notNull(weights, "Weights must not be null!"); + + return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // + .map(this::rawKey) // + .collectList() // + .flatMapMany(sets -> connection.zUnionWithScores(sets, weights, aggregate)).map(this::readTypedTuple)); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionAndStore(java.lang.Object, java.lang.Object, java.lang.Object) @@ -551,56 +738,6 @@ class DefaultReactiveZSetOperations implements ReactiveZSetOperations connection.zUnionStore(rawKey(destKey), serialized, weights, aggregate))); } - /* - * (non-Javadoc) - * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectAndStore(java.lang.Object, java.lang.Object, java.lang.Object) - */ - @Override - public Mono intersectAndStore(K key, K otherKey, K destKey) { - - Assert.notNull(key, "Key must not be null!"); - Assert.notNull(otherKey, "Other key must not be null!"); - Assert.notNull(destKey, "Destination key must not be null!"); - - return intersectAndStore(key, Collections.singleton(otherKey), destKey); - } - - /* - * (non-Javadoc) - * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectAndStore(java.lang.Object, java.util.Collection, java.lang.Object) - */ - @Override - public Mono intersectAndStore(K key, Collection otherKeys, K destKey) { - - Assert.notNull(key, "Key must not be null!"); - Assert.notNull(otherKeys, "Other keys must not be null!"); - Assert.notNull(destKey, "Destination key must not be null!"); - - return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // - .map(this::rawKey) // - .collectList() // - .flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized))); - } - - /* - * (non-Javadoc) - * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectAndStore(java.lang.Object, java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights) - */ - @Override - public Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights) { - - Assert.notNull(key, "Key must not be null!"); - Assert.notNull(otherKeys, "Other keys must not be null!"); - Assert.notNull(destKey, "Destination key must not be null!"); - Assert.notNull(aggregate, "Aggregate must not be null!"); - Assert.notNull(weights, "Weights must not be null!"); - - return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) // - .map(this::rawKey) // - .collectList() // - .flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized, weights, aggregate))); - } - /* * (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeByLex(java.lang.Object, org.springframework.data.domain.Range) diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java index 32d994618..0e4565754 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java @@ -20,6 +20,7 @@ import reactor.core.publisher.Mono; import java.time.Duration; import java.util.Collection; +import java.util.Collections; import java.util.List; import org.springframework.data.domain.Range; @@ -412,6 +413,283 @@ public interface ReactiveZSetOperations { */ Mono removeRangeByScore(K key, Range range); + /** + * Diff sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + default Flux difference(K key, K otherKey) { + return difference(key, Collections.singleton(otherKey)); + } + + /** + * Diff sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + Flux difference(K key, Collection otherKeys); + + /** + * Diff sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + default Flux> differenceWithScores(K key, K otherKey) { + return differenceWithScores(key, Collections.singleton(otherKey)); + } + + /** + * Diff sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFF + */ + Flux> differenceWithScores(K key, Collection otherKeys); + + /** + * Diff sorted {@code sets} and store result in destination {@code destKey}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param destKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFFSTORE + */ + default Mono differenceAndStore(K key, K otherKey, K destKey) { + return differenceAndStore(key, Collections.singleton(otherKey), destKey); + } + + /** + * Diff sorted {@code sets} and store result in destination {@code destKey}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param destKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZDIFFSTORE + */ + Mono differenceAndStore(K key, Collection otherKeys, K destKey); + + /** + * Intersect sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux intersect(K key, K otherKey) { + return intersect(key, Collections.singleton(otherKey)); + } + + /** + * Intersect sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + Flux intersect(K key, Collection otherKeys); + + /** + * Intersect sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux> intersectWithScores(K key, K otherKey) { + return intersectWithScores(key, Collections.singleton(otherKey)); + } + + /** + * Intersect sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + Flux> intersectWithScores(K key, Collection otherKeys); + + /** + * Intersect sorted sets at {@code key} and {@code otherKeys} . + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + default Flux> intersectWithScores(K key, Collection otherKeys, Aggregate aggregate) { + return intersectWithScores(key, otherKeys, aggregate, Weights.fromSetCount(1 + otherKeys.size())); + } + + /** + * Intersect sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZINTER + */ + Flux> intersectWithScores(K key, Collection otherKeys, Aggregate aggregate, Weights weights); + + /** + * Intersect sorted sets at {@code key} and {@code otherKey} and store result in destination {@code destKey}. + * + * @param key must not be {@literal null}. + * @param otherKey must not be {@literal null}. + * @param destKey must not be {@literal null}. + * @return + * @see Redis Documentation: ZINTERSTORE + */ + default Mono intersectAndStore(K key, K otherKey, K destKey) { + return intersectAndStore(key, Collections.singleton(otherKey), destKey); + } + + /** + * Intersect sorted sets at {@code key} and {@code otherKeys} and store result in destination {@code destKey}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param destKey must not be {@literal null}. + * @return + * @see Redis Documentation: ZINTERSTORE + */ + Mono intersectAndStore(K key, Collection otherKeys, K destKey); + + /** + * Intersect sorted sets at {@code key} and {@code otherKeys} and store result in destination {@code destKey}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param destKey must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @return + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + default Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate) { + return intersectAndStore(key, otherKeys, destKey, aggregate, Weights.fromSetCount(1 + otherKeys.size())); + } + + /** + * Intersect sorted sets at {@code key} and {@code otherKeys} and store result in destination {@code destKey}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param destKey must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights); + + /** + * Union sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux union(K key, K otherKey) { + return union(key, Collections.singleton(otherKey)); + } + + /** + * Union sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + Flux union(K key, Collection otherKeys); + + /** + * Union sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKey must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux> unionWithScores(K key, K otherKey) { + return unionWithScores(key, Collections.singleton(otherKey)); + } + + /** + * Union sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + Flux> unionWithScores(K key, Collection otherKeys); + + /** + * Union sorted sets at {@code key} and {@code otherKeys} . + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + default Flux> unionWithScores(K key, Collection otherKeys, Aggregate aggregate) { + return unionWithScores(key, otherKeys, aggregate, Weights.fromSetCount(1 + otherKeys.size())); + } + + /** + * Union sorted {@code sets}. + * + * @param key must not be {@literal null}. + * @param otherKeys must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @param weights must not be {@literal null}. + * @return + * @since 2.6 + * @see Redis Documentation: ZUNION + */ + Flux> unionWithScores(K key, Collection otherKeys, Aggregate aggregate, Weights weights); + /** * Union sorted sets at {@code key} and {@code otherKeys} and store result in destination {@code destKey}. * @@ -463,57 +741,6 @@ public interface ReactiveZSetOperations { */ Mono unionAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights); - /** - * Intersect sorted sets at {@code key} and {@code otherKey} and store result in destination {@code destKey}. - * - * @param key must not be {@literal null}. - * @param otherKey must not be {@literal null}. - * @param destKey must not be {@literal null}. - * @return - * @see Redis Documentation: ZINTERSTORE - */ - Mono intersectAndStore(K key, K otherKey, K destKey); - - /** - * Intersect sorted sets at {@code key} and {@code otherKeys} and store result in destination {@code destKey}. - * - * @param key must not be {@literal null}. - * @param otherKeys must not be {@literal null}. - * @param destKey must not be {@literal null}. - * @return - * @see Redis Documentation: ZINTERSTORE - */ - Mono intersectAndStore(K key, Collection otherKeys, K destKey); - - /** - * Intersect sorted sets at {@code key} and {@code otherKeys} and store result in destination {@code destKey}. - * - * @param key must not be {@literal null}. - * @param otherKeys must not be {@literal null}. - * @param destKey must not be {@literal null}. - * @param aggregate must not be {@literal null}. - * @return - * @since 2.1 - * @see Redis Documentation: ZINTERSTORE - */ - default Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate) { - return intersectAndStore(key, otherKeys, destKey, aggregate, Weights.fromSetCount(1 + otherKeys.size())); - } - - /** - * Intersect sorted sets at {@code key} and {@code otherKeys} and store result in destination {@code destKey}. - * - * @param key must not be {@literal null}. - * @param otherKeys must not be {@literal null}. - * @param destKey must not be {@literal null}. - * @param aggregate must not be {@literal null}. - * @param weights must not be {@literal null}. - * @return - * @since 2.1 - * @see Redis Documentation: ZINTERSTORE - */ - Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights); - /** * Get all elements with lexicographical ordering from {@literal ZSET} at {@code key} with a value between * {@link Range#getLowerBound()} and {@link Range#getUpperBound()}. diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java index 9609c9870..acc003d28 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java @@ -555,8 +555,50 @@ public class LettuceReactiveZSetCommandsIntegrationTests extends LettuceReactive .isEqualTo(1L); } - @ParameterizedRedisTest // DATAREDIS-525 - void zUnionStoreShouldWorkCorrectly() { + @ParameterizedRedisTest // GH-2041 + void zDiffShouldWorkCorrectly() { + + assumeThat(connectionProvider).isInstanceOf(StandaloneConnectionProvider.class); + + nativeCommands.zadd(KEY_1, 1D, VALUE_1); + nativeCommands.zadd(KEY_1, 2D, VALUE_2); + nativeCommands.zadd(KEY_1, 3D, VALUE_3); + nativeCommands.zadd(KEY_2, 1D, VALUE_1); + nativeCommands.zadd(KEY_2, 2D, VALUE_2); + + connection.zSetCommands().zDiff(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER)) // + .collectList() // + .as(StepVerifier::create) // + .assertNext(actual -> { + assertThat(actual).containsOnly(VALUE_3_BBUFFER); + }).verifyComplete(); + + connection.zSetCommands().zDiffWithScores(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER)) // + .collectList() // + .as(StepVerifier::create) // + .assertNext(actual -> { + assertThat(actual).containsOnly(new DefaultTuple(VALUE_3_BYTES, 3D)); + }).verifyComplete(); + } + + @ParameterizedRedisTest // GH-2041 + void zDiffStoreShouldWorkCorrectly() { + + assumeThat(connectionProvider).isInstanceOf(StandaloneConnectionProvider.class); + + nativeCommands.zadd(KEY_1, 1D, VALUE_1); + nativeCommands.zadd(KEY_1, 2D, VALUE_2); + nativeCommands.zadd(KEY_1, 3D, VALUE_3); + nativeCommands.zadd(KEY_2, 1D, VALUE_1); + nativeCommands.zadd(KEY_2, 2D, VALUE_2); + + connection.zSetCommands().zDiffStore(KEY_3_BBUFFER, Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER)) // + .as(StepVerifier::create) // + .expectNext(1L).verifyComplete(); + } + + @ParameterizedRedisTest // GH-2042 + void zInterShouldWorkCorrectly() { assumeThat(connectionProvider).isInstanceOf(StandaloneConnectionProvider.class); @@ -566,9 +608,19 @@ public class LettuceReactiveZSetCommandsIntegrationTests extends LettuceReactive nativeCommands.zadd(KEY_2, 2D, VALUE_2); nativeCommands.zadd(KEY_2, 3D, VALUE_3); - assertThat(connection.zSetCommands() - .zUnionStore(KEY_3_BBUFFER, Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER), Arrays.asList(2D, 3D)).block()) - .isEqualTo(3L); + connection.zSetCommands().zInter(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER)) // + .collectList() // + .as(StepVerifier::create) // + .assertNext(actual -> { + assertThat(actual).contains(VALUE_1_BBUFFER, VALUE_2_BBUFFER); + }).verifyComplete(); + + connection.zSetCommands().zInterWithScores(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER), Arrays.asList(2D, 3D)) // + .collectList() // + .as(StepVerifier::create) // + .assertNext(actual -> { + assertThat(actual).contains(new DefaultTuple(VALUE_1_BYTES, 5D), new DefaultTuple(VALUE_2_BYTES, 10D)); + }).verifyComplete(); } @ParameterizedRedisTest // DATAREDIS-525 @@ -587,6 +639,49 @@ public class LettuceReactiveZSetCommandsIntegrationTests extends LettuceReactive .isEqualTo(2L); } + @ParameterizedRedisTest // GH-2042 + void zUnionShouldWorkCorrectly() { + + assumeThat(connectionProvider).isInstanceOf(StandaloneConnectionProvider.class); + + nativeCommands.zadd(KEY_1, 1D, VALUE_1); + nativeCommands.zadd(KEY_1, 2D, VALUE_2); + nativeCommands.zadd(KEY_2, 1D, VALUE_1); + nativeCommands.zadd(KEY_2, 2D, VALUE_2); + nativeCommands.zadd(KEY_2, 3D, VALUE_3); + + connection.zSetCommands().zUnion(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER)) // + .collectList() // + .as(StepVerifier::create) // + .assertNext(actual -> { + assertThat(actual).contains(VALUE_1_BBUFFER, VALUE_2_BBUFFER, VALUE_3_BBUFFER); + }).verifyComplete(); + + connection.zSetCommands().zUnionWithScores(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER), Arrays.asList(2D, 3D)) // + .collectList() // + .as(StepVerifier::create) // + .assertNext(actual -> { + assertThat(actual).contains(new DefaultTuple(VALUE_1_BYTES, 5D), new DefaultTuple(VALUE_2_BYTES, 10D), + new DefaultTuple(VALUE_3_BYTES, 9D)); + }).verifyComplete(); + } + + @ParameterizedRedisTest // DATAREDIS-525 + void zUnionStoreShouldWorkCorrectly() { + + assumeThat(connectionProvider).isInstanceOf(StandaloneConnectionProvider.class); + + nativeCommands.zadd(KEY_1, 1D, VALUE_1); + nativeCommands.zadd(KEY_1, 2D, VALUE_2); + nativeCommands.zadd(KEY_2, 1D, VALUE_1); + nativeCommands.zadd(KEY_2, 2D, VALUE_2); + nativeCommands.zadd(KEY_2, 3D, VALUE_3); + + assertThat(connection.zSetCommands() + .zUnionStore(KEY_3_BBUFFER, Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER), Arrays.asList(2D, 3D)).block()) + .isEqualTo(3L); + } + @ParameterizedRedisTest // DATAREDIS-525 void zRangeByLex() { diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java index 6c2d00e87..2667d8103 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java @@ -559,6 +559,168 @@ public class DefaultReactiveZSetOperationsIntegrationTests { .verifyComplete(); } + @ParameterizedRedisTest // GH-2041 + void difference() { + + K key = keyFactory.instance(); + K otherKey = keyFactory.instance(); + + V onlyInKey = valueFactory.instance(); + V shared = valueFactory.instance(); + V onlyInOtherKey = valueFactory.instance(); + + zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.difference(key, otherKey).as(StepVerifier::create).expectNext(onlyInKey).verifyComplete(); + + zSetOperations.differenceWithScores(key, otherKey).as(StepVerifier::create) + .expectNext(new DefaultTypedTuple<>(onlyInKey, 10D)).verifyComplete(); + } + + @ParameterizedRedisTest // GH-2041 + void differenceAndStore() { + + K key = keyFactory.instance(); + K otherKey = keyFactory.instance(); + K destKey = keyFactory.instance(); + + V onlyInKey = valueFactory.instance(); + V shared = valueFactory.instance(); + V onlyInOtherKey = valueFactory.instance(); + + zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.differenceAndStore(key, otherKey, destKey).as(StepVerifier::create).expectNext(1L).verifyComplete(); + + zSetOperations.range(destKey, ZERO_TO_FIVE).as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + } + + @ParameterizedRedisTest // GH-2042 + @EnabledOnCommand("ZINTER") + void intersect() { + + K key = keyFactory.instance(); + K otherKey = keyFactory.instance(); + + V onlyInKey = valueFactory.instance(); + V shared = valueFactory.instance(); + V onlyInOtherKey = valueFactory.instance(); + + zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.intersect(key, otherKey).as(StepVerifier::create).expectNext(shared).verifyComplete(); + + zSetOperations.intersectWithScores(key, otherKey).as(StepVerifier::create) + .expectNext(new DefaultTypedTuple<>(shared, 22D)).verifyComplete(); + + zSetOperations.intersectWithScores(key, Collections.singleton(otherKey), Aggregate.SUM, Weights.of(1, 2)) + .as(StepVerifier::create).expectNext(new DefaultTypedTuple<>(shared, 33D)).verifyComplete(); + } + + @ParameterizedRedisTest // DATAREDIS-602 + void intersectAndStore() { + + K key = keyFactory.instance(); + K otherKey = keyFactory.instance(); + K destKey = keyFactory.instance(); + + V onlyInKey = valueFactory.instance(); + V shared = valueFactory.instance(); + V onlyInOtherKey = valueFactory.instance(); + + zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.intersectAndStore(key, otherKey, destKey).as(StepVerifier::create).expectNext(1L).expectComplete() + .verify(); + + zSetOperations.range(destKey, ZERO_TO_FIVE).as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + } + + @ParameterizedRedisTest // DATAREDIS-746 + void intersectAndStoreWithAggregation() { + + K key = keyFactory.instance(); + K otherKey = keyFactory.instance(); + K destKey = keyFactory.instance(); + + V onlyInKey = valueFactory.instance(); + V shared = valueFactory.instance(); + V onlyInOtherKey = valueFactory.instance(); + + zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.intersectAndStore(key, Collections.singletonList(otherKey), destKey, Aggregate.SUM) + .as(StepVerifier::create).expectNext(1L).expectComplete().verify(); + + zSetOperations.score(destKey, shared).as(StepVerifier::create) // + .expectNext(22d) // + .verifyComplete(); + + zSetOperations.intersectAndStore(key, Collections.singletonList(otherKey), destKey, Aggregate.SUM, Weights.of(1, 2)) + .as(StepVerifier::create).expectNext(1L).expectComplete().verify(); + + zSetOperations.score(destKey, shared).as(StepVerifier::create) // + .expectNext(33d) // + .verifyComplete(); + } + + @ParameterizedRedisTest // GH-2042 + @EnabledOnCommand("ZUNION") + void union() { + + K key = keyFactory.instance(); + K otherKey = keyFactory.instance(); + + V onlyInKey = valueFactory.instance(); + V shared = valueFactory.instance(); + V onlyInOtherKey = valueFactory.instance(); + + zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); + zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); + + zSetOperations.union(key, otherKey).as(StepVerifier::create).expectNextCount(3).verifyComplete(); + + zSetOperations.unionWithScores(key, otherKey).collectList().as(StepVerifier::create).assertNext(actual -> { + assertThat(actual).containsOnly(new DefaultTypedTuple<>(onlyInKey, 10D), new DefaultTypedTuple<>(shared, 22D), + new DefaultTypedTuple<>(onlyInOtherKey, 10D)); + + }).verifyComplete(); + + zSetOperations.unionWithScores(key, Collections.singleton(otherKey), Aggregate.SUM, Weights.of(1, 2)).collectList() + .as(StepVerifier::create).assertNext(actual -> { + assertThat(actual).containsOnly(new DefaultTypedTuple<>(onlyInKey, 10D), new DefaultTypedTuple<>(shared, 33D), + new DefaultTypedTuple<>(onlyInOtherKey, 20D)); + + }).verifyComplete(); + } + @ParameterizedRedisTest // DATAREDIS-602 void unionAndStore() { @@ -607,64 +769,6 @@ public class DefaultReactiveZSetOperationsIntegrationTests { zSetOperations.score(destKey, shared).as(StepVerifier::create).expectNext(33d).verifyComplete(); } - @ParameterizedRedisTest // DATAREDIS-602 - void intersectAndStore() { - - K key = keyFactory.instance(); - K otherKey = keyFactory.instance(); - K destKey = keyFactory.instance(); - - V onlyInKey = valueFactory.instance(); - V shared = valueFactory.instance(); - V onlyInOtherKey = valueFactory.instance(); - - zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); - zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); - - zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); - zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); - - zSetOperations.intersectAndStore(key, otherKey, destKey).as(StepVerifier::create).expectNext(1L).expectComplete() - .verify(); - - zSetOperations.range(destKey, ZERO_TO_FIVE).as(StepVerifier::create) // - .expectNextCount(1) // - .verifyComplete(); - } - - @ParameterizedRedisTest // DATAREDIS-746 - void intersectAndStoreWithAggregation() { - - K key = keyFactory.instance(); - K otherKey = keyFactory.instance(); - K destKey = keyFactory.instance(); - - V onlyInKey = valueFactory.instance(); - V shared = valueFactory.instance(); - V onlyInOtherKey = valueFactory.instance(); - - zSetOperations.add(key, onlyInKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); - zSetOperations.add(key, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); - - zSetOperations.add(otherKey, onlyInOtherKey, 10).as(StepVerifier::create).expectNext(true).verifyComplete(); - zSetOperations.add(otherKey, shared, 11).as(StepVerifier::create).expectNext(true).verifyComplete(); - - zSetOperations.intersectAndStore(key, Collections.singletonList(otherKey), destKey, Aggregate.SUM) - .as(StepVerifier::create) - .expectNext(1L).expectComplete().verify(); - - zSetOperations.score(destKey, shared).as(StepVerifier::create) // - .expectNext(22d) // - .verifyComplete(); - - zSetOperations.intersectAndStore(key, Collections.singletonList(otherKey), destKey, Aggregate.SUM, Weights.of(1, 2)) - .as(StepVerifier::create).expectNext(1L).expectComplete().verify(); - - zSetOperations.score(destKey, shared).as(StepVerifier::create) // - .expectNext(33d) // - .verifyComplete(); - } - @ParameterizedRedisTest // DATAREDIS-602 void rangeByLex() {