diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index e1b32cacc..7617df63d 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java @@ -15,17 +15,8 @@ */ package org.springframework.data.redis.connection; -import java.util.ArrayList; -import java.util.Collection; -import java.util.HashMap; -import java.util.LinkedHashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.Map.Entry; -import java.util.Properties; -import java.util.Queue; -import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; @@ -1313,6 +1304,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco return convertAndReturn(delegate.zInterStore(destKey, aggregate, weights, sets), identityConverter); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zInterStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zInterStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { + return convertAndReturn(delegate.zInterStore(destKey, aggregate, weights, sets), identityConverter); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisZSetCommands#zInterStore(byte[], byte[][]) @@ -1573,6 +1573,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco return convertAndReturn(delegate.zUnionStore(destKey, aggregate, weights, sets), identityConverter); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zUnionStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zUnionStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { + return convertAndReturn(delegate.zUnionStore(destKey, aggregate, weights, sets), identityConverter); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisZSetCommands#zUnionStore(byte[], byte[][]) diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java index 55956d29d..110836632 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java @@ -685,6 +685,13 @@ public interface DefaultedRedisConnection extends RedisConnection { return zSetCommands().zInterStore(destKey, aggregate, weights, sets); } + /** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */ + @Override + @Deprecated + default Long zInterStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { + return zSetCommands().zInterStore(destKey, aggregate, weights, sets); + } + /** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */ @Override @Deprecated @@ -811,6 +818,13 @@ public interface DefaultedRedisConnection extends RedisConnection { return zSetCommands().zUnionStore(destKey, aggregate, weights, sets); } + /** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */ + @Override + @Deprecated + default Long zUnionStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { + return zSetCommands().zUnionStore(destKey, aggregate, weights, sets); + } + /** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */ @Override @Deprecated 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 e6c4c95f9..70d4a2ebb 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java @@ -34,6 +34,7 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Numeric import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -1366,6 +1367,17 @@ public interface ReactiveZSetCommands { return new ZUnionStoreCommand(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 ZUnionStoreCommand} with {@literal weights} applied. + * @since 2.1 + */ + public ZUnionStoreCommand applyWeights(Weights weights) { + return new ZUnionStoreCommand(getKey(), sourceKeys, weights.toList(), aggregateFunction); + } + /** * Applies a specific {@link Aggregate} function. Constructs a new command instance with all previously configured * properties. @@ -1440,6 +1452,21 @@ public interface ReactiveZSetCommands { 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. @@ -1462,6 +1489,29 @@ public interface ReactiveZSetCommands { .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. @@ -1517,6 +1567,17 @@ public interface ReactiveZSetCommands { return new ZInterStoreCommand(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 ZInterStoreCommand} with {@literal weights} applied. + * @since 2.1 + */ + public ZInterStoreCommand applyWeights(Weights weights) { + return new ZInterStoreCommand(getKey(), sourceKeys, weights.toList(), aggregateFunction); + } + /** * Applies a specific {@link Aggregate} function. Constructs a new command instance with all previously configured * properties. @@ -1591,6 +1652,21 @@ public interface ReactiveZSetCommands { return zInterStore(destinationKey, sets, weights, null); } + /** + * Intersect 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: ZINTERSTORE + */ + default Mono zInterStore(ByteBuffer destinationKey, List sets, Weights weights) { + return zInterStore(destinationKey, sets, weights, null); + } + /** * Intersect sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination * {@literal destinationKey} and apply weights to individual sets. @@ -1613,6 +1689,29 @@ public interface ReactiveZSetCommands { .next().map(NumericResponse::getOutput); } + /** + * Intersect 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 must not be {@literal null}. + * @param aggregateFunction can be {@literal null}. + * @return + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + default Mono zInterStore(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 zInterStore(Mono.just( + ZInterStoreCommand.sets(sets).aggregateUsing(aggregateFunction).applyWeights(weights).storeAs(destinationKey))) + .next().map(NumericResponse::getOutput); + } + /** * Intersect sorted {@literal sets} by applying {@literal aggregateFunction} and store result in destination * {@literal destinationKey} and apply weights to individual sets. 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 226632f71..3b6655197 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java @@ -15,7 +15,13 @@ */ package org.springframework.data.redis.connection; +import java.util.Arrays; +import java.util.List; import java.util.Set; +import java.util.function.DoubleUnaryOperator; +import java.util.stream.Collectors; +import java.util.stream.DoubleStream; +import java.util.stream.IntStream; import org.springframework.data.redis.core.Cursor; import org.springframework.data.redis.core.ScanOptions; @@ -40,6 +46,128 @@ public interface RedisZSetCommands { SUM, MIN, MAX; } + /** + * Value object encapsulating a multiplication factor for each input sorted set. This means that the score of every + * element in every input sorted set is multiplied by this factor before being passed to the aggregation function. + * + * @author Mark Paluch + * @since 2.1 + */ + class Weights { + + private final double[] weights; + + Weights(double[] weights) { + this.weights = weights; + } + + /** + * Create new {@link Weights} given {@code weights} as {@code int}. + * + * @param weights must not be {@literal null}. + * @return the {@link Weights} for {@code weights}. + */ + public static Weights of(int... weights) { + + Assert.notNull(weights, "Weights must not be null!"); + return new Weights(Arrays.stream(weights).mapToDouble(value -> value).toArray()); + } + + /** + * Create new {@link Weights} given {@code weights} as {@code double}. + * + * @param weights must not be {@literal null}. + * @return the {@link Weights} for {@code weights}. + */ + public static Weights of(double... weights) { + + Assert.notNull(weights, "Weights must not be null!"); + + return new Weights(Arrays.copyOf(weights, weights.length)); + } + + /** + * Creates equal {@link Weights} for a number of input sets {@code count} with a weight of one. + * + * @param count number of input sets. Must be greater or equal to zero. + * @return equal {@link Weights} for a number of input sets with a weight of one. + */ + public static Weights fromSetCount(int count) { + + Assert.isTrue(count >= 0, "Count of input sorted sets must be greater or equal to zero!"); + + return new Weights(IntStream.range(0, count).mapToDouble(value -> 1).toArray()); + } + + /** + * Creates a new {@link Weights} object that contains all weights multiplied by {@code multiplier} + * + * @param multiplier multiplier used to multiply each weight with. + * @return equal {@link Weights} for a number of input sets with a weight of one. + */ + public Weights multiply(int multiplier) { + return apply(it -> it * multiplier); + } + + /** + * Creates a new {@link Weights} object that contains all weights multiplied by {@code multiplier} + * + * @param multiplier multiplier used to multiply each weight with. + * @return equal {@link Weights} for a number of input sets with a weight of one. + */ + public Weights multiply(double multiplier) { + return apply(it -> it * multiplier); + } + + /** + * Creates a new {@link Weights} object that contains all weights with {@link DoubleUnaryOperator} applied. + * + * @param operator operator function. + * @return the new {@link Weights} with {@link DoubleUnaryOperator} applied. + */ + public Weights apply(DoubleUnaryOperator operator) { + return new Weights(DoubleStream.of(weights).map(operator).toArray()); + } + + /** + * Retrieve the weight at {@code index}. + * + * @param index the weight index. + * @return the weight at {@code index}. + * @throws IndexOutOfBoundsException if the index is out of range + */ + public double getWeight(int index) { + + if (index > size() || index < 0) { + throw new IndexOutOfBoundsException("No such weight"); + } + + return weights[index]; + } + + /** + * @return number of weights. + */ + public int size() { + return weights.length; + } + + /** + * @return an array containing all of the weights in this list in proper sequence (from first to last element). + */ + public double[] toArray() { + return Arrays.copyOf(weights, weights.length); + } + + /** + * @return a {@link List} containing all of the weights in this list in proper sequence (from first to last + * element). + */ + public List toList() { + return Arrays.stream(weights).boxed().collect(Collectors.toList()); + } + } + /** * ZSet tuple. */ @@ -687,7 +815,7 @@ public interface RedisZSetCommands { * * @param destKey must not be {@literal null}. * @param aggregate must not be {@literal null}. - * @param weights + * @param weights must not be {@literal null}. * @param sets must not be {@literal null}. * @return {@literal null} when used in pipeline / transaction. * @see Redis Documentation: ZUNIONSTORE @@ -695,6 +823,20 @@ public interface RedisZSetCommands { @Nullable Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets); + /** + * Union sorted {@code sets} and store result in destination {@code key}. + * + * @param destKey must not be {@literal null}. + * @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.1 + * @see Redis Documentation: ZUNIONSTORE + */ + @Nullable + Long zUnionStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets); + /** * Intersect sorted {@code sets} and store result in destination {@code key}. * @@ -719,6 +861,20 @@ public interface RedisZSetCommands { @Nullable Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets); + /** + * Intersect sorted {@code sets} and store result in destination {@code key}. + * + * @param destKey must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @param weights must not be {@literal null}. + * @param sets must not be {@literal null}. + * @return + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + @Nullable + Long zInterStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets); + /** * Use a {@link Cursor} to iterate over elements in sorted set at {@code key}. * diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterZSetCommands.java index 1f83f88ef..47488958a 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterZSetCommands.java @@ -631,16 +631,28 @@ class JedisClusterZSetCommands implements RedisZSetCommands { */ @Override public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + return zUnionStore(destKey, aggregate, Weights.of(weights), sets); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zUnionStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zUnionStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { Assert.notNull(destKey, "Destination key must not be null!"); Assert.notNull(sets, "Source sets must not be null!"); Assert.noNullElements(sets, "Source sets must not contain null elements!"); + Assert.isTrue(weights.size() == sets.length, () -> String + .format("The number of weights (%d) must match the number of source sets (%d)!", weights.size(), sets.length)); byte[][] allKeys = ByteUtils.mergeArrays(destKey, sets); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { - ZParams zparams = new ZParams().weights(weights).aggregate(ZParams.Aggregate.valueOf(aggregate.name())); + ZParams zparams = new ZParams().weightsByDouble(weights.toArray()) + .aggregate(ZParams.Aggregate.valueOf(aggregate.name())); try { return connection.getCluster().zunionstore(destKey, zparams, sets); @@ -683,16 +695,28 @@ class JedisClusterZSetCommands implements RedisZSetCommands { */ @Override public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + return zInterStore(destKey, aggregate, Weights.of(weights), sets); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zInterStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zInterStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { Assert.notNull(destKey, "Destination key must not be null!"); Assert.notNull(sets, "Source sets must not be null!"); Assert.noNullElements(sets, "Source sets must not contain null elements!"); + Assert.isTrue(weights.size() == sets.length, () -> String + .format("The number of weights (%d) must match the number of source sets (%d)!", weights.size(), sets.length)); byte[][] allKeys = ByteUtils.mergeArrays(destKey, sets); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { - ZParams zparams = new ZParams().weights(weights).aggregate(ZParams.Aggregate.valueOf(aggregate.name())); + ZParams zparams = new ZParams().weightsByDouble(weights.toArray()) + .aggregate(ZParams.Aggregate.valueOf(aggregate.name())); try { return connection.getCluster().zinterstore(destKey, zparams, sets); diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java index fee322603..7d36a1647 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java @@ -585,13 +585,26 @@ class JedisZSetCommands implements RedisZSetCommands { */ @Override public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + return zUnionStore(destKey, aggregate, Weights.of(weights), sets); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zUnionStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zUnionStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { Assert.notNull(destKey, "Destination key must not be null!"); Assert.notNull(sets, "Source sets must not be null!"); + Assert.notNull(weights, "Weights must not be null!"); Assert.noNullElements(sets, "Source sets must not contain null elements!"); + Assert.isTrue(weights.size() == sets.length, () -> String + .format("The number of weights (%d) must match the number of source sets (%d)!", weights.size(), sets.length)); try { - ZParams zparams = new ZParams().weights(weights).aggregate(ZParams.Aggregate.valueOf(aggregate.name())); + ZParams zparams = new ZParams().weightsByDouble(weights.toArray()) + .aggregate(ZParams.Aggregate.valueOf(aggregate.name())); if (isPipelined()) { pipeline(connection.newJedisResult(connection.getRequiredPipeline().zunionstore(destKey, zparams, sets))); @@ -639,13 +652,25 @@ class JedisZSetCommands implements RedisZSetCommands { */ @Override public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + return zInterStore(destKey, aggregate, Weights.of(weights), sets); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zInterStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zInterStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { Assert.notNull(destKey, "Destination key must not be null!"); Assert.notNull(sets, "Source sets must not be null!"); Assert.noNullElements(sets, "Source sets must not contain null elements!"); + Assert.isTrue(weights.size() == sets.length, () -> String + .format("The number of weights (%d) must match the number of source sets (%d)!", weights.size(), sets.length)); try { - ZParams zparams = new ZParams().weights(weights).aggregate(ZParams.Aggregate.valueOf(aggregate.name())); + ZParams zparams = new ZParams().weightsByDouble(weights.toArray()) + .aggregate(ZParams.Aggregate.valueOf(aggregate.name())); if (isPipelined()) { pipeline(connection.newJedisResult(connection.getRequiredPipeline().zinterstore(destKey, zparams, sets))); 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 a33e2caab..41a7f5ae4 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 @@ -39,6 +39,7 @@ import org.springframework.data.redis.connection.ReactiveZSetCommands; import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; import org.springframework.data.redis.util.ByteUtils; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -466,7 +467,7 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { })); } - private ZStoreArgs zStoreArgs(Aggregate aggregate, List weights) { + private static ZStoreArgs zStoreArgs(@Nullable Aggregate aggregate, @Nullable List weights) { ZStoreArgs args = new ZStoreArgs(); if (aggregate != null) { @@ -484,12 +485,9 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { } if (weights != null) { - double[] lg = new double[weights.size()]; - for (int i = 0; i < lg.length; i++) { - lg[i] = weights.get(i).longValue(); - } - args.weights(lg); + args.weights(weights.stream().mapToDouble(it -> it).toArray()); } + return args; } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java index 04431f262..d2da37e75 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java @@ -33,6 +33,7 @@ import org.springframework.data.redis.core.Cursor; import org.springframework.data.redis.core.KeyBoundCursor; import org.springframework.data.redis.core.ScanIteration; import org.springframework.data.redis.core.ScanOptions; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -574,10 +575,21 @@ class LettuceZSetCommands implements RedisZSetCommands { */ @Override public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + return zUnionStore(destKey, aggregate, Weights.of(weights), sets); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zUnionStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zUnionStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { Assert.notNull(destKey, "Destination key must not be null!"); Assert.notNull(sets, "Source sets must not be null!"); Assert.noNullElements(sets, "Source sets must not contain null elements!"); + Assert.isTrue(weights.size() == sets.length, () -> String + .format("The number of weights (%d) must match the number of source sets (%d)!", weights.size(), sets.length)); ZStoreArgs storeArgs = zStoreArgs(aggregate, weights); @@ -628,10 +640,21 @@ class LettuceZSetCommands implements RedisZSetCommands { */ @Override public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { + return zInterStore(destKey, aggregate, Weights.of(weights), sets); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisZSetCommands#zInterStore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights, byte[][]) + */ + @Override + public Long zInterStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { Assert.notNull(destKey, "Destination key must not be null!"); Assert.notNull(sets, "Source sets must not be null!"); Assert.noNullElements(sets, "Source sets must not contain null elements!"); + Assert.isTrue(weights.size() == sets.length, () -> String + .format("The number of weights (%d) must match the number of source sets (%d)!", weights.size(), sets.length)); ZStoreArgs storeArgs = zStoreArgs(aggregate, weights); @@ -898,8 +921,10 @@ class LettuceZSetCommands implements RedisZSetCommands { return connection.convertLettuceAccessException(ex); } - private ZStoreArgs zStoreArgs(Aggregate aggregate, int[] weights) { + private static ZStoreArgs zStoreArgs(@Nullable Aggregate aggregate, Weights weights) { + ZStoreArgs args = new ZStoreArgs(); + if (aggregate != null) { switch (aggregate) { case MIN: @@ -913,11 +938,9 @@ class LettuceZSetCommands implements RedisZSetCommands { break; } } - double[] lg = new double[weights.length]; - for (int i = 0; i < lg.length; i++) { - lg[i] = weights[i]; - } - args.weights(lg); + + args.weights(weights.toArray()); + return args; } diff --git a/src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java b/src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java index 3566610c4..2e59a8372 100644 --- a/src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java @@ -22,6 +22,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Range; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.data.redis.core.ZSetOperations.TypedTuple; import org.springframework.lang.Nullable; @@ -278,13 +279,13 @@ public interface BoundZSetOperations extends BoundKeyOperations { * Union sorted sets at the bound key and {@code otherKeys} and store result in destination {@code destKey}. * * @param otherKeys must not be {@literal null}. - * @param weights 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}. * @since 2.1 * @see Redis Documentation: ZUNIONSTORE */ - void unionAndStore(Collection otherKeys, int[] weights, K destKey, Aggregate aggregate); + void unionAndStore(Collection otherKeys, K destKey, Aggregate aggregate, Weights weights); /** * Intersect sorted sets at the bound key and {@code otherKey} and store result in destination {@code destKey}. @@ -304,6 +305,29 @@ public interface BoundZSetOperations extends BoundKeyOperations { */ void intersectAndStore(Collection otherKeys, K destKey); + /** + * Intersect sorted sets at the bound key and {@code otherKeys} and store result in destination {@code destKey}. + * + * @param otherKeys must not be {@literal null}. + * @param destKey must not be {@literal null}. + * @param aggregate must not be {@literal null}. + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + void intersectAndStore(Collection otherKeys, K destKey, Aggregate aggregate); + + /** + * Intersect sorted sets at the bound key and {@code otherKeys} and store result in destination {@code destKey}. + * + * @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}. + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + void intersectAndStore(Collection otherKeys, K destKey, Aggregate aggregate, Weights weights); + /** * Iterate over elements in zset at the bound key.
* Important: Call {@link Cursor#close()} when done to avoid resource leak. diff --git a/src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java index 31233cf97..2244d382b 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java @@ -20,9 +20,10 @@ import java.util.Collection; import java.util.Set; import org.springframework.data.redis.connection.DataType; -import org.springframework.data.redis.connection.RedisZSetCommands; +import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Range; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.data.redis.core.ZSetOperations.TypedTuple; /** @@ -103,6 +104,24 @@ class DefaultBoundZSetOperations extends DefaultBoundKeyOperations impl ops.intersectAndStore(getKey(), otherKeys, destKey); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.BoundZSetOperations#intersectAndStore(java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate) + */ + @Override + public void intersectAndStore(Collection otherKeys, K destKey, Aggregate aggregate) { + ops.intersectAndStore(getKey(), otherKeys, destKey, aggregate); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.BoundZSetOperations#intersectAndStore(java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights) + */ + @Override + public void intersectAndStore(Collection otherKeys, K destKey, Aggregate aggregate, Weights weights) { + ops.intersectAndStore(getKey(), otherKeys, destKey, aggregate, weights); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.core.BoundZSetOperations#range(long, long) @@ -292,22 +311,22 @@ class DefaultBoundZSetOperations extends DefaultBoundKeyOperations impl ops.unionAndStore(getKey(), otherKeys, destKey); } - /* + /* * (non-Javadoc) - * @see org.springframework.data.redis.core.BoundZSetOperations#unionAndStore(Collection, Object, RedisZSetCommands.Aggregate) + * @see org.springframework.data.redis.core.BoundZSetOperations#unionAndStore(java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate) */ @Override - public void unionAndStore(Collection otherKeys, K destKey, RedisZSetCommands.Aggregate aggregate) { + public void unionAndStore(Collection otherKeys, K destKey, Aggregate aggregate) { ops.unionAndStore(getKey(), otherKeys, destKey, aggregate); } - /* + /* * (non-Javadoc) - * @see org.springframework.data.redis.core.BoundZSetOperations#unionAndStore(Collection, int[], Object, RedisZSetCommands.Aggregate) + * @see org.springframework.data.redis.core.BoundZSetOperations#unionAndStore(java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate, org.springframework.data.redis.connection.RedisZSetCommands.Weights) */ @Override - public void unionAndStore(Collection otherKeys, int[] weights, K destKey, RedisZSetCommands.Aggregate aggregate) { - ops.unionAndStore(getKey(), otherKeys, weights, destKey, aggregate); + public void unionAndStore(Collection otherKeys, K destKey, Aggregate aggregate, Weights weights) { + ops.unionAndStore(getKey(), otherKeys, destKey, aggregate, weights); } /* 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 0d1abe79d..ef97deed4 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java @@ -31,8 +31,10 @@ import org.reactivestreams.Publisher; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.DefaultTuple; import org.springframework.data.redis.connection.ReactiveZSetCommands; +import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.data.redis.core.ZSetOperations.TypedTuple; import org.springframework.data.redis.serializer.RedisSerializationContext; import org.springframework.data.redis.util.ByteUtils; @@ -340,7 +342,8 @@ class DefaultReactiveZSetOperations implements ReactiveZSetOperations connection.zRemRangeByScore(rawKey(key), range)); } - /* (non-Javadoc) + /* + * (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionAndStore(java.lang.Object, java.lang.Object, java.lang.Object) */ @Override @@ -353,7 +356,8 @@ class DefaultReactiveZSetOperations implements ReactiveZSetOperations implements ReactiveZSetOperations connection.zUnionStore(rawKey(destKey), serialized))); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionAndStore(java.lang.Object, java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate) + */ + @Override + public Mono unionAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate) { + + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return unionAndStore(key, otherKeys, destKey, aggregate, Weights.fromSetCount(1 + otherKeys.size())); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionAndStore(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 unionAndStore(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.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 @@ -382,7 +418,8 @@ class DefaultReactiveZSetOperations implements ReactiveZSetOperations implements ReactiveZSetOperations 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) + */ + @Override + public Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate) { + + Assert.notNull(otherKeys, "Other keys must not be null!"); + + return intersectAndStore(key, otherKeys, destKey, aggregate, Weights.fromSetCount(1 + otherKeys.size())); + } + + /* + * (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/DefaultZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultZSetOperations.java index 671d957a6..f1fb1e565 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultZSetOperations.java @@ -19,10 +19,11 @@ import java.util.Collection; import java.util.Collections; import java.util.Set; -import org.springframework.data.redis.connection.RedisZSetCommands; +import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Range; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; /** * Default implementation of {@link ZSetOperations}. @@ -94,9 +95,32 @@ class DefaultZSetOperations extends AbstractOperations implements ZS byte[][] rawKeys = rawKeys(key, otherKeys); byte[] rawDestKey = rawKey(destKey); + return execute(connection -> connection.zInterStore(rawDestKey, rawKeys), true); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ZSetOperations#intersectAndStore(java.lang.Object, java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate) + */ + @Override + public Long intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate) { + return intersectAndStore(key, otherKeys, destKey, aggregate, Weights.fromSetCount(1 + otherKeys.size())); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.ZSetOperations#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 Long intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights) { + + byte[][] rawKeys = rawKeys(key, otherKeys); + byte[] rawDestKey = rawKey(destKey); + + return execute(connection -> connection.zInterStore(rawDestKey, aggregate, weights, rawKeys), true); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.core.ZSetOperations#range(java.lang.Object, long, long) @@ -402,8 +426,10 @@ class DefaultZSetOperations extends AbstractOperations implements ZS */ @Override public Long unionAndStore(K key, Collection otherKeys, K destKey) { + byte[][] rawKeys = rawKeys(key, otherKeys); byte[] rawDestKey = rawKey(destKey); + return execute(connection -> connection.zUnionStore(rawDestKey, rawKeys), true); } @@ -412,22 +438,20 @@ class DefaultZSetOperations extends AbstractOperations implements ZS * @see org.springframework.data.redis.core.ZSetOperations#unionAndStore(java.lang.Object, java.util.Collection, java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate) */ @Override - public Long unionAndStore(K key, Collection otherKeys, K destKey, RedisZSetCommands.Aggregate aggregate) { - int weights[] = new int[otherKeys.size() + (key != null ? 1 : 0)]; - for (int i = 0; i < weights.length; i++) { - weights[i] = 1; - } - return unionAndStore(key, otherKeys, weights, destKey, aggregate); + public Long unionAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate) { + return unionAndStore(key, otherKeys, destKey, aggregate, Weights.fromSetCount(1 + otherKeys.size())); } /* * (non-Javadoc) - * @see org.springframework.data.redis.core.ZSetOperations#unionAndStore(java.lang.Object, java.util.Collection, int[], java.lang.Object, org.springframework.data.redis.connection.RedisZSetCommands.Aggregate) + * @see org.springframework.data.redis.core.ZSetOperations#unionAndStore(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 Long unionAndStore(K key, Collection otherKeys, int[] weights, K destKey, RedisZSetCommands.Aggregate aggregate) { + public Long unionAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights) { + byte[][] rawKeys = rawKeys(key, otherKeys); byte[] rawDestKey = rawKey(destKey); + return execute(connection -> connection.zUnionStore(rawDestKey, aggregate, weights, rawKeys), true); } 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 e4957aab0..3468118ac 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java @@ -21,8 +21,10 @@ import reactor.core.publisher.Mono; import java.util.Collection; import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.data.redis.core.ZSetOperations.TypedTuple; /** @@ -297,6 +299,33 @@ public interface ReactiveZSetOperations { */ Mono unionAndStore(K key, Collection otherKeys, K destKey); + /** + * Union 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: ZUNIONSTORE + */ + Mono unionAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate); + + /** + * Union 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: ZUNIONSTORE + */ + 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}. * @@ -319,6 +348,33 @@ public interface ReactiveZSetOperations { */ 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 + */ + Mono intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate); + + /** + * 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/main/java/org/springframework/data/redis/core/ZSetOperations.java b/src/main/java/org/springframework/data/redis/core/ZSetOperations.java index 100c9eadc..94a1c3640 100644 --- a/src/main/java/org/springframework/data/redis/core/ZSetOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ZSetOperations.java @@ -22,6 +22,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.RedisZSetCommands.Range; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.lang.Nullable; /** @@ -384,15 +385,15 @@ public interface ZSetOperations { * * @param key must not be {@literal null}. * @param otherKeys must not be {@literal null}. - * @param weights 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 {@literal null} when used in pipeline / transaction. * @since 2.1 * @see Redis Documentation: ZUNIONSTORE */ @Nullable - Long unionAndStore(K key, Collection otherKeys, int[] weights, K destKey, Aggregate aggregate); + Long 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}. @@ -418,6 +419,35 @@ public interface ZSetOperations { @Nullable Long 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 {@literal null} when used in pipeline / transaction. + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + @Nullable + Long intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate); + + /** + * 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 {@literal null} when used in pipeline / transaction. + * @since 2.1 + * @see Redis Documentation: ZINTERSTORE + */ + @Nullable + Long intersectAndStore(K key, Collection otherKeys, K destKey, Aggregate aggregate, Weights weights); + /** * Iterate over elements in zset at {@code key}.
* Important: Call {@link Cursor#close()} when done to avoid resource leak. diff --git a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java index a3749f362..dd2960530 100644 --- a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java @@ -844,6 +844,10 @@ public class RedisConnectionUnitTests { return delegate.zUnionStore(destKey, aggregate, weights, sets); } + public Long zUnionStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { + return delegate.zUnionStore(destKey, aggregate, weights, sets); + } + public Long zInterStore(byte[] destKey, byte[]... sets) { return delegate.zInterStore(destKey, sets); } @@ -852,6 +856,10 @@ public class RedisConnectionUnitTests { return delegate.zInterStore(destKey, aggregate, weights, sets); } + public Long zInterStore(byte[] destKey, Aggregate aggregate, Weights weights, byte[]... sets) { + return delegate.zInterStore(destKey, aggregate, weights, sets); + } + public Cursor zScan(byte[] key, ScanOptions options) { return delegate.zScan(key, options); } diff --git a/src/test/java/org/springframework/data/redis/connection/WeightsUnitTests.java b/src/test/java/org/springframework/data/redis/connection/WeightsUnitTests.java new file mode 100644 index 000000000..250181ace --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/WeightsUnitTests.java @@ -0,0 +1,75 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection; + +import static org.assertj.core.api.Assertions.*; + +import org.junit.Test; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; + +/** + * Unit tests for {@link org.springframework.data.redis.connection.RedisZSetCommands.Weights}. + * + * @author Mark Paluch + */ +public class WeightsUnitTests { + + @Test // DATAREDIS-746 + public void shouldCreateWeights() { + + assertThat(Weights.of(1, 2, 3).toArray()).contains(1, 2, 3); + assertThat(Weights.of(1, 2d, 3).toArray()).contains(1d, 2d, 3d); + } + + @Test // DATAREDIS-746 + public void shouldRejectCreationWithNull() { + + assertThatThrownBy(() -> Weights.of((int[]) null)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> Weights.of((double[]) null)).isInstanceOf(IllegalArgumentException.class); + } + + @Test // DATAREDIS-746 + public void shouldCreateEqualWeights() { + + Weights weights = Weights.fromSetCount(3); + assertThat(weights.getWeight(0)).isOne(); + assertThat(weights.getWeight(1)).isOne(); + assertThat(weights.getWeight(2)).isOne(); + } + + @Test // DATAREDIS-746 + public void getShouldThrowIndexOutOfBoundsException() { + + assertThatThrownBy(() -> Weights.fromSetCount(1).getWeight(1)).isInstanceOf(IndexOutOfBoundsException.class); + assertThatThrownBy(() -> Weights.fromSetCount(1).getWeight(-1)).isInstanceOf(IndexOutOfBoundsException.class); + } + + @Test // DATAREDIS-746 + public void shouldMultiplyDouble() { + + Weights weights = Weights.of(1, 2, 3).multiply(2.5); + assertThat(weights.getWeight(0)).isEqualTo(2.5); + assertThat(weights.getWeight(2)).isEqualTo(7.5); + } + + @Test // DATAREDIS-746 + public void shouldMultiplyInt() { + + Weights weights = Weights.of(1, 2, 3).multiply(2); + assertThat(weights.getWeight(0)).isEqualTo(2); + assertThat(weights.getWeight(2)).isEqualTo(6); + } +} 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 4f996b770..d71930640 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java @@ -21,6 +21,7 @@ import reactor.test.StepVerifier; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; import org.junit.AfterClass; @@ -35,7 +36,9 @@ import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.ObjectFactory; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; @@ -484,6 +487,33 @@ public class DefaultReactiveZSetOperationsIntegrationTests { StepVerifier.create(zSetOperations.range(destKey, new Range<>(0L, 100L))).expectNextCount(3).verifyComplete(); } + @Test // DATAREDIS-746 + public void unionAndStoreWithAggregation() { + + K key = keyFactory.instance(); + K otherKey = keyFactory.instance(); + K destKey = keyFactory.instance(); + + V onlyInKey = valueFactory.instance(); + V shared = valueFactory.instance(); + V onlyInOtherKey = valueFactory.instance(); + + StepVerifier.create(zSetOperations.add(key, onlyInKey, 10)).expectNext(true).verifyComplete(); + StepVerifier.create(zSetOperations.add(key, shared, 11)).expectNext(true).verifyComplete(); + + StepVerifier.create(zSetOperations.add(otherKey, onlyInOtherKey, 10)).expectNext(true).verifyComplete(); + StepVerifier.create(zSetOperations.add(otherKey, shared, 11)).expectNext(true).verifyComplete(); + + StepVerifier.create(zSetOperations.unionAndStore(key, Collections.singleton(otherKey), destKey, Aggregate.SUM)) + .expectNext(3L).verifyComplete(); + StepVerifier.create(zSetOperations.score(destKey, shared)).expectNext(22d).verifyComplete(); + + StepVerifier.create( + zSetOperations.unionAndStore(key, Collections.singleton(otherKey), destKey, Aggregate.SUM, Weights.of(2, 1))) + .expectNext(3L).verifyComplete(); + StepVerifier.create(zSetOperations.score(destKey, shared)).expectNext(33d).verifyComplete(); + } + @Test // DATAREDIS-602 public void intersectAndStore() { @@ -507,7 +537,39 @@ public class DefaultReactiveZSetOperationsIntegrationTests { StepVerifier.create(zSetOperations.range(destKey, new Range<>(0L, 5L))) // .expectNextCount(1) // .verifyComplete(); + } + @Test // DATAREDIS-746 + public 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(); + + StepVerifier.create(zSetOperations.add(key, onlyInKey, 10)).expectNext(true).verifyComplete(); + StepVerifier.create(zSetOperations.add(key, shared, 11)).expectNext(true).verifyComplete(); + + StepVerifier.create(zSetOperations.add(otherKey, onlyInOtherKey, 10)).expectNext(true).verifyComplete(); + StepVerifier.create(zSetOperations.add(otherKey, shared, 11)).expectNext(true).verifyComplete(); + + StepVerifier + .create(zSetOperations.intersectAndStore(key, Collections.singletonList(otherKey), destKey, Aggregate.SUM)) + .expectNext(1L).expectComplete().verify(); + + StepVerifier.create(zSetOperations.score(destKey, shared)) // + .expectNext(22d) // + .verifyComplete(); + + StepVerifier.create(zSetOperations.intersectAndStore(key, Collections.singletonList(otherKey), destKey, + Aggregate.SUM, Weights.of(1, 2))).expectNext(1L).expectComplete().verify(); + + StepVerifier.create(zSetOperations.score(destKey, shared)) // + .expectNext(33d) // + .verifyComplete(); } @Test // DATAREDIS-602 diff --git a/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsTests.java index 5ac8b2006..a61944241 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsTests.java @@ -42,6 +42,7 @@ import org.springframework.data.redis.LongAsStringObjectFactory; import org.springframework.data.redis.LongObjectFactory; import org.springframework.data.redis.ObjectFactory; import org.springframework.data.redis.connection.RedisZSetCommands; +import org.springframework.data.redis.connection.RedisZSetCommands.Weights; import org.springframework.data.redis.core.ZSetOperations.TypedTuple; import org.springframework.data.redis.test.util.MinimumRedisVersionRule; import org.springframework.test.annotation.IfProfileValue; @@ -394,9 +395,42 @@ public class DefaultZSetOperationsTests { zSetOps.add(key1, value1, 4.0); zSetOps.add(key2, value1, 3.0); - int weight[] = { 1, 2 }; + zSetOps.unionAndStore(key1, Collections.singletonList(key2), key1, RedisZSetCommands.Aggregate.MAX, + Weights.of(1, 2)); - zSetOps.unionAndStore(key1, Collections.singletonList(key2), weight, key1, RedisZSetCommands.Aggregate.MAX); + assertThat(zSetOps.score(key1, value1), closeTo(6.0, 0.1)); + } + + @Test // DATAREDIS-746 + public void testZsetIntersectWithAggregate() { + + K key1 = keyFactory.instance(); + K key2 = keyFactory.instance(); + + V value1 = valueFactory.instance(); + V value2 = valueFactory.instance(); + + zSetOps.add(key1, value1, 1.0); + zSetOps.add(key1, value2, 2.0); + zSetOps.add(key2, value2, 3.0); + + zSetOps.intersectAndStore(key1, Collections.singletonList(key2), key1, RedisZSetCommands.Aggregate.MIN); + + assertThat(zSetOps.score(key1, value2), closeTo(2.0, 0.1)); + } + + @Test // DATAREDIS-746 + public void testZsetIntersectWithAggregateWeights() { + + K key1 = keyFactory.instance(); + K key2 = keyFactory.instance(); + V value1 = valueFactory.instance(); + + zSetOps.add(key1, value1, 4.0); + zSetOps.add(key2, value1, 3.0); + + zSetOps.intersectAndStore(key1, Collections.singletonList(key2), key1, RedisZSetCommands.Aggregate.MAX, + Weights.of(1, 2)); assertThat(zSetOps.score(key1, value1), closeTo(6.0, 0.1)); }