Correctly apply Reactive zadd NX/XX command flags.

We now correctly apply if exists/if not exists constraints on the reactive zadd command.

Previously, we only considered upsert which wasn't sufficient to apply xx/nx.

Closes #2731

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<Tuple> tuples;
-		private final boolean upsert;
-		private final boolean returnTotalChanged;
+		private final Set<Flag> flags;
 		private final boolean incr;
-		private final boolean gt;
-		private final boolean lt;
 
-		private ZAddCommand(@Nullable ByteBuffer key, List<Tuple> tuples, boolean upsert, boolean returnTotalChanged,
-				boolean incr, boolean gt, boolean lt) {
+		private ZAddCommand(@Nullable ByteBuffer key, List<Tuple> tuples, Set<Flag> 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<Flag> 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<Flag> 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<Flag> 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<Flag> 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<Flag> 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);
 		}
 	}
 

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<ByteBuffer>[] values = (ScoredValue<ByteBuffer>[]) 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());
 

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<Number> zAdd(ByteBuffer key, Tuple tuple, Function<ZAddCommand, ZAddCommand> commandCustomizer) {
+		return connection.zSetCommands().zAdd(Mono.just(commandCustomizer.apply(ZAddCommand.tuple(tuple).to(key))))
+				.map(NumericResponse::getOutput);
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-525
 	void zRemShouldRemoveValuesFromSet() {