Add reactive support for ZDIFF, ZDIFFSTORE, ZINTER, and ZUNION commands.

See: #2041 & #2042
Original Pull Request: #2097

Author: mp911de

src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java

@@ -1174,34 +1174,6 @@ default Long zRemRangeByScore(byte[] key, double min, double max) {
 	@Nullable
 	Set<byte[]> zInter(byte[]... sets);
 
-	/**
-	 * Intersect sorted {@code sets}.
-	 *
-	 * @param aggregate must not be {@literal null}.
-	 * @param weights must not be {@literal null}.
-	 * @param sets must not be {@literal null}.
-	 * @return {@literal null} when used in pipeline / transaction.
-	 * @since 2.6
-	 * @see <a href="https://redis.io/commands/zinter">Redis Documentation: ZINTER</a>
-	 */
-	@Nullable
-	default Set<byte[]> zInter(Aggregate aggregate, int[] weights, byte[]... sets) {
-		return zInter(aggregate, Weights.of(weights), sets);
-	}
-
-	/**
-	 * Intersect sorted {@code sets}.
-	 *
-	 * @param aggregate must not be {@literal null}.
-	 * @param weights must not be {@literal null}.
-	 * @param sets must not be {@literal null}.
-	 * @return {@literal null} when used in pipeline / transaction.
-	 * @since 2.6
-	 * @see <a href="https://redis.io/commands/zinter">Redis Documentation: ZINTER</a>
-	 */
-	@Nullable
-	Set<byte[]> zInter(Aggregate aggregate, Weights weights, byte[]... sets);
-
 	/**
 	 * Intersect sorted {@code sets}.
 	 *
@@ -1293,34 +1265,6 @@ default Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byt
 	@Nullable
 	Set<byte[]> zUnion(byte[]... sets);
 
-	/**
-	 * Union sorted {@code sets}.
-	 *
-	 * @param aggregate must not be {@literal null}.
-	 * @param weights must not be {@literal null}.
-	 * @param sets must not be {@literal null}.
-	 * @return {@literal null} when used in pipeline / transaction.
-	 * @since 2.6
-	 * @see <a href="https://redis.io/commands/zunion">Redis Documentation: ZUNION</a>
-	 */
-	@Nullable
-	default Set<byte[]> zUnion(Aggregate aggregate, int[] weights, byte[]... sets) {
-		return zUnion(aggregate, Weights.of(weights), sets);
-	}
-
-	/**
-	 * Union sorted {@code sets}.
-	 *
-	 * @param aggregate must not be {@literal null}.
-	 * @param weights must not be {@literal null}.
-	 * @param sets must not be {@literal null}.
-	 * @return {@literal null} when used in pipeline / transaction.
-	 * @since 2.6
-	 * @see <a href="https://redis.io/commands/zunion">Redis Documentation: ZUNION</a>
-	 */
-	@Nullable
-	Set<byte[]> zUnion(Aggregate aggregate, Weights weights, byte[]... sets);
-
 	/**
 	 * Union sorted {@code sets}.
 	 *

src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java

@@ -41,11 +41,13 @@ class LettuceReactiveClusterZSetCommands extends LettuceReactiveZSetCommands imp
 		super(connection);
 	}
 
-	/* (non-Javadoc)
+	/*
+	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveZSetCommands#zUnionStore(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<NumericResponse<ZUnionStoreCommand, Long>> zUnionStore(Publisher<ZUnionStoreCommand> commands) {
+	public Flux<NumericResponse<ZAggregateStoreCommand, Long>> zUnionStore(
+			Publisher<? extends ZAggregateStoreCommand> commands) {
 
 		return getConnection().execute(cmd -> Flux.from(commands).concatMap(command -> {
 
@@ -60,11 +62,13 @@ public Flux<NumericResponse<ZUnionStoreCommand, Long>> zUnionStore(Publisher<ZUn
 		}));
 	}
 
-	/* (non-Javadoc)
+	/*
+	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveZSetCommands#zInterStore(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<NumericResponse<ZInterStoreCommand, Long>> zInterStore(Publisher<ZInterStoreCommand> commands) {
+	public Flux<NumericResponse<ZAggregateStoreCommand, Long>> zInterStore(
+			Publisher<? extends ZAggregateStoreCommand> commands) {
 		return getConnection().execute(cmd -> Flux.from(commands).concatMap(command -> {
 
 			Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty.");

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterZSetCommands.java

@@ -200,22 +200,18 @@ public Flux<CommandResponse<ZRangeCommand, Flux<Tuple>>> zRange(Publisher<ZRange
 			if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
 				if (command.isWithScores()) {
 
-					result = cmd.zrangeWithScores(command.getKey(), start, stop)
-							.map(this::toTuple);
+					result = cmd.zrangeWithScores(command.getKey(), start, stop).map(this::toTuple);
 				} else {
 
-					result = cmd.zrange(command.getKey(), start, stop)
-							.map(value -> toTuple(value, Double.NaN));
+					result = cmd.zrange(command.getKey(), start, stop).map(value -> toTuple(value, Double.NaN));
 				}
 			} else {
 				if (command.isWithScores()) {
 
-					result = cmd.zrevrangeWithScores(command.getKey(), start, stop)
-							.map(this::toTuple);
+					result = cmd.zrevrangeWithScores(command.getKey(), start, stop).map(this::toTuple);
 				} else {
 
-					result = cmd.zrevrange(command.getKey(), start, stop)
-							.map(value -> toTuple(value, Double.NaN));
+					result = cmd.zrevrange(command.getKey(), start, stop).map(value -> toTuple(value, Double.NaN));
 				}
 			}
 
@@ -247,8 +243,7 @@ public Flux<CommandResponse<ZRangeByScoreCommand, Flux<Tuple>>> zRangeByScore(
 				if (command.isWithScores()) {
 
 					if (!isLimited) {
-						result = cmd.zrangebyscoreWithScores(command.getKey(), range)
-								.map(this::toTuple);
+						result = cmd.zrangebyscoreWithScores(command.getKey(), range).map(this::toTuple);
 					} else {
 						result = cmd
 								.zrangebyscoreWithScores(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
@@ -257,8 +252,7 @@ public Flux<CommandResponse<ZRangeByScoreCommand, Flux<Tuple>>> zRangeByScore(
 				} else {
 
 					if (!isLimited) {
-						result = cmd.zrangebyscore(command.getKey(), range)
-								.map(value -> toTuple(value, Double.NaN));
+						result = cmd.zrangebyscore(command.getKey(), range).map(value -> toTuple(value, Double.NaN));
 					} else {
 
 						result = cmd.zrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
@@ -272,20 +266,16 @@ public Flux<CommandResponse<ZRangeByScoreCommand, Flux<Tuple>>> zRangeByScore(
 				if (command.isWithScores()) {
 
 					if (!isLimited) {
-						result = cmd.zrevrangebyscoreWithScores(command.getKey(), range)
-								.map(this::toTuple);
+						result = cmd.zrevrangebyscoreWithScores(command.getKey(), range).map(this::toTuple);
 					} else {
 
-						result = cmd
-								.zrevrangebyscoreWithScores(command.getKey(), range,
-										LettuceConverters.toLimit(command.getLimit().get()))
-								.map(this::toTuple);
+						result = cmd.zrevrangebyscoreWithScores(command.getKey(), range,
+								LettuceConverters.toLimit(command.getLimit().get())).map(this::toTuple);
 					}
 				} else {
 
 					if (!isLimited) {
-						result = cmd.zrevrangebyscore(command.getKey(), range)
-								.map(value -> toTuple(value, Double.NaN));
+						result = cmd.zrevrangebyscore(command.getKey(), range).map(value -> toTuple(value, Double.NaN));
 					} else {
 
 						result = cmd.zrevrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
@@ -520,26 +510,99 @@ public Flux<NumericResponse<ZRemRangeByLexCommand, Long>> zRemRangeByLex(Publish
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zUnionStore(org.reactivestreams.Publisher)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zDiff(Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<ZDiffCommand, Flux<ByteBuffer>>> zDiff(Publisher<? extends ZDiffCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notEmpty(command.getKeys(), "Keys must not be null or empty!");
+
+			ByteBuffer[] sourceKeys = command.getKeys().toArray(new ByteBuffer[0]);
+			return new CommandResponse<>(command, cmd.zdiff(sourceKeys));
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zDiffWithScores(Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<ZDiffCommand, Flux<Tuple>>> zDiffWithScores(Publisher<? extends ZDiffCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notEmpty(command.getKeys(), "Keys must not be null or empty!");
+
+			ByteBuffer[] sourceKeys = command.getKeys().toArray(new ByteBuffer[0]);
+			return new CommandResponse<>(command, cmd.zdiffWithScores(sourceKeys).map(this::toTuple));
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zDiffStore(Publisher)
 	 */
 	@Override
-	public Flux<NumericResponse<ZUnionStoreCommand, Long>> zUnionStore(Publisher<ZUnionStoreCommand> commands) {
+	public Flux<NumericResponse<ZDiffStoreCommand, Long>> zDiffStore(Publisher<ZDiffStoreCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
 
 			Assert.notNull(command.getKey(), "Destination key must not be null!");
 			Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!");
 
+			ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]);
+			return cmd.zdiffstore(command.getKey(), sourceKeys).map(value -> new NumericResponse<>(command, value));
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zInter(Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<ZAggregateCommand, Flux<ByteBuffer>>> zInter(
+			Publisher<? extends ZAggregateCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!");
+
 			ZStoreArgs args = null;
 			if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) {
 				args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null,
 						command.getWeights());
 			}
 
-			ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new);
-			Mono<Long> result = args != null ? cmd.zunionstore(command.getKey(), args, sourceKeys)
-					: cmd.zunionstore(command.getKey(), sourceKeys);
-			return result.map(value -> new NumericResponse<>(command, value));
+			ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]);
+			Flux<ByteBuffer> result = args != null ? cmd.zinter(args, sourceKeys) : cmd.zinter(sourceKeys);
+			return new CommandResponse<>(command, result);
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zInterWithScores(Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<ZAggregateCommand, Flux<Tuple>>> zInterWithScores(
+			Publisher<? extends ZAggregateCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!");
+
+			ZStoreArgs args = null;
+			if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) {
+				args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null,
+						command.getWeights());
+			}
+
+			ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]);
+			Flux<ScoredValue<ByteBuffer>> result = args != null ? cmd.zinterWithScores(args, sourceKeys)
+					: cmd.zinterWithScores(sourceKeys);
+			return new CommandResponse<>(command, result.map(this::toTuple));
 		}));
 	}
 
