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 f3e02e9c4..9524c161c 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java @@ -25,8 +25,10 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.EnumSet; import java.util.List; import java.util.Optional; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -37,6 +39,7 @@ 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.connection.RedisZSetCommands.ZAddArgs.Flag; import org.springframework.data.redis.core.ScanOptions; import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; @@ -61,23 +64,16 @@ public interface ReactiveZSetCommands { class ZAddCommand extends KeyCommand { private final List tuples; - private final boolean upsert; - private final boolean returnTotalChanged; + private final Set flags; private final boolean incr; - private final boolean gt; - private final boolean lt; - private ZAddCommand(@Nullable ByteBuffer key, List tuples, boolean upsert, boolean returnTotalChanged, - boolean incr, boolean gt, boolean lt) { + private ZAddCommand(@Nullable ByteBuffer key, List tuples, Set flags, boolean incr) { super(key); this.tuples = tuples; - this.upsert = upsert; - this.returnTotalChanged = returnTotalChanged; + this.flags = flags; this.incr = incr; - this.gt = gt; - this.lt = lt; } /** @@ -103,7 +99,7 @@ public interface ReactiveZSetCommands { Assert.notNull(tuples, "Tuples must not be null!"); - return new ZAddCommand(null, new ArrayList<>(tuples), false, false, false, false, false); + return new ZAddCommand(null, new ArrayList<>(tuples), EnumSet.noneOf(Flag.class), false); } /** @@ -116,7 +112,7 @@ public interface ReactiveZSetCommands { Assert.notNull(key, "Key must not be null!"); - return new ZAddCommand(key, tuples, upsert, returnTotalChanged, incr, gt, lt); + return new ZAddCommand(key, tuples, flags, incr); } /** @@ -126,7 +122,11 @@ public interface ReactiveZSetCommands { * @return a new {@link ZAddCommand} with {@literal xx} applied. */ public ZAddCommand xx() { - return new ZAddCommand(getKey(), tuples, false, returnTotalChanged, incr, gt, lt); + + EnumSet flags = EnumSet.copyOf(this.flags); + flags.remove(Flag.NX); + flags.add(Flag.XX); + return new ZAddCommand(getKey(), tuples, flags, incr); } /** @@ -136,7 +136,11 @@ public interface ReactiveZSetCommands { * @return a new {@link ZAddCommand} with {@literal nx} applied. */ public ZAddCommand nx() { - return new ZAddCommand(getKey(), tuples, true, returnTotalChanged, incr, gt, lt); + + EnumSet flags = EnumSet.copyOf(this.flags); + flags.remove(Flag.XX); + flags.add(Flag.NX); + return new ZAddCommand(getKey(), tuples, flags, incr); } /** @@ -146,17 +150,20 @@ public interface ReactiveZSetCommands { * @return a new {@link ZAddCommand} with {@literal ch} applied. */ public ZAddCommand ch() { - return new ZAddCommand(getKey(), tuples, upsert, true, incr, gt, lt); + + EnumSet flags = EnumSet.copyOf(this.flags); + flags.add(Flag.CH); + return new ZAddCommand(getKey(), tuples, flags, incr); } /** * Applies {@literal incr} mode (When this option is specified ZADD acts like ZINCRBY). Constructs a new command - * instance with all previously configured properties. + * instance with all previously configured properties. Note that the command result returns the score of the member. * * @return a new {@link ZAddCommand} with {@literal incr} applied. */ public ZAddCommand incr() { - return new ZAddCommand(getKey(), tuples, upsert, upsert, true, gt, lt); + return new ZAddCommand(getKey(), tuples, flags, true); } /** @@ -166,7 +173,11 @@ public interface ReactiveZSetCommands { * @since 2.5 */ public ZAddCommand gt() { - return new ZAddCommand(getKey(), tuples, upsert, upsert, incr, true, lt); + + EnumSet flags = EnumSet.copyOf(this.flags); + flags.remove(Flag.LT); + flags.add(Flag.GT); + return new ZAddCommand(getKey(), tuples, flags, incr); } /** @@ -176,7 +187,11 @@ public interface ReactiveZSetCommands { * @since 2.5 */ public ZAddCommand lt() { - return new ZAddCommand(getKey(), tuples, upsert, upsert, incr, gt, true); + + EnumSet flags = EnumSet.copyOf(this.flags); + flags.remove(Flag.GT); + flags.add(Flag.LT); + return new ZAddCommand(getKey(), tuples, flags, incr); } /** @@ -187,10 +202,26 @@ public interface ReactiveZSetCommands { } /** - * @return + * @return {@code true} if the command does not contain NX or XX flags. */ public boolean isUpsert() { - return upsert; + return !flags.contains(Flag.NX) && !flags.contains(Flag.XX); + } + + /** + * @return {@code true} if the command contains the XX flag. + * @since 2.7.17 + */ + public boolean isIfExists() { + return flags.contains(Flag.XX); + } + + /** + * @return {@code true} if the command contains the NX flag. + * @since 2.7.17 + */ + public boolean isIfNotExists() { + return flags.contains(Flag.NX); } /** @@ -205,7 +236,7 @@ public interface ReactiveZSetCommands { * @since 2.5 */ public boolean isGt() { - return gt; + return flags.contains(Flag.GT); } /** @@ -213,14 +244,14 @@ public interface ReactiveZSetCommands { * @since 2.5 */ public boolean isLt() { - return lt; + return flags.contains(Flag.LT); } /** * @return */ public boolean isReturnTotalChanged() { - return returnTotalChanged; + return flags.contains(Flag.CH); } } 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 745e14138..f1846bed7 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 @@ -21,7 +21,6 @@ import io.lettuce.core.ScoredValue; import io.lettuce.core.Value; import io.lettuce.core.ZAddArgs; import io.lettuce.core.ZStoreArgs; -import org.springframework.data.redis.core.TimeoutUtils; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -30,7 +29,6 @@ import java.util.List; import java.util.concurrent.TimeUnit; import org.reactivestreams.Publisher; - import org.springframework.data.domain.Sort.Direction; import org.springframework.data.redis.connection.DefaultTuple; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; @@ -41,6 +39,7 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Numeric 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.core.TimeoutUtils; import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -84,36 +83,32 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { ZAddArgs args = null; - if (command.isIncr() || command.isUpsert() || command.isReturnTotalChanged()) { + if (command.isIncr()) { - if (command.isIncr()) { - - if (command.getTuples().size() > 1) { - throw new IllegalArgumentException("ZADD INCR must not contain more than one tuple!"); - } - - Tuple tuple = command.getTuples().iterator().next(); - - return cmd.zaddincr(command.getKey(), tuple.getScore(), ByteBuffer.wrap(tuple.getValue())) - .map(value -> new NumericResponse<>(command, value)); + if (command.getTuples().size() > 1) { + throw new IllegalArgumentException("ZADD INCR must not contain more than one tuple!"); } - if (command.isReturnTotalChanged()) { - args = ZAddArgs.Builder.ch(); - } + Tuple tuple = command.getTuples().iterator().next(); - if (command.isUpsert()) { - args = args == null ? ZAddArgs.Builder.nx() : args.nx(); - } else { - args = args == null ? ZAddArgs.Builder.xx() : args.xx(); - } + return cmd.zaddincr(command.getKey(), tuple.getScore(), ByteBuffer.wrap(tuple.getValue())) + .map(value -> new NumericResponse<>(command, value)); + } - if (command.isGt()) { - args = args == null ? ZAddArgs.Builder.gt() : args.gt(); - } - if (command.isLt()) { - args = args == null ? ZAddArgs.Builder.lt() : args.lt(); - } + if (command.isReturnTotalChanged()) { + args = ZAddArgs.Builder.ch(); + } + + if (command.isIfNotExists()) { + args = args == null ? ZAddArgs.Builder.nx() : args.nx(); + } else if (command.isIfExists()) { + args = args == null ? ZAddArgs.Builder.xx() : args.xx(); + } + + if (command.isGt()) { + args = args == null ? ZAddArgs.Builder.gt() : args.gt(); + } else if (command.isLt()) { + args = args == null ? ZAddArgs.Builder.lt() : args.lt(); } ScoredValue[] values = (ScoredValue[]) command.getTuples().stream() @@ -161,7 +156,7 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { })); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zRandMember(Publisher) */ @@ -177,7 +172,7 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { })); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zRandMemberWithScore(Publisher) */ @@ -189,8 +184,8 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { Assert.notNull(command.getKey(), "Key must not be null!"); - return new CommandResponse<>(command, cmd.zrandmemberWithScores(command.getKey(), command.getCount()) - .map(this::toTuple)); + return new CommandResponse<>(command, + cmd.zrandmemberWithScores(command.getKey(), command.getCount()).map(this::toTuple)); })); } @@ -414,7 +409,7 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands { Assert.notNull(command.getKey(), "Key must not be null!"); Assert.notNull(command.getTimeout(), "Timeout must not be null!"); - if(command.getTimeUnit() == TimeUnit.MILLISECONDS) { + if (command.getTimeUnit() == TimeUnit.MILLISECONDS) { double timeout = TimeoutUtils.toDoubleSeconds(command.getTimeout(), command.getTimeUnit()); 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 bf50c13b1..c0167b465 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 @@ -19,14 +19,20 @@ import static org.assertj.core.api.Assertions.*; import static org.assertj.core.api.Assumptions.*; import static org.springframework.data.domain.Range.Bound.*; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import java.nio.ByteBuffer; import java.time.Duration; import java.util.Arrays; +import java.util.function.Function; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.DefaultTuple; +import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; +import org.springframework.data.redis.connection.ReactiveZSetCommands.ZAddCommand; +import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; import org.springframework.data.redis.core.ScanOptions; import org.springframework.data.redis.test.condition.EnabledOnCommand; import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest; @@ -55,6 +61,149 @@ public class LettuceReactiveZSetCommandsIntegrationTests extends LettuceReactive assertThat(connection.zSetCommands().zAdd(KEY_1_BBUFFER, 3.5D, VALUE_1_BBUFFER).block()).isEqualTo(1L); } + @ParameterizedRedisTest // GH-2731 + void zAddShouldConsiderAbsentPresentUpsertFlags() { + + Tuple tuple = new DefaultTuple(VALUE_1_BYTES, 3.5D); + + zAdd(KEY_1_BBUFFER, tuple, Function.identity()).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + // NX + zAdd(KEY_1_BBUFFER, tuple, ZAddCommand::nx).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + + zAdd(KEY_2_BBUFFER, tuple, ZAddCommand::nx).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + // XX + zAdd(KEY_1_BBUFFER, new DefaultTuple(VALUE_1_BYTES, 3.0D), ZAddCommand::xx).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + + connection.zSetCommands().zScore(KEY_1_BBUFFER, VALUE_1_BBUFFER).map(Number::doubleValue) // + .as(StepVerifier::create) // + .expectNext(3.0) // + .verifyComplete(); + + zAdd(KEY_3_BBUFFER, tuple, ZAddCommand::xx).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + } + + @ParameterizedRedisTest // GH-2731 + void zAddShouldConsiderLessThan() { + + Tuple tuple = new DefaultTuple(VALUE_1_BYTES, 3.5D); + + zAdd(KEY_1_BBUFFER, tuple, Function.identity()).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + zAdd(KEY_1_BBUFFER, new DefaultTuple(VALUE_1_BYTES, 6D), ZAddCommand::lt).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + + connection.zSetCommands().zScore(KEY_1_BBUFFER, VALUE_1_BBUFFER).map(Number::doubleValue) // + .as(StepVerifier::create) // + .expectNext(3.5) // + .verifyComplete(); + + zAdd(KEY_1_BBUFFER, new DefaultTuple(VALUE_1_BYTES, 1D), ZAddCommand::lt).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + + connection.zSetCommands().zScore(KEY_1_BBUFFER, VALUE_1_BBUFFER).map(Number::doubleValue) // + .as(StepVerifier::create) // + .expectNext(1.0) // + .verifyComplete(); + } + + @ParameterizedRedisTest // GH-2731 + void zAddShouldConsiderGreaterThan() { + + Tuple tuple = new DefaultTuple(VALUE_1_BYTES, 3.5D); + + zAdd(KEY_1_BBUFFER, tuple, Function.identity()).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + zAdd(KEY_1_BBUFFER, new DefaultTuple(VALUE_1_BYTES, 1D), ZAddCommand::gt).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + + connection.zSetCommands().zScore(KEY_1_BBUFFER, VALUE_1_BBUFFER).map(Number::doubleValue) // + .as(StepVerifier::create) // + .expectNext(3.5) // + .verifyComplete(); + + zAdd(KEY_1_BBUFFER, new DefaultTuple(VALUE_1_BYTES, 6D), ZAddCommand::gt).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + + connection.zSetCommands().zScore(KEY_1_BBUFFER, VALUE_1_BBUFFER).map(Number::doubleValue) // + .as(StepVerifier::create) // + .expectNext(6.0) // + .verifyComplete(); + } + + @ParameterizedRedisTest // GH-2731 + void zAddShouldConsiderIncrFlag() { + + Tuple tuple = new DefaultTuple(VALUE_1_BYTES, 3.5D); + + zAdd(KEY_1_BBUFFER, tuple, Function.identity()).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + zAdd(KEY_1_BBUFFER, tuple, ZAddCommand::incr).map(Number::intValue) // + + .as(StepVerifier::create) // + .expectNext(7) // + .verifyComplete(); + } + + @ParameterizedRedisTest // GH-2731 + void zAddShouldConsiderChFlag() { + + Tuple tuple = new DefaultTuple(VALUE_1_BYTES, 3.5D); + + zAdd(KEY_1_BBUFFER, tuple, Function.identity()).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + + zAdd(KEY_1_BBUFFER, tuple, ZAddCommand::ch).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(0) // + .verifyComplete(); + + zAdd(KEY_1_BBUFFER, new DefaultTuple(VALUE_1_BYTES, 3.0D), ZAddCommand::ch).map(Number::intValue) // + .as(StepVerifier::create) // + .expectNext(1) // + .verifyComplete(); + } + + private Flux zAdd(ByteBuffer key, Tuple tuple, Function commandCustomizer) { + return connection.zSetCommands().zAdd(Mono.just(commandCustomizer.apply(ZAddCommand.tuple(tuple).to(key)))) + .map(NumericResponse::getOutput); + } + @ParameterizedRedisTest // DATAREDIS-525 void zRemShouldRemoveValuesFromSet() {