@@ -548,7 +611,8 @@ public Flux<NumericResponse<ZUnionStoreCommand, Long>> zUnionStore(Publisher<ZUn
 	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zInterStore(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<NumericResponse<ZInterStoreCommand, Long>> zInterStore(Publisher<ZInterStoreCommand> commands) {
+	public Flux<NumericResponse<ZAggregateStoreCommand, Long>> zInterStore(
+			Publisher<? extends ZAggregateStoreCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
 
@@ -561,13 +625,88 @@ public Flux<NumericResponse<ZInterStoreCommand, Long>> zInterStore(Publisher<ZIn
 						command.getWeights());
 			}
 
-			ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new);
+			ByteBuffer[] sourceKeys = command.getSourceKeys().toArray(new ByteBuffer[0]);
 			Mono<Long> result = args != null ? cmd.zinterstore(command.getKey(), args, sourceKeys)
 					: cmd.zinterstore(command.getKey(), sourceKeys);
 			return result.map(value -> new NumericResponse<>(command, value));
 		}));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zUnion(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<ZAggregateCommand, Flux<ByteBuffer>>> zUnion(
+			Publisher<? extends ZAggregateCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!");
+
+			ZStoreArgs args = null;
+			if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) {
+				args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null,
+						command.getWeights());
+			}
+
+			ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new);
+			Flux<ByteBuffer> result = args != null ? cmd.zunion(args, sourceKeys) : cmd.zunion(sourceKeys);
+			return new CommandResponse<>(command, result);
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zUnion(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<ZAggregateCommand, Flux<Tuple>>> zUnionWithScores(
+			Publisher<? extends ZAggregateCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!");
+
+			ZStoreArgs args = null;
+			if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) {
+				args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null,
+						command.getWeights());
+			}
+
+			ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new);
+			Flux<ScoredValue<ByteBuffer>> result = args != null ? cmd.zunionWithScores(args, sourceKeys)
+					: cmd.zunionWithScores(sourceKeys);
+			return new CommandResponse<>(command, result.map(this::toTuple));
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zUnionStore(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<NumericResponse<ZAggregateStoreCommand, Long>> zUnionStore(
+			Publisher<? extends ZAggregateStoreCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
+
+			Assert.notNull(command.getKey(), "Destination key must not be null!");
+			Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null or empty!");
+
+			ZStoreArgs args = null;
+			if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) {
+				args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null,
+						command.getWeights());
+			}
+
+			ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new);
+			Mono<Long> result = args != null ? cmd.zunionstore(command.getKey(), args, sourceKeys)
+					: cmd.zunionstore(command.getKey(), sourceKeys);
+			return result.map(value -> new NumericResponse<>(command, value));
+		}));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zRangeByLex(org.reactivestreams.Publisher)