DATAREDIS-613 - Stream results instead of returning collections where possible.

Allow streaming where possible without breaking Redis API contract. This means we keep the list collection for those operations potentially returning NULL to indicate a value is not present but was requested (eg MGET).

Additionally remove the Spring 4 related travis builds.

Original pull request: #244.

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

@@ -25,14 +25,12 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
-import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.geo.Point;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
@@ -859,9 +857,8 @@ public interface ReactiveGeoCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/georadius">Redis Documentation: GEORADIUS</a>
 	 */
-	default Mono<List<GeoLocation<ByteBuffer>>> geoRadius(ByteBuffer key, Circle circle) {
-		return geoRadius(key, circle, null)
-				.map(res -> res.getContent().stream().map(GeoResult::getContent).collect(Collectors.toList()));
+	default Flux<GeoResult<GeoLocation<ByteBuffer>>> geoRadius(ByteBuffer key, Circle circle) {
+		return geoRadius(key, circle, null);
 	}
 
 	/**
@@ -873,14 +870,14 @@ public interface ReactiveGeoCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/georadius">Redis Documentation: GEORADIUS</a>
 	 */
-	default Mono<GeoResults<GeoLocation<ByteBuffer>>> geoRadius(ByteBuffer key, Circle circle,
+	default Flux<GeoResult<GeoLocation<ByteBuffer>>> geoRadius(ByteBuffer key, Circle circle,
 			GeoRadiusCommandArgs geoRadiusArgs) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(circle, "Circle must not be null!");
 
 		return geoRadius(Mono.just(GeoRadiusCommand.within(circle).withArgs(geoRadiusArgs).forKey(key))).next()
-				.map(CommandResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -890,7 +887,7 @@ public interface ReactiveGeoCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/georadius">Redis Documentation: GEORADIUS</a>
 	 */
-	Flux<CommandResponse<GeoRadiusCommand, GeoResults<GeoLocation<ByteBuffer>>>> geoRadius(
+	Flux<CommandResponse<GeoRadiusCommand, Flux<GeoResult<GeoLocation<ByteBuffer>>>>> geoRadius(
 			Publisher<GeoRadiusCommand> commands);
 
 	/**
@@ -1184,9 +1181,9 @@ public interface ReactiveGeoCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/georadiusbymember">Redis Documentation: GEORADIUSBYMEMBER</a>
 	 */
-	default Mono<List<GeoLocation<ByteBuffer>>> geoRadiusByMember(ByteBuffer key, ByteBuffer member, Distance distance) {
-		return geoRadiusByMember(key, member, distance, null)
-				.map(res -> res.getContent().stream().map(GeoResult::getContent).collect(Collectors.toList()));
+	default Flux<GeoResult<GeoLocation<ByteBuffer>>> geoRadiusByMember(ByteBuffer key, ByteBuffer member,
+			Distance distance) {
+		return geoRadiusByMember(key, member, distance, null);
 	}
 
 	/**
@@ -1198,7 +1195,7 @@ public interface ReactiveGeoCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/georadiusbymember">Redis Documentation: GEORADIUSBYMEMBER</a>
 	 */
-	default Mono<GeoResults<GeoLocation<ByteBuffer>>> geoRadiusByMember(ByteBuffer key, ByteBuffer member,
+	default Flux<GeoResult<GeoLocation<ByteBuffer>>> geoRadiusByMember(ByteBuffer key, ByteBuffer member,
 			Distance distance, GeoRadiusCommandArgs geoRadiusArgs) {
 
 		Assert.notNull(key, "Key must not be null!");
@@ -1207,7 +1204,7 @@ public interface ReactiveGeoCommands {
 
 		return geoRadiusByMember(
 				Mono.just(GeoRadiusByMemberCommand.within(distance).from(member).forKey(key).withArgs(geoRadiusArgs))).next()
-						.map(CommandResponse::getOutput);
+						.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -1217,6 +1214,6 @@ public interface ReactiveGeoCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/georadiusbymember">Redis Documentation: GEORADIUSBYMEMBER</a>
 	 */
-	Flux<CommandResponse<GeoRadiusByMemberCommand, GeoResults<GeoLocation<ByteBuffer>>>> geoRadiusByMember(
+	Flux<CommandResponse<GeoRadiusByMemberCommand, Flux<GeoResult<GeoLocation<ByteBuffer>>>>> geoRadiusByMember(
 			Publisher<GeoRadiusByMemberCommand> commands);
 }

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

@@ -516,11 +516,11 @@ public interface ReactiveHashCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/hkeys">Redis Documentation: HKEYS</a>
 	 */
-	default Mono<List<ByteBuffer>> hKeys(ByteBuffer key) {
+	default Flux<ByteBuffer> hKeys(ByteBuffer key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return hKeys(Mono.just(new KeyCommand(key))).next().map(MultiValueResponse::getOutput);
+		return hKeys(Mono.just(new KeyCommand(key))).next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -530,7 +530,7 @@ public interface ReactiveHashCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/hkeys">Redis Documentation: HKEYS</a>
 	 */
-	Flux<MultiValueResponse<KeyCommand, ByteBuffer>> hKeys(Publisher<KeyCommand> commands);
+	Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> hKeys(Publisher<KeyCommand> commands);
 
 	/**
 	 * Get entry set (values) of hash at {@literal key}.
@@ -539,11 +539,11 @@ public interface ReactiveHashCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/hvals">Redis Documentation: HVALS</a>
 	 */
-	default Mono<List<ByteBuffer>> hVals(ByteBuffer key) {
+	default Flux<ByteBuffer> hVals(ByteBuffer key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return hVals(Mono.just(new KeyCommand(key))).next().map(MultiValueResponse::getOutput);
+		return hVals(Mono.just(new KeyCommand(key))).next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -553,7 +553,7 @@ public interface ReactiveHashCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/hvals">Redis Documentation: HVALS</a>
 	 */
-	Flux<MultiValueResponse<KeyCommand, ByteBuffer>> hVals(Publisher<KeyCommand> commands);
+	Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> hVals(Publisher<KeyCommand> commands);
 
 	/**
 	 * Get entire hash stored at {@literal key}.
@@ -562,11 +562,11 @@ public interface ReactiveHashCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/hgetall">Redis Documentation: HGETALL</a>
 	 */
-	default Mono<Map<ByteBuffer, ByteBuffer>> hGetAll(ByteBuffer key) {
+	default Flux<Map.Entry<ByteBuffer, ByteBuffer>> hGetAll(ByteBuffer key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return hGetAll(Mono.just(new KeyCommand(key))).next().map(CommandResponse::getOutput);
+		return hGetAll(Mono.just(new KeyCommand(key))).next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -576,5 +576,5 @@ public interface ReactiveHashCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/hgetall">Redis Documentation: HGETALL</a>
 	 */
-	Flux<CommandResponse<KeyCommand, Map<ByteBuffer, ByteBuffer>>> hGetAll(Publisher<KeyCommand> commands);
+	Flux<CommandResponse<KeyCommand, Flux<Map.Entry<ByteBuffer, ByteBuffer>>>> hGetAll(Publisher<KeyCommand> commands);
 }

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

@@ -30,7 +30,6 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBuf
 import org.springframework.data.redis.connection.ReactiveRedisConnection.Command;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
-import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.RangeCommand;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
@@ -269,12 +268,12 @@ public interface ReactiveListCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/lrange">Redis Documentation: LRANGE</a>
 	 */
-	default Mono<List<ByteBuffer>> lRange(ByteBuffer key, long start, long end) {
+	default Flux<ByteBuffer> lRange(ByteBuffer key, long start, long end) {
 
 		Assert.notNull(key, "Key must not be null!");
 
 		return lRange(Mono.just(RangeCommand.key(key).fromIndex(start).toIndex(end))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -284,7 +283,7 @@ public interface ReactiveListCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/lrange">Redis Documentation: LRANGE</a>
 	 */
-	Flux<MultiValueResponse<RangeCommand, ByteBuffer>> lRange(Publisher<RangeCommand> commands);
+	Flux<CommandResponse<RangeCommand, Flux<ByteBuffer>>> lRange(Publisher<RangeCommand> commands);
 
 	/**
 	 * Trim list at {@literal key} to elements between {@literal begin} and {@literal end}.

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

@@ -29,8 +29,8 @@ import org.reactivestreams.Publisher;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBufferResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.Command;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
-import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
 import org.springframework.util.Assert;
 
@@ -524,11 +524,11 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sinter">Redis Documentation: SINTER</a>
 	 */
-	default Mono<List<ByteBuffer>> sInter(Collection<ByteBuffer> keys) {
+	default Flux<ByteBuffer> sInter(Collection<ByteBuffer> keys) {
 
 		Assert.notNull(keys, "Keys must not be null!");
 
-		return sInter(Mono.just(SInterCommand.keys(keys))).next().map(MultiValueResponse::getOutput);
+		return sInter(Mono.just(SInterCommand.keys(keys))).next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -538,7 +538,7 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sinter">Redis Documentation: SINTER</a>
 	 */
-	Flux<MultiValueResponse<SInterCommand, ByteBuffer>> sInter(Publisher<SInterCommand> commands);
+	Flux<CommandResponse<SInterCommand, Flux<ByteBuffer>>> sInter(Publisher<SInterCommand> commands);
 
 	/**
 	 * {@code SINTERSTORE} command parameters.
@@ -668,11 +668,11 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sunion">Redis Documentation: SUNION</a>
 	 */
-	default Mono<List<ByteBuffer>> sUnion(Collection<ByteBuffer> keys) {
+	default Flux<ByteBuffer> sUnion(Collection<ByteBuffer> keys) {
 
 		Assert.notNull(keys, "Keys must not be null!");
 
-		return sUnion(Mono.just(SUnionCommand.keys(keys))).next().map(MultiValueResponse::getOutput);
+		return sUnion(Mono.just(SUnionCommand.keys(keys))).next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -682,7 +682,7 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sunion">Redis Documentation: SUNION</a>
 	 */
-	Flux<MultiValueResponse<SUnionCommand, ByteBuffer>> sUnion(Publisher<SUnionCommand> commands);
+	Flux<CommandResponse<SUnionCommand, Flux<ByteBuffer>>> sUnion(Publisher<SUnionCommand> commands);
 
 	/**
 	 * {@code SUNIONSTORE} command parameters.
@@ -812,11 +812,11 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sdiff">Redis Documentation: SDIFF</a>
 	 */
-	default Mono<List<ByteBuffer>> sDiff(Collection<ByteBuffer> keys) {
+	default Flux<ByteBuffer> sDiff(Collection<ByteBuffer> keys) {
 
 		Assert.notNull(keys, "Keys must not be null!");
 
-		return sDiff(Mono.just(SDiffCommand.keys(keys))).next().map(MultiValueResponse::getOutput);
+		return sDiff(Mono.just(SDiffCommand.keys(keys))).next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -826,7 +826,7 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sdiff">Redis Documentation: SDIFF</a>
 	 */
-	Flux<MultiValueResponse<SDiffCommand, ByteBuffer>> sDiff(Publisher<SDiffCommand> commands);
+	Flux<CommandResponse<SDiffCommand, Flux<ByteBuffer>>> sDiff(Publisher<SDiffCommand> commands);
 
 	/**
 	 * {@code SDIFFSTORE} command parameters.
@@ -913,11 +913,11 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/smembers">Redis Documentation: SMEMBERS</a>
 	 */
-	default Mono<List<ByteBuffer>> sMembers(ByteBuffer key) {
+	default Flux<ByteBuffer> sMembers(ByteBuffer key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return sMembers(Mono.just(new KeyCommand(key))).next().map(MultiValueResponse::getOutput);
+		return sMembers(Mono.just(new KeyCommand(key))).next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -927,7 +927,7 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/smembers">Redis Documentation: SMEMBERS</a>
 	 */
-	Flux<MultiValueResponse<KeyCommand, ByteBuffer>> sMembers(Publisher<KeyCommand> commands);
+	Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> sMembers(Publisher<KeyCommand> commands);
 
 	/**
 	 * {@code SRANDMEMBER} command parameters.
@@ -993,7 +993,7 @@ public interface ReactiveSetCommands {
 	 * @see <a href="http://redis.io/commands/srandmember">Redis Documentation: SRANDMEMBER</a>
 	 */
 	default Mono<ByteBuffer> sRandMember(ByteBuffer key) {
-		return sRandMember(key, 1L).map(vals -> vals.isEmpty() ? null : vals.iterator().next());
+		return sRandMember(key, 1L).singleOrEmpty();
 	}
 
 	/**
@@ -1004,13 +1004,13 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/srandmember">Redis Documentation: SRANDMEMBER</a>
 	 */
-	default Mono<List<ByteBuffer>> sRandMember(ByteBuffer key, Long count) {
+	default Flux<ByteBuffer> sRandMember(ByteBuffer key, Long count) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(count, "Count must not be null!");
 
 		return sRandMember(Mono.just(SRandMembersCommand.valueCount(count).from(key))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -1020,5 +1020,5 @@ public interface ReactiveSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/srandmember">Redis Documentation: SRANDMEMBER</a>
 	 */
-	Flux<MultiValueResponse<SRandMembersCommand, ByteBuffer>> sRandMember(Publisher<SRandMembersCommand> commands);
+	Flux<CommandResponse<SRandMembersCommand, Flux<ByteBuffer>>> sRandMember(Publisher<SRandMembersCommand> commands);
 }

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

@@ -24,13 +24,12 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.domain.Range;
 import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
-import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
@@ -652,17 +651,14 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrange">Redis Documentation: ZRANGE</a>
 	 */
-	default Mono<List<ByteBuffer>> zRange(ByteBuffer key, Range<Long> range) {
+	default Flux<ByteBuffer> zRange(ByteBuffer key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRange(Mono.just(ZRangeCommand.valuesWithin(range).from(key))) //
 				.next() //
-				.map(resp -> resp.getOutput() //
-						.stream() //
-						.map(tuple -> ByteBuffer.wrap(tuple.getValue())) //
-						.collect(Collectors.toList()));
+				.flatMapMany(CommandResponse::getOutput).map(tuple -> ByteBuffer.wrap(tuple.getValue()));
 	}
 
 	/**
@@ -673,12 +669,12 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrange">Redis Documentation: ZRANGE</a>
 	 */
-	default Mono<List<Tuple>> zRangeWithScores(ByteBuffer key, Range<Long> range) {
+	default Flux<Tuple> zRangeWithScores(ByteBuffer key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 
 		return zRange(Mono.just(ZRangeCommand.valuesWithin(range).withScores().from(key))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -689,12 +685,12 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrange">Redis Documentation: ZREVRANGE</a>
 	 */
-	default Mono<List<ByteBuffer>> zRevRange(ByteBuffer key, Range<Long> range) {
+	default Flux<ByteBuffer> zRevRange(ByteBuffer key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return zRange(Mono.just(ZRangeCommand.reverseValuesWithin(range).from(key))).next().map(
-				resp -> resp.getOutput().stream().map(tuple -> ByteBuffer.wrap(tuple.getValue())).collect(Collectors.toList()));
+		return zRange(Mono.just(ZRangeCommand.reverseValuesWithin(range).from(key))).next()
+				.flatMapMany(CommandResponse::getOutput).map(tuple -> ByteBuffer.wrap(tuple.getValue()));
 	}
 
 	/**
@@ -705,12 +701,12 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrange">Redis Documentation: ZREVRANGE</a>
 	 */
-	default Mono<List<Tuple>> zRevRangeWithScores(ByteBuffer key, Range<Long> range) {
+	default Flux<Tuple> zRevRangeWithScores(ByteBuffer key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 
 		return zRange(Mono.just(ZRangeCommand.reverseValuesWithin(range).withScores().from(key))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -721,7 +717,7 @@ public interface ReactiveZSetCommands {
 	 * @see <a href="http://redis.io/commands/zrange">Redis Documentation: ZRANGE</a>
 	 * @see <a href="http://redis.io/commands/zrevrange">Redis Documentation: ZREVRANGE</a>
 	 */
-	Flux<MultiValueResponse<ZRangeCommand, Tuple>> zRange(Publisher<ZRangeCommand> commands);
+	Flux<CommandResponse<ZRangeCommand, Flux<Tuple>>> zRange(Publisher<ZRangeCommand> commands);
 
 	/**
 	 * {@literal ZRANGEBYSCORE}/{@literal ZREVRANGEBYSCORE}.
@@ -846,17 +842,15 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	default Mono<List<ByteBuffer>> zRangeByScore(ByteBuffer key, Range<Double> range) {
+	default Flux<ByteBuffer> zRangeByScore(ByteBuffer key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByScore(Mono.just(ZRangeByScoreCommand.scoresWithin(range).from(key))) //
 				.next() //
-				.map(resp -> resp.getOutput() //
-						.stream() //
-						.map(tuple -> ByteBuffer.wrap(tuple.getValue())) //
-						.collect(Collectors.toList()));
+				.flatMapMany(CommandResponse::getOutput) //
+				.map(tuple -> ByteBuffer.wrap(tuple.getValue()));
 	}
 
 	/**
@@ -868,17 +862,15 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	default Mono<List<ByteBuffer>> zRangeByScore(ByteBuffer key, Range<Double> range, Limit limit) {
+	default Flux<ByteBuffer> zRangeByScore(ByteBuffer key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByScore(Mono.just(ZRangeByScoreCommand.scoresWithin(range).from(key).limitTo(limit))) //
 				.next() //
-				.map(resp -> resp.getOutput() //
-						.stream() //
-						.map(tuple -> ByteBuffer.wrap(tuple.getValue())) //
-						.collect(Collectors.toList()));
+				.flatMapMany(CommandResponse::getOutput) //
+				.map(tuple -> ByteBuffer.wrap(tuple.getValue()));
 	}
 
 	/**
@@ -889,13 +881,13 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	default Mono<List<Tuple>> zRangeByScoreWithScores(ByteBuffer key, Range<Double> range) {
+	default Flux<Tuple> zRangeByScoreWithScores(ByteBuffer key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByScore(Mono.just(ZRangeByScoreCommand.scoresWithin(range).withScores().from(key))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -907,13 +899,13 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	default Mono<List<Tuple>> zRangeByScoreWithScores(ByteBuffer key, Range<Double> range, Limit limit) {
+	default Flux<Tuple> zRangeByScoreWithScores(ByteBuffer key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByScore(Mono.just(ZRangeByScoreCommand.scoresWithin(range).withScores().from(key).limitTo(limit)))
-				.next().map(MultiValueResponse::getOutput);
+				.next().flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -924,16 +916,14 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	default Mono<List<ByteBuffer>> zRevRangeByScore(ByteBuffer key, Range<Double> range) {
+	default Flux<ByteBuffer> zRevRangeByScore(ByteBuffer key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 
 		return zRangeByScore(Mono.just(ZRangeByScoreCommand.reverseScoresWithin(range).from(key))) //
 				.next() //
-				.map(resp -> resp.getOutput() //
-						.stream() //
-						.map(tuple -> ByteBuffer.wrap(tuple.getValue())) //
-						.collect(Collectors.toList()));
+				.flatMapMany(CommandResponse::getOutput) //
+				.map(tuple -> ByteBuffer.wrap(tuple.getValue()));
 	}
 
 	/**
@@ -945,15 +935,15 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	default Mono<List<ByteBuffer>> zRevRangeByScore(ByteBuffer key, Range<Double> range, Limit limit) {
+	default Flux<ByteBuffer> zRevRangeByScore(ByteBuffer key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByScore(Mono.just(ZRangeByScoreCommand.reverseScoresWithin(range).from(key).limitTo(limit))) //
 				.next() //
-				.map(resp -> resp.getOutput() //
-						.stream().map(tuple -> ByteBuffer.wrap(tuple.getValue())).collect(Collectors.toList()));
+				.flatMapMany(CommandResponse::getOutput) //
+				.map(tuple -> ByteBuffer.wrap(tuple.getValue()));
 	}
 
 	/**
@@ -964,13 +954,13 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	default Mono<List<Tuple>> zRevRangeByScoreWithScores(ByteBuffer key, Range<Double> range) {
+	default Flux<Tuple> zRevRangeByScoreWithScores(ByteBuffer key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByScore(Mono.just(ZRangeByScoreCommand.reverseScoresWithin(range).withScores().from(key))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -982,14 +972,14 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	default Mono<List<Tuple>> zRevRangeByScoreWithScores(ByteBuffer key, Range<Double> range, Limit limit) {
+	default Flux<Tuple> zRevRangeByScoreWithScores(ByteBuffer key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByScore(
 				Mono.just(ZRangeByScoreCommand.reverseScoresWithin(range).withScores().from(key).limitTo(limit))).next()
-						.map(MultiValueResponse::getOutput);
+						.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -1000,7 +990,7 @@ public interface ReactiveZSetCommands {
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	Flux<MultiValueResponse<ZRangeByScoreCommand, Tuple>> zRangeByScore(Publisher<ZRangeByScoreCommand> commands);
+	Flux<CommandResponse<ZRangeByScoreCommand, Flux<Tuple>>> zRangeByScore(Publisher<ZRangeByScoreCommand> commands);
 
 	/**
 	 * {@code ZCOUNT} command parameters.
@@ -1727,7 +1717,7 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebylex">Redis Documentation: ZRANGEBYLEX</a>
 	 */
-	default Mono<List<ByteBuffer>> zRangeByLex(ByteBuffer key, Range<String> range) {
+	default Flux<ByteBuffer> zRangeByLex(ByteBuffer key, Range<String> range) {
 		return zRangeByLex(key, range, null);
 	}
 
@@ -1741,13 +1731,13 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebylex">Redis Documentation: ZRANGEBYLEX</a>
 	 */
-	default Mono<List<ByteBuffer>> zRangeByLex(ByteBuffer key, Range<String> range, Limit limit) {
+	default Flux<ByteBuffer> zRangeByLex(ByteBuffer key, Range<String> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByLex(Mono.just(ZRangeByLexCommand.stringsWithin(range).from(key).limitTo(limit))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -1758,7 +1748,7 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebylex">Redis Documentation: ZREVRANGEBYLEX</a>
 	 */
-	default Mono<List<ByteBuffer>> zRevRangeByLex(ByteBuffer key, Range<String> range) {
+	default Flux<ByteBuffer> zRevRangeByLex(ByteBuffer key, Range<String> range) {
 		return zRevRangeByLex(key, range, null);
 	}
 
@@ -1772,13 +1762,13 @@ public interface ReactiveZSetCommands {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebylex">Redis Documentation: ZREVRANGEBYLEX</a>
 	 */
-	default Mono<List<ByteBuffer>> zRevRangeByLex(ByteBuffer key, Range<String> range, Limit limit) {
+	default Flux<ByteBuffer> zRevRangeByLex(ByteBuffer key, Range<String> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
 		return zRangeByLex(Mono.just(ZRangeByLexCommand.reverseStringsWithin(range).from(key).limitTo(limit))).next()
-				.map(MultiValueResponse::getOutput);
+				.flatMapMany(CommandResponse::getOutput);
 	}
 
 	/**
@@ -1790,5 +1780,5 @@ public interface ReactiveZSetCommands {
 	 * @see <a href="http://redis.io/commands/zrangebylex">Redis Documentation: ZRANGEBYLEX</a>
 	 * @see <a href="http://redis.io/commands/zrevrangebylex">Redis Documentation: ZREVRANGEBYLEX</a>
 	 */
-	Flux<MultiValueResponse<ZRangeByLexCommand, ByteBuffer>> zRangeByLex(Publisher<ZRangeByLexCommand> commands);
+	Flux<CommandResponse<ZRangeByLexCommand, Flux<ByteBuffer>>> zRangeByLex(Publisher<ZRangeByLexCommand> commands);
 }

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

@@ -15,6 +15,9 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
@@ -24,13 +27,10 @@ import org.reactivestreams.Publisher;
 import org.springframework.data.redis.connection.ClusterSlotHashUtil;
 import org.springframework.data.redis.connection.ReactiveClusterSetCommands;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
-import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
 import org.springframework.util.Assert;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
@@ -52,7 +52,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sUnion(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<SUnionCommand, ByteBuffer>> sUnion(Publisher<SUnionCommand> commands) {
+	public Flux<CommandResponse<SUnionCommand, Flux<ByteBuffer>>> sUnion(Publisher<SUnionCommand> commands) {
 
 		return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
 
@@ -62,10 +62,10 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 				return super.sUnion(Mono.just(command));
 			}
 
-			Mono<List<ByteBuffer>> result = Flux
-					.merge(command.getKeys().stream().map(cmd::smembers).collect(Collectors.toList())).distinct().collectList();
+			Flux<ByteBuffer> result = Flux.merge(command.getKeys().stream().map(cmd::smembers).collect(Collectors.toList()))
+					.distinct();
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -88,7 +88,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 			}
 
 			return sUnion(Mono.just(SUnionCommand.keys(command.getKeys()))).next().flatMap(values -> {
-				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
+				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().toStream().toArray(ByteBuffer[]::new));
 				return result.map(value -> new NumericResponse<>(command, value));
 			});
 		}));
@@ -98,7 +98,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sInter(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<SInterCommand, ByteBuffer>> sInter(Publisher<SInterCommand> commands) {
+	public Flux<CommandResponse<SInterCommand, Flux<ByteBuffer>>> sInter(Publisher<SInterCommand> commands) {
 
 		return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
 
@@ -122,7 +122,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 				return source;
 			});
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result.flatMap(v -> Flux.fromStream(v.stream()))));
 		}));
 	}
 
@@ -145,7 +145,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 			}
 
 			return sInter(Mono.just(SInterCommand.keys(command.getKeys()))).next().flatMap(values -> {
-				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
+				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().toStream().toArray(ByteBuffer[]::new));
 				return result.map(value -> new NumericResponse<>(command, value));
 			});
 		}));
@@ -155,7 +155,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveSetCommands#sDiff(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<SDiffCommand, ByteBuffer>> sDiff(Publisher<SDiffCommand> commands) {
+	public Flux<CommandResponse<SDiffCommand, Flux<ByteBuffer>>> sDiff(Publisher<SDiffCommand> commands) {
 
 		return getConnection().execute(cmd -> Flux.from(commands).flatMap(command -> {
 
@@ -179,7 +179,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 				return source;
 			});
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result.flatMap(v -> Flux.fromStream(v.stream()))));
 
 		}));
 	}
@@ -203,7 +203,7 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
 			}
 
 			return sDiff(Mono.just(SDiffCommand.keys(command.getKeys()))).next().flatMap(values -> {
-				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
+				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().toStream().toArray(ByteBuffer[]::new));
 				return result.map(value -> new NumericResponse<>(command, value));
 			});
 		}));

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

@@ -31,7 +31,6 @@ import org.reactivestreams.Publisher;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
-import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.geo.Point;
 import org.springframework.data.redis.connection.ReactiveGeoCommands;
@@ -157,7 +156,7 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveGeoCommands#geoRadius(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<CommandResponse<GeoRadiusCommand, GeoResults<GeoLocation<ByteBuffer>>>> geoRadius(
+	public Flux<CommandResponse<GeoRadiusCommand, Flux<GeoResult<GeoLocation<ByteBuffer>>>>> geoRadius(
 			Publisher<GeoRadiusCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -169,12 +168,13 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
 			GeoArgs geoArgs = command.getArgs().isPresent() ? LettuceConverters.toGeoArgs(command.getArgs().get())
 					: new GeoArgs();
 
-			Mono<GeoResults<GeoLocation<ByteBuffer>>> result = cmd.georadius(command.getKey(), command.getPoint().getX(),
-					command.getPoint().getY(), command.getDistance().getValue(),
-					LettuceConverters.toGeoArgsUnit(command.getDistance().getMetric()), geoArgs)
-					.map(converter(command.getDistance().getMetric())::convert).collectList().map(GeoResults::new);
+			Flux<GeoResult<GeoLocation<ByteBuffer>>> result = cmd
+					.georadius(command.getKey(), command.getPoint().getX(), command.getPoint().getY(),
+							command.getDistance().getValue(), LettuceConverters.toGeoArgsUnit(command.getDistance().getMetric()),
+							geoArgs) //
+					.map(converter(command.getDistance().getMetric())::convert);
 
-			return result.map(value -> new CommandResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -183,7 +183,7 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveGeoCommands#geoRadiusByMember(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<CommandResponse<GeoRadiusByMemberCommand, GeoResults<GeoLocation<ByteBuffer>>>> geoRadiusByMember(
+	public Flux<CommandResponse<GeoRadiusByMemberCommand, Flux<GeoResult<GeoLocation<ByteBuffer>>>>> geoRadiusByMember(
 			Publisher<GeoRadiusByMemberCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -195,12 +195,12 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
 			GeoArgs geoArgs = command.getArgs().isPresent() ? LettuceConverters.toGeoArgs(command.getArgs().get())
 					: new GeoArgs();
 
-			Mono<GeoResults<GeoLocation<ByteBuffer>>> result = cmd
+			Flux<GeoResult<GeoLocation<ByteBuffer>>> result = cmd
 					.georadiusbymember(command.getKey(), command.getMember(), command.getDistance().getValue(),
 							LettuceConverters.toGeoArgsUnit(command.getDistance().getMetric()), geoArgs)
-					.map(converter(command.getDistance().getMetric())::convert).collectList().map(GeoResults::new);
+					.map(converter(command.getDistance().getMetric())::convert);
 
-			return result.map(value -> new CommandResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 

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

@@ -168,15 +168,15 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveHashCommands#hKeys(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<KeyCommand, ByteBuffer>> hKeys(Publisher<KeyCommand> commands) {
+	public Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> hKeys(Publisher<KeyCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKey(), "Key must not be null!");
 
-			Mono<List<ByteBuffer>> result = cmd.hkeys(command.getKey()).collectList();
+			Flux<ByteBuffer> result = cmd.hkeys(command.getKey());
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -185,15 +185,15 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveHashCommands#hKeys(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<KeyCommand, ByteBuffer>> hVals(Publisher<KeyCommand> commands) {
+	public Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> hVals(Publisher<KeyCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKey(), "Key must not be null!");
 
-			Mono<List<ByteBuffer>> result = cmd.hvals(command.getKey()).collectList();
+			Flux<ByteBuffer> result = cmd.hvals(command.getKey());
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -202,7 +202,8 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveHashCommands#hGetAll(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<CommandResponse<KeyCommand, Map<ByteBuffer, ByteBuffer>>> hGetAll(Publisher<KeyCommand> commands) {
+	public Flux<CommandResponse<KeyCommand, Flux<Map.Entry<ByteBuffer, ByteBuffer>>>> hGetAll(
+			Publisher<KeyCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
@@ -210,7 +211,7 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
 
 			Mono<Map<ByteBuffer, ByteBuffer>> result = cmd.hgetall(command.getKey());
 
-			return result.map(value -> new CommandResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result.flatMapMany(v -> Flux.fromStream(v.entrySet().stream()))));
 		}));
 	}
 }

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

@@ -24,6 +24,7 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.ReactiveListCommands;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBufferResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
@@ -109,15 +110,15 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveListCommands#lRange(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<RangeCommand, ByteBuffer>> lRange(Publisher<RangeCommand> commands) {
+	public Flux<CommandResponse<RangeCommand, Flux<ByteBuffer>>> lRange(Publisher<RangeCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKey(), "Key must not be null!");
 			Assert.notNull(command.getRange(), "Range must not be null!");
 
-			return cmd.lrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
-					.collectList().map(value -> new MultiValueResponse<>(command, value));
+			Flux<ByteBuffer> result = cmd.lrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound());
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 

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

@@ -15,22 +15,20 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.List;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.ByteBufferResponse;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
-import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
 import org.springframework.data.redis.connection.ReactiveSetCommands;
 import org.springframework.util.Assert;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
@@ -155,14 +153,14 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveSetCommands#sInter(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<SInterCommand, ByteBuffer>> sInter(Publisher<SInterCommand> commands) {
+	public Flux<CommandResponse<SInterCommand, Flux<ByteBuffer>>> sInter(Publisher<SInterCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKeys(), "Keys must not be null!");
 
-			return cmd.sinter(command.getKeys().stream().toArray(ByteBuffer[]::new)).collectList()
-					.map(value -> new MultiValueResponse<>(command, value));
+			Flux<ByteBuffer> result = cmd.sinter(command.getKeys().stream().toArray(ByteBuffer[]::new));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -188,14 +186,14 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveSetCommands#sInter(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<SUnionCommand, ByteBuffer>> sUnion(Publisher<SUnionCommand> commands) {
+	public Flux<CommandResponse<SUnionCommand, Flux<ByteBuffer>>> sUnion(Publisher<SUnionCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKeys(), "Keys must not be null!");
 
-			return cmd.sunion(command.getKeys().stream().toArray(ByteBuffer[]::new)).collectList()
-					.map(value -> new MultiValueResponse<>(command, value));
+			Flux<ByteBuffer> result = cmd.sunion(command.getKeys().stream().toArray(ByteBuffer[]::new));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -221,14 +219,14 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveSetCommands#sInter(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<SDiffCommand, ByteBuffer>> sDiff(Publisher<SDiffCommand> commands) {
+	public Flux<CommandResponse<SDiffCommand, Flux<ByteBuffer>>> sDiff(Publisher<SDiffCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKeys(), "Keys must not be null!");
 
-			return cmd.sdiff(command.getKeys().stream().toArray(ByteBuffer[]::new)).collectList()
-					.map(value -> new MultiValueResponse<>(command, value));
+			Flux<ByteBuffer> result = cmd.sdiff(command.getKeys().stream().toArray(ByteBuffer[]::new));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -254,13 +252,14 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveSetCommands#sMembers(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<KeyCommand, ByteBuffer>> sMembers(Publisher<KeyCommand> commands) {
+	public Flux<CommandResponse<KeyCommand, Flux<ByteBuffer>>> sMembers(Publisher<KeyCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKey(), "Key must not be null!");
 
-			return cmd.smembers(command.getKey()).collectList().map(value -> new MultiValueResponse<>(command, value));
+			Flux<ByteBuffer> result = cmd.smembers(command.getKey());
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -269,7 +268,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveSetCommands#sRandMembers(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<SRandMembersCommand, ByteBuffer>> sRandMember(
+	public Flux<CommandResponse<SRandMembersCommand, Flux<ByteBuffer>>> sRandMember(
 			Publisher<SRandMembersCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
@@ -278,10 +277,10 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
 
 			boolean singleElement = !command.getCount().isPresent() || command.getCount().get().equals(1L);
 
-			Mono<List<ByteBuffer>> result = singleElement ? cmd.srandmember(command.getKey()).map(Collections::singletonList)
-					: cmd.srandmember(command.getKey(), command.getCount().get()).collectList();
+			Publisher<ByteBuffer> result = singleElement ? cmd.srandmember(command.getKey())
+					: cmd.srandmember(command.getKey(), command.getCount().get());
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, Flux.from(result)));
 		}));
 	}
 

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

@@ -32,8 +32,8 @@ import org.reactivestreams.Publisher;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.redis.connection.DefaultTuple;
+import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
-import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiValueResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
 import org.springframework.data.redis.connection.ReactiveZSetCommands;
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
@@ -174,43 +174,39 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zRange(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<ZRangeCommand, Tuple>> zRange(Publisher<ZRangeCommand> commands) {
+	public Flux<CommandResponse<ZRangeCommand, Flux<Tuple>>> zRange(Publisher<ZRangeCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
 			Assert.notNull(command.getKey(), "Key must not be null!");
 			Assert.notNull(command.getRange(), "Range must not be null!");
 
-			Mono<List<Tuple>> result;
+			Flux<Tuple> result;
 
 			if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
 				if (command.isWithScores()) {
 
-					result = cmd
-							.zrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
-									command.getRange().getUpperBound())
-							.map(sc -> (Tuple) new DefaultTuple(getBytes(sc), sc.getScore())).collectList();
+					result = cmd.zrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
+							command.getRange().getUpperBound()).map(sc -> (Tuple) new DefaultTuple(getBytes(sc), sc.getScore()));
 				} else {
 
 					result = cmd.zrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
-							.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
+							.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
 				}
 			} else {
 				if (command.isWithScores()) {
 
-					result = cmd
-							.zrevrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
-									command.getRange().getUpperBound())
-							.map(sc -> (Tuple) new DefaultTuple(getBytes(sc), sc.getScore())).collectList();
+					result = cmd.zrevrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
+							command.getRange().getUpperBound()).map(sc -> (Tuple) new DefaultTuple(getBytes(sc), sc.getScore()));
 				} else {
 
 					result = cmd
 							.zrevrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
-							.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
+							.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
 				}
 			}
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 
@@ -219,7 +215,8 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zRange(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<ZRangeByScoreCommand, Tuple>> zRangeByScore(Publisher<ZRangeByScoreCommand> commands) {
+	public Flux<CommandResponse<ZRangeByScoreCommand, Flux<Tuple>>> zRangeByScore(
+			Publisher<ZRangeByScoreCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
@@ -228,7 +225,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 
 			boolean isLimited = command.getLimit().isPresent();
 
-			Mono<List<Tuple>> result;
+			Publisher<Tuple> result;
 
 			if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
 
@@ -238,21 +235,21 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 
 					if (!isLimited) {
 						result = cmd.zrangebyscoreWithScores(command.getKey(), range)
-								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
+								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
 					} else {
 						result = cmd
 								.zrangebyscoreWithScores(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
-								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
+								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
 					}
 				} else {
 
 					if (!isLimited) {
 						result = cmd.zrangebyscore(command.getKey(), range)
-								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
+								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
 					} else {
 
 						result = cmd.zrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
-								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
+								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
 					}
 				}
 			} else {
@@ -263,28 +260,28 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 
 					if (!isLimited) {
 						result = cmd.zrevrangebyscoreWithScores(command.getKey(), range)
-								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
+								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
 					} else {
 
 						result = cmd
 								.zrevrangebyscoreWithScores(command.getKey(), range,
 										LettuceConverters.toLimit(command.getLimit().get()))
-								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore())).collectList();
+								.map(sc -> (Tuple) new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
 					}
 				} else {
 
 					if (!isLimited) {
 						result = cmd.zrevrangebyscore(command.getKey(), range)
-								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
+								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
 					} else {
 
 						result = cmd.zrevrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
-								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN)).collectList();
+								.map(value -> (Tuple) new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
 					}
 				}
 			}
 
-			return result.map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, Flux.from(result)));
 		}));
 	}
 
@@ -432,7 +429,8 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zRangeByLex(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<MultiValueResponse<ZRangeByLexCommand, ByteBuffer>> zRangeByLex(Publisher<ZRangeByLexCommand> commands) {
+	public Flux<CommandResponse<ZRangeByLexCommand, Flux<ByteBuffer>>> zRangeByLex(
+			Publisher<ZRangeByLexCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).flatMap(command -> {
 
@@ -457,7 +455,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 				}
 			}
 
-			return result.collectList().map(value -> new MultiValueResponse<>(command, value));
+			return Mono.just(new CommandResponse<>(command, result));
 		}));
 	}
 

src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.serializer.RedisSerializationContext;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -30,12 +29,12 @@ import org.reactivestreams.Publisher;
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
-import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.geo.Point;
 import org.springframework.data.redis.connection.ReactiveGeoCommands;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.util.Assert;
 
 /**
@@ -234,72 +233,60 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#geoRadius(java.lang.Object, org.springframework.data.geo.Circle)
 	 */
 	@Override
-	public Mono<List<GeoLocation<V>>> geoRadius(K key, Circle within) {
+	public Flux<GeoResult<GeoLocation<V>>> geoRadius(K key, Circle within) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(within, "Circle must not be null!");
 
-		return createMono(connection -> connection.geoRadius(rawKey(key), within) //
-				.flatMapMany(Flux::fromIterable) //
-				.map(location -> new GeoLocation<>(readValue(location.getName()), location.getPoint())) //
-				.collectList());
+		return createFlux(connection -> connection.geoRadius(rawKey(key), within).map(this::readGeoResult));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#geoRadius(java.lang.Object, org.springframework.data.geo.Circle, org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs)
 	 */
 	@Override
-	public Mono<GeoResults<GeoLocation<V>>> geoRadius(K key, Circle within, GeoRadiusCommandArgs args) {
+	public Flux<GeoResult<GeoLocation<V>>> geoRadius(K key, Circle within, GeoRadiusCommandArgs args) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(within, "Circle must not be null!");
 		Assert.notNull(args, "GeoRadiusCommandArgs must not be null!");
 
-		return createMono(connection -> connection.geoRadius(rawKey(key), within, args) //
-				.flatMapMany(Flux::fromIterable) //
-				.map(geoResult -> new GeoResult<>(
-						new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()),
-						geoResult.getDistance())) //
-				.collectList() //
-				.map(GeoResults::new));
+		return createFlux(connection -> connection.geoRadius(rawKey(key), within, args) //
+				.map(this::readGeoResult));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#geoRadiusByMember(java.lang.Object, java.lang.Object, double)
 	 */
 	@Override
-	public Mono<List<GeoLocation<V>>> geoRadiusByMember(K key, V member, double radius) {
+	public Flux<GeoResult<GeoLocation<V>>> geoRadiusByMember(K key, V member, double radius) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(member, "Member must not be null!");
 
-		return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), new Distance(radius)) //
-				.flatMapMany(Flux::fromIterable) //
-				.map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) //
-				.collectList());
+		return createFlux(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), new Distance(radius)) //
+				.map(this::readGeoResult));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#geoRadiusByMember(java.lang.Object, java.lang.Object, org.springframework.data.geo.Distance)
 	 */
 	@Override
-	public Mono<List<GeoLocation<V>>> geoRadiusByMember(K key, V member, Distance distance) {
+	public Flux<GeoResult<GeoLocation<V>>> geoRadiusByMember(K key, V member, Distance distance) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(member, "Member must not be null!");
 		Assert.notNull(distance, "Distance must not be null!");
 
-		return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance) //
-				.flatMapMany(Flux::fromIterable) //
-				.map(geoLocation -> new GeoLocation<>(readValue(geoLocation.getName()), geoLocation.getPoint())) //
-				.collectList());
+		return createFlux(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance) //
+				.map(this::readGeoResult));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#geoRadiusByMember(java.lang.Object, java.lang.Object, org.springframework.data.geo.Distance, org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs)
 	 */
 	@Override
-	public Mono<GeoResults<GeoLocation<V>>> geoRadiusByMember(K key, V member, Distance distance,
+	public Flux<GeoResult<GeoLocation<V>>> geoRadiusByMember(K key, V member, Distance distance,
 			GeoRadiusCommandArgs args) {
 
 		Assert.notNull(key, "Key must not be null!");
@@ -307,13 +294,8 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
 		Assert.notNull(distance, "Distance must not be null!");
 		Assert.notNull(args, "GeoRadiusCommandArgs must not be null!");
 
-		return createMono(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance, args) //
-				.flatMapMany(Flux::fromIterable) //
-				.map(geoResult -> new GeoResult<>(
-						new GeoLocation<>(readValue(geoResult.getContent().getName()), geoResult.getContent().getPoint()),
-						geoResult.getDistance())) //
-				.collectList() //
-				.map(GeoResults::new));
+		return createFlux(connection -> connection.geoRadiusByMember(rawKey(key), rawValue(member), distance, args))
+				.map(this::readGeoResult);
 	}
 
 	/* (non-Javadoc)
@@ -369,4 +351,10 @@ public class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations
 	private V readValue(ByteBuffer buffer) {
 		return serializationContext.getValueSerializationPair().read(buffer);
 	}
+
+	private GeoResult<GeoLocation<V>> readGeoResult(GeoResult<GeoLocation<ByteBuffer>> source) {
+
+		return new GeoResult<>(new GeoLocation(readValue(source.getContent().getName()), source.getContent().getPoint()),
+				source.getDistance());
+	}
 }

src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java

@@ -21,6 +21,7 @@ import reactor.core.publisher.Mono;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -151,14 +152,12 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
 	 * @see org.springframework.data.redis.core.ReactiveHashOperations#keys(java.lang.Object)
 	 */
 	@Override
-	public Mono<List<HK>> keys(H key) {
+	public Flux<HK> keys(H key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.hKeys(rawKey(key)) //
-				.flatMapMany(Flux::fromIterable) //
-				.map(this::readHashKey) //
-				.collectList());
+		return createFlux(connection -> connection.hKeys(rawKey(key)) //
+				.map(this::readHashKey));
 	}
 
 	/* (non-Javadoc)
@@ -216,26 +215,24 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
 	 * @see org.springframework.data.redis.core.ReactiveHashOperations#values(java.lang.Object)
 	 */
 	@Override
-	public Mono<List<HV>> values(H key) {
+	public Flux<HV> values(H key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.hVals(rawKey(key)) //
-				.flatMapMany(Flux::fromIterable) //
-				.map(this::readHashValue) //
-				.collectList());
+		return createFlux(connection -> connection.hVals(rawKey(key)) //
+				.map(this::readHashValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveHashOperations#entries(java.lang.Object)
 	 */
 	@Override
-	public Mono<Map<HK, HV>> entries(H key) {
+	public Flux<Map.Entry<HK, HV>> entries(H key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.hGetAll(rawKey(key)) //
-				.map(this::deserializeHashEntries));
+		return createFlux(connection -> connection.hGetAll(rawKey(key)) //
+				.map(this::deserializeHashEntry));
 	}
 
 	/* (non-Javadoc)
@@ -256,6 +253,13 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
 		return template.createMono(connection -> function.apply(connection.hashCommands()));
 	}
 
+	private <T> Flux<T> createFlux(Function<ReactiveHashCommands, Publisher<T>> function) {
+
+		Assert.notNull(function, "Function must not be null!");
+
+		return template.createFlux(connection -> function.apply(connection.hashCommands()));
+	}
+
 	private ByteBuffer rawKey(H key) {
 		return serializationContext.getKeySerializationPair().write(key);
 	}
@@ -278,15 +282,9 @@ public class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOpe
 		return (HV) serializationContext.getHashValueSerializationPair().read(value);
 	}
 
-	private Map<HK, HV> deserializeHashEntries(Map<ByteBuffer, ByteBuffer> source) {
-
-		Map<HK, HV> deserialized = new LinkedHashMap<>(source.size());
-
-		source.forEach((k, v) -> {
-			deserialized.put(readHashKey(k), readHashValue(v));
-		});
-
-		return deserialized;
+	private Map.Entry<HK, HV> deserializeHashEntry(Map.Entry<ByteBuffer, ByteBuffer> source) {
+		return Collections.singletonMap(readHashKey(source.getKey()), readHashValue(source.getValue())).entrySet()
+				.iterator().next();
 	}
 
 	private List<HV> deserializeHashValues(List<ByteBuffer> source) {

src/main/java/org/springframework/data/redis/core/DefaultReactiveListOperations.java

@@ -20,11 +20,9 @@ import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
@@ -66,11 +64,11 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
 	 * @see org.springframework.data.redis.core.ReactiveListOperations#range(java.lang.Object, long, long)
 	 */
 	@Override
-	public Mono<List<V>> range(K key, long start, long end) {
+	public Flux<V> range(K key, long start, long end) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.lRange(rawKey(key), start, end).map(this::deserializeValues));
+		return createFlux(connection -> connection.lRange(rawKey(key), start, end).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
@@ -340,6 +338,13 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
 		return template.createMono(connection -> function.apply(connection.listCommands()));
 	}
 
+	private <T> Flux<T> createFlux(Function<ReactiveListCommands, Publisher<T>> function) {
+
+		Assert.notNull(function, "Function must not be null!");
+
+		return template.createFlux(connection -> function.apply(connection.listCommands()));
+	}
+
 	private boolean isZeroOrGreater1Second(Duration timeout) {
 		return timeout.isZero() || timeout.getNano() % TimeUnit.NANOSECONDS.convert(1, TimeUnit.SECONDS) == 0;
 	}
@@ -355,15 +360,4 @@ public class DefaultReactiveListOperations<K, V> implements ReactiveListOperatio
 	private V readValue(ByteBuffer buffer) {
 		return serializationContext.getValueSerializationPair().read(buffer);
 	}
-
-	private List<V> deserializeValues(List<ByteBuffer> source) {
-
-		List<V> result = new ArrayList<V>(source.size());
-
-		for (ByteBuffer buffer : source) {
-			result.add(readValue(buffer));
-		}
-
-		return result;
-	}
 }

src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java

@@ -22,9 +22,7 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Set;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
@@ -147,7 +145,7 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#intersect(java.lang.Object, java.lang.Object)
 	 */
 	@Override
-	public Mono<Set<V>> intersect(K key, K otherKey) {
+	public Flux<V> intersect(K key, K otherKey) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(otherKey, "Other key must not be null!");
@@ -159,16 +157,16 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#intersect(java.lang.Object, java.util.Collection)
 	 */
 	@Override
-	public Mono<Set<V>> intersect(K key, Collection<K> otherKeys) {
+	public Flux<V> intersect(K key, Collection<K> otherKeys) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(otherKeys, "Other keys must not be null!");
 
-		return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
+		return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
 				.map(this::rawKey) //
 				.collectList() //
-				.flatMap(connection::sInter) //
-				.map(this::readValueSet));
+				.flatMapMany(connection::sInter) //
+				.map(this::readValue));
 	}
 
 	/* (non-Javadoc)
@@ -204,7 +202,7 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#union(java.lang.Object, java.lang.Object)
 	 */
 	@Override
-	public Mono<Set<V>> union(K key, K otherKey) {
+	public Flux<V> union(K key, K otherKey) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(otherKey, "Other key must not be null!");
@@ -216,16 +214,16 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#union(java.lang.Object, java.util.Collection)
 	 */
 	@Override
-	public Mono<Set<V>> union(K key, Collection<K> otherKeys) {
+	public Flux<V> union(K key, Collection<K> otherKeys) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(otherKeys, "Other keys must not be null!");
 
-		return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
+		return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
 				.map(this::rawKey) //
 				.collectList() //
-				.flatMap(connection::sUnion) //
-				.map(this::readValueSet));
+				.flatMapMany(connection::sUnion) //
+				.map(this::readValue));
 	}
 
 	/* (non-Javadoc)
@@ -261,7 +259,7 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#difference(java.lang.Object, java.lang.Object)
 	 */
 	@Override
-	public Mono<Set<V>> difference(K key, K otherKey) {
+	public Flux<V> difference(K key, K otherKey) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(otherKey, "Other key must not be null!");
@@ -273,16 +271,16 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#difference(java.lang.Object, java.util.Collection)
 	 */
 	@Override
-	public Mono<Set<V>> difference(K key, Collection<K> otherKeys) {
+	public Flux<V> difference(K key, Collection<K> otherKeys) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(otherKeys, "Other keys must not be null!");
 
-		return createMono(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
+		return createFlux(connection -> Flux.fromIterable(getKeys(key, otherKeys)) //
 				.map(this::rawKey) //
 				.collectList() //
-				.flatMap(connection::sDiff) //
-				.map(this::readValueSet));
+				.flatMapMany(connection::sDiff) //
+				.map(this::readValue));
 	}
 
 	/* (non-Javadoc)
@@ -318,11 +316,11 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#members(java.lang.Object)
 	 */
 	@Override
-	public Mono<Set<V>> members(K key) {
+	public Flux<V> members(K key) {
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.sMembers(rawKey(key)).map(this::readValueSet));
+		return createFlux(connection -> connection.sMembers(rawKey(key)).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
@@ -340,22 +338,22 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#distinctRandomMembers(java.lang.Object, long)
 	 */
 	@Override
-	public Mono<Set<V>> distinctRandomMembers(K key, long count) {
+	public Flux<V> distinctRandomMembers(K key, long count) {
 
 		Assert.isTrue(count > 0, "Negative count not supported. Use randomMembers to allow duplicate elements.");
 
-		return createMono(connection -> connection.sRandMember(rawKey(key), count).map(this::readValueSet));
+		return createFlux(connection -> connection.sRandMember(rawKey(key), count).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveSetOperations#randomMembers(java.lang.Object, long)
 	 */
 	@Override
-	public Mono<List<V>> randomMembers(K key, long count) {
+	public Flux<V> randomMembers(K key, long count) {
 
 		Assert.isTrue(count > 0, "Use a positive number for count. This method is already allowing duplicate elements.");
 
-		return createMono(connection -> connection.sRandMember(rawKey(key), -count).map(this::readValueList));
+		return createFlux(connection -> connection.sRandMember(rawKey(key), -count).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
@@ -376,6 +374,13 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 		return template.createMono(connection -> function.apply(connection.setCommands()));
 	}
 
+	private <T> Flux<T> createFlux(Function<ReactiveSetCommands, Publisher<T>> function) {
+
+		Assert.notNull(function, "Function must not be null!");
+
+		return template.createFlux(connection -> function.apply(connection.setCommands()));
+	}
+
 	private ByteBuffer rawKey(K key) {
 		return serializationContext.getKeySerializationPair().write(key);
 	}
@@ -397,26 +402,4 @@ public class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations
 	private V readValue(ByteBuffer buffer) {
 		return serializationContext.getValueSerializationPair().read(buffer);
 	}
-
-	private Set<V> readValueSet(Collection<ByteBuffer> raw) {
-
-		Set<V> result = new LinkedHashSet<>(raw.size());
-
-		for (ByteBuffer buffer : raw) {
-			result.add(readValue(buffer));
-		}
-
-		return result;
-	}
-
-	private List<V> readValueList(Collection<ByteBuffer> raw) {
-
-		List<V> result = new ArrayList<>(raw.size());
-
-		for (ByteBuffer buffer : raw) {
-			result.add(readValue(buffer));
-		}
-
-		return result;
-	}
 }

src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java

@@ -22,9 +22,7 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Set;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
@@ -151,150 +149,149 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#range(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<V>> range(K key, Range<Long> range) {
+	public Flux<V> range(K key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRange(rawKey(key), range).map(this::readValueSet));
+		return createFlux(connection -> connection.zRange(rawKey(key), range).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeWithScores(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<TypedTuple<V>>> rangeWithScores(K key, Range<Long> range) {
+	public Flux<TypedTuple<V>> rangeWithScores(K key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRangeWithScores(rawKey(key), range).map(this::readTypedTupleSet));
+		return createFlux(connection -> connection.zRangeWithScores(rawKey(key), range).map(this::readTypedTuple));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeByScore(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<V>> rangeByScore(K key, Range<Double> range) {
+	public Flux<V> rangeByScore(K key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRangeByScore(rawKey(key), range).map(this::readValueSet));
+		return createFlux(connection -> connection.zRangeByScore(rawKey(key), range).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeByScoreWithScores(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<TypedTuple<V>>> rangeByScoreWithScores(K key, Range<Double> range) {
+	public Flux<TypedTuple<V>> rangeByScoreWithScores(K key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(
-				connection -> connection.zRangeByScoreWithScores(rawKey(key), range).map(this::readTypedTupleSet));
+		return createFlux(connection -> connection.zRangeByScoreWithScores(rawKey(key), range).map(this::readTypedTuple));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeByScore(java.lang.Object, org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
 	 */
 	@Override
-	public Mono<Set<V>> rangeByScore(K key, Range<Double> range, Limit limit) {
+	public Flux<V> rangeByScore(K key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRangeByScore(rawKey(key), range, limit).map(this::readValueSet));
+		return createFlux(connection -> connection.zRangeByScore(rawKey(key), range, limit).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeByScoreWithScores(java.lang.Object, org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
 	 */
 	@Override
-	public Mono<Set<TypedTuple<V>>> rangeByScoreWithScores(K key, Range<Double> range, Limit limit) {
+	public Flux<TypedTuple<V>> rangeByScoreWithScores(K key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 		Assert.notNull(limit, "Limit must not be null!");
 
-		return createMono(
-				connection -> connection.zRangeByScoreWithScores(rawKey(key), range, limit).map(this::readTypedTupleSet));
+		return createFlux(
+				connection -> connection.zRangeByScoreWithScores(rawKey(key), range, limit).map(this::readTypedTuple));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRange(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<V>> reverseRange(K key, Range<Long> range) {
+	public Flux<V> reverseRange(K key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRevRange(rawKey(key), range).map(this::readValueSet));
+		return createFlux(connection -> connection.zRevRange(rawKey(key), range).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRangeWithScores(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<TypedTuple<V>>> reverseRangeWithScores(K key, Range<Long> range) {
+	public Flux<TypedTuple<V>> reverseRangeWithScores(K key, Range<Long> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRevRangeWithScores(rawKey(key), range).map(this::readTypedTupleSet));
+		return createFlux(connection -> connection.zRevRangeWithScores(rawKey(key), range).map(this::readTypedTuple));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRangeByScore(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<V>> reverseRangeByScore(K key, Range<Double> range) {
+	public Flux<V> reverseRangeByScore(K key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRevRangeByScore(rawKey(key), range).map(this::readValueSet));
+		return createFlux(connection -> connection.zRevRangeByScore(rawKey(key), range).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRangeByScoreWithScores(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<TypedTuple<V>>> reverseRangeByScoreWithScores(K key, Range<Double> range) {
+	public Flux<TypedTuple<V>> reverseRangeByScoreWithScores(K key, Range<Double> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(
-				connection -> connection.zRevRangeByScoreWithScores(rawKey(key), range).map(this::readTypedTupleSet));
+		return createFlux(
+				connection -> connection.zRevRangeByScoreWithScores(rawKey(key), range).map(this::readTypedTuple));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRangeByScore(java.lang.Object, org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
 	 */
 	@Override
-	public Mono<Set<V>> reverseRangeByScore(K key, Range<Double> range, Limit limit) {
+	public Flux<V> reverseRangeByScore(K key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRevRangeByScore(rawKey(key), range, limit).map(this::readValueSet));
+		return createFlux(connection -> connection.zRevRangeByScore(rawKey(key), range, limit).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRangeByScoreWithScores(java.lang.Object, org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
 	 */
 	@Override
-	public Mono<Set<TypedTuple<V>>> reverseRangeByScoreWithScores(K key, Range<Double> range, Limit limit) {
+	public Flux<TypedTuple<V>> reverseRangeByScoreWithScores(K key, Range<Double> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 		Assert.notNull(limit, "Limit must not be null!");
 
-		return createMono(
-				connection -> connection.zRevRangeByScoreWithScores(rawKey(key), range, limit).map(this::readTypedTupleSet));
+		return createFlux(
+				connection -> connection.zRevRangeByScoreWithScores(rawKey(key), range, limit).map(this::readTypedTuple));
 	}
 
 	/* (non-Javadoc)
@@ -418,50 +415,50 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeByLex(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<V>> rangeByLex(K key, Range<String> range) {
+	public Flux<V> rangeByLex(K key, Range<String> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRangeByLex(rawKey(key), range).map(this::readValueSet));
+		return createFlux(connection -> connection.zRangeByLex(rawKey(key), range).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#rangeByLex(java.lang.Object, org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
 	 */
 	@Override
-	public Mono<Set<V>> rangeByLex(K key, Range<String> range, Limit limit) {
+	public Flux<V> rangeByLex(K key, Range<String> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 		Assert.notNull(limit, "Limit must not be null!");
 
-		return createMono(connection -> connection.zRangeByLex(rawKey(key), range, limit).map(this::readValueSet));
+		return createFlux(connection -> connection.zRangeByLex(rawKey(key), range, limit).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRangeByLex(java.lang.Object, org.springframework.data.domain.Range)
 	 */
 	@Override
-	public Mono<Set<V>> reverseRangeByLex(K key, Range<String> range) {
+	public Flux<V> reverseRangeByLex(K key, Range<String> range) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 
-		return createMono(connection -> connection.zRevRangeByLex(rawKey(key), range).map(this::readValueSet));
+		return createFlux(connection -> connection.zRevRangeByLex(rawKey(key), range).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#reverseRangeByLex(java.lang.Object, org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
 	 */
 	@Override
-	public Mono<Set<V>> reverseRangeByLex(K key, Range<String> range, Limit limit) {
+	public Flux<V> reverseRangeByLex(K key, Range<String> range, Limit limit) {
 
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(range, "Range must not be null!");
 		Assert.notNull(limit, "Limit must not be null!");
 
-		return createMono(connection -> connection.zRevRangeByLex(rawKey(key), range, limit).map(this::readValueSet));
+		return createFlux(connection -> connection.zRevRangeByLex(rawKey(key), range, limit).map(this::readValue));
 	}
 
 	/* (non-Javadoc)
@@ -482,6 +479,13 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
 		return template.createMono(connection -> function.apply(connection.zSetCommands()));
 	}
 
+	private <T> Flux<T> createFlux(Function<ReactiveZSetCommands, Publisher<T>> function) {
+
+		Assert.notNull(function, "Function must not be null!");
+
+		return template.createFlux(connection -> function.apply(connection.zSetCommands()));
+	}
+
 	private ByteBuffer rawKey(K key) {
 		return serializationContext.getKeySerializationPair().write(key);
 	}
@@ -504,25 +508,7 @@ public class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperatio
 		return serializationContext.getValueSerializationPair().read(buffer);
 	}
 
-	private Set<V> readValueSet(Collection<ByteBuffer> raw) {
-
-		Set<V> result = new LinkedHashSet<>(raw.size());
-
-		for (ByteBuffer buffer : raw) {
-			result.add(readValue(buffer));
-		}
-
-		return result;
-	}
-
-	private Set<TypedTuple<V>> readTypedTupleSet(Collection<Tuple> raw) {
-
-		Set<TypedTuple<V>> result = new LinkedHashSet<>(raw.size());
-
-		for (Tuple tuple : raw) {
-			result.add(new DefaultTypedTuple<>(readValue(ByteBuffer.wrap(tuple.getValue())), tuple.getScore()));
-		}
-
-		return result;
+	private TypedTuple<V> readTypedTuple(Tuple raw) {
+		return new DefaultTypedTuple<>(readValue(ByteBuffer.wrap(raw.getValue())), raw.getScore());
 	}
 }

src/main/java/org/springframework/data/redis/core/ReactiveGeoOperations.java

@@ -25,7 +25,7 @@ import java.util.Map;
 import org.reactivestreams.Publisher;
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
-import org.springframework.data.geo.GeoResults;
+import org.springframework.data.geo.GeoResult;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.geo.Point;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
@@ -35,6 +35,7 @@ import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusComma
  * Reactive Redis operations for geo commands.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @see <a href="http://redis.io/commands#geo">Redis Documentation: Geo Commands</a>
  * @since 2.0
  */
@@ -162,7 +163,7 @@ public interface ReactiveGeoOperations<K, M> {
 	 * @return never {@literal null}.
 	 * @see <a href="http://redis.io/commands/georadius">Redis Documentation: GEORADIUS</a>
 	 */
-	Mono<List<GeoLocation<M>>> geoRadius(K key, Circle within);
+	Flux<GeoResult<GeoLocation<M>>> geoRadius(K key, Circle within);
 
 	/**
 	 * Get the {@literal member}s within the boundaries of a given {@link Circle} applying {@link GeoRadiusCommandArgs}.
@@ -173,7 +174,7 @@ public interface ReactiveGeoOperations<K, M> {
 	 * @return never {@literal null}.
 	 * @see <a href="http://redis.io/commands/georadius">Redis Documentation: GEORADIUS</a>
 	 */
-	Mono<GeoResults<GeoLocation<M>>> geoRadius(K key, Circle within, GeoRadiusCommandArgs args);
+	Flux<GeoResult<GeoLocation<M>>> geoRadius(K key, Circle within, GeoRadiusCommandArgs args);
 
 	/**
 	 * Get the {@literal member}s within the circle defined by the {@literal members} coordinates and given
@@ -185,7 +186,7 @@ public interface ReactiveGeoOperations<K, M> {
 	 * @return never {@literal null}.
 	 * @see <a href="http://redis.io/commands/georadiusbymember">Redis Documentation: GEORADIUSBYMEMBER</a>
 	 */
-	Mono<List<GeoLocation<M>>> geoRadiusByMember(K key, M member, double radius);
+	Flux<GeoResult<GeoLocation<M>>> geoRadiusByMember(K key, M member, double radius);
 
 	/**
 	 * Get the {@literal member}s within the circle defined by the {@literal members} coordinates and given
@@ -197,7 +198,7 @@ public interface ReactiveGeoOperations<K, M> {
 	 * @return never {@literal null}.
 	 * @see <a href="http://redis.io/commands/georadiusbymember">Redis Documentation: GEORADIUSBYMEMBER</a>
 	 */
-	Mono<List<GeoLocation<M>>> geoRadiusByMember(K key, M member, Distance distance);
+	Flux<GeoResult<GeoLocation<M>>> geoRadiusByMember(K key, M member, Distance distance);
 
 	/**
 	 * Get the {@literal member}s within the circle defined by the {@literal members} coordinates and given
@@ -210,7 +211,7 @@ public interface ReactiveGeoOperations<K, M> {
 	 * @return never {@literal null}.
 	 * @see <a href="http://redis.io/commands/georadiusbymember">Redis Documentation: GEORADIUSBYMEMBER</a>
 	 */
-	Mono<GeoResults<GeoLocation<M>>> geoRadiusByMember(K key, M member, Distance distance, GeoRadiusCommandArgs args);
+	Flux<GeoResult<GeoLocation<M>>> geoRadiusByMember(K key, M member, Distance distance, GeoRadiusCommandArgs args);
 
 	/**
 	 * Remove the {@literal member}s.

src/main/java/org/springframework/data/redis/core/ReactiveHashOperations.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.core;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.Collection;
@@ -25,6 +26,7 @@ import java.util.Map;
  * Redis map specific operations working on a hash.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
 public interface ReactiveHashOperations<H, HK, HV> {
@@ -91,7 +93,7 @@ public interface ReactiveHashOperations<H, HK, HV> {
 	 * @param key must not be {@literal null}.
 	 * @return
 	 */
-	Mono<List<HK>> keys(H key);
+	Flux<HK> keys(H key);
 
 	/**
 	 * Get size of hash at {@code key}.
@@ -134,7 +136,7 @@ public interface ReactiveHashOperations<H, HK, HV> {
 	 * @param key must not be {@literal null}.
 	 * @return
 	 */
-	Mono<List<HV>> values(H key);
+	Flux<HV> values(H key);
 
 	/**
 	 * Get entire hash stored at {@code key}.
@@ -142,7 +144,7 @@ public interface ReactiveHashOperations<H, HK, HV> {
 	 * @param key must not be {@literal null}.
 	 * @return
 	 */
-	Mono<Map<HK, HV>> entries(H key);
+	Flux<Map.Entry<HK, HV>> entries(H key);
 
 	/**
 	 * Removes the given {@literal key}.

src/main/java/org/springframework/data/redis/core/ReactiveListOperations.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.core;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
@@ -25,6 +26,8 @@ import java.util.List;
  * Redis list specific operations.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
+ * @see <a href="http://redis.io/commands#list">Redis Documentation: List Commands</a>
  * @since 2.0
  */
 public interface ReactiveListOperations<K, V> {
@@ -38,7 +41,7 @@ public interface ReactiveListOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/lrange">Redis Documentation: LRANGE</a>
 	 */
-	Mono<List<V>> range(K key, long start, long end);
+	Flux<V> range(K key, long start, long end);
 
 	/**
 	 * Trim list at {@code key} to elements between {@code start} and {@code end}.

src/main/java/org/springframework/data/redis/core/ReactiveSetOperations.java

@@ -15,16 +15,17 @@
  */
 package org.springframework.data.redis.core;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.Collection;
-import java.util.List;
-import java.util.Set;
 
 /**
  * Redis set specific operations.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
+ * @see <a href="http://redis.io/commands#set">Redis Documentation: Set Commands</a>
  * @since 2.0
  */
 public interface ReactiveSetOperations<K, V> {
@@ -96,7 +97,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sinter">Redis Documentation: SINTER</a>
 	 */
-	Mono<Set<V>> intersect(K key, K otherKey);
+	Flux<V> intersect(K key, K otherKey);
 
 	/**
 	 * Returns the members intersecting all given sets at {@code key} and {@code otherKeys}.
@@ -106,7 +107,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sinter">Redis Documentation: SINTER</a>
 	 */
-	Mono<Set<V>> intersect(K key, Collection<K> otherKeys);
+	Flux<V> intersect(K key, Collection<K> otherKeys);
 
 	/**
 	 * Intersect all given sets at {@code key} and {@code otherKey} and store result in {@code destKey}.
@@ -138,7 +139,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sunion">Redis Documentation: SUNION</a>
 	 */
-	Mono<Set<V>> union(K key, K otherKey);
+	Flux<V> union(K key, K otherKey);
 
 	/**
 	 * Union all sets at given {@code keys} and {@code otherKeys}.
@@ -148,7 +149,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sunion">Redis Documentation: SUNION</a>
 	 */
-	Mono<Set<V>> union(K key, Collection<K> otherKeys);
+	Flux<V> union(K key, Collection<K> otherKeys);
 
 	/**
 	 * Union all sets at given {@code key} and {@code otherKey} and store result in {@code destKey}.
@@ -180,7 +181,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sdiff">Redis Documentation: SDIFF</a>
 	 */
-	Mono<Set<V>> difference(K key, K otherKey);
+	Flux<V> difference(K key, K otherKey);
 
 	/**
 	 * Diff all sets for given {@code key} and {@code otherKeys}.
@@ -190,7 +191,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/sdiff">Redis Documentation: SDIFF</a>
 	 */
-	Mono<Set<V>> difference(K key, Collection<K> otherKeys);
+	Flux<V> difference(K key, Collection<K> otherKeys);
 
 	/**
 	 * Diff all sets for given {@code key} and {@code otherKey} and store result in {@code destKey}.
@@ -221,7 +222,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/smembers">Redis Documentation: SMEMBERS</a>
 	 */
-	Mono<Set<V>> members(K key);
+	Flux<V> members(K key);
 
 	/**
 	 * Get random element from set at {@code key}.
@@ -240,7 +241,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/srandmember">Redis Documentation: SRANDMEMBER</a>
 	 */
-	Mono<Set<V>> distinctRandomMembers(K key, long count);
+	Flux<V> distinctRandomMembers(K key, long count);
 
 	/**
 	 * Get {@code count} random elements from set at {@code key}.
@@ -250,7 +251,7 @@ public interface ReactiveSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/srandmember">Redis Documentation: SRANDMEMBER</a>
 	 */
-	Mono<List<V>> randomMembers(K key, long count);
+	Flux<V> randomMembers(K key, long count);
 
 	/**
 	 * Removes the given {@literal key}.

src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java

@@ -15,10 +15,10 @@
  */
 package org.springframework.data.redis.core;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.Collection;
-import java.util.Set;
 
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
@@ -29,6 +29,8 @@ import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
  * Redis ZSet/sorted set specific operations.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
+ * @see <a href="http://redis.io/commands#zset">Redis Documentation: Sorted Set Commands</a>
  * @since 2.0
  */
 public interface ReactiveZSetOperations<K, V> {
@@ -103,7 +105,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrange">Redis Documentation: ZRANGE</a>
 	 */
-	Mono<Set<V>> range(K key, Range<Long> range);
+	Flux<V> range(K key, Range<Long> range);
 
 	/**
 	 * Get set of {@link Tuple}s between {@code start} and {@code end} from sorted set.
@@ -113,7 +115,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrange">Redis Documentation: ZRANGE</a>
 	 */
-	Mono<Set<TypedTuple<V>>> rangeWithScores(K key, Range<Long> range);
+	Flux<TypedTuple<V>> rangeWithScores(K key, Range<Long> range);
 
 	/**
 	 * Get elements where score is between {@code min} and {@code max} from sorted set.
@@ -123,7 +125,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	Mono<Set<V>> rangeByScore(K key, Range<Double> range);
+	Flux<V> rangeByScore(K key, Range<Double> range);
 
 	/**
 	 * Get set of {@link Tuple}s where score is between {@code min} and {@code max} from sorted set.
@@ -133,7 +135,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	Mono<Set<TypedTuple<V>>> rangeByScoreWithScores(K key, Range<Double> range);
+	Flux<TypedTuple<V>> rangeByScoreWithScores(K key, Range<Double> range);
 
 	/**
 	 * Get elements in range from {@code start} to {@code end} where score is between {@code min} and {@code max} from
@@ -145,7 +147,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	Mono<Set<V>> rangeByScore(K key, Range<Double> range, Limit limit);
+	Flux<V> rangeByScore(K key, Range<Double> range, Limit limit);
 
 	/**
 	 * Get set of {@link Tuple}s in range from {@code start} to {@code end} where score is between {@code min} and
@@ -157,7 +159,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebyscore">Redis Documentation: ZRANGEBYSCORE</a>
 	 */
-	Mono<Set<TypedTuple<V>>> rangeByScoreWithScores(K key, Range<Double> range, Limit limit);
+	Flux<TypedTuple<V>> rangeByScoreWithScores(K key, Range<Double> range, Limit limit);
 
 	/**
 	 * Get elements in range from {@code start} to {@code end} from sorted set ordered from high to low.
@@ -167,7 +169,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrange">Redis Documentation: ZREVRANGE</a>
 	 */
-	Mono<Set<V>> reverseRange(K key, Range<Long> range);
+	Flux<V> reverseRange(K key, Range<Long> range);
 
 	/**
 	 * Get set of {@link Tuple}s in range from {@code start} to {@code end} from sorted set ordered from high to low.
@@ -177,7 +179,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrange">Redis Documentation: ZREVRANGE</a>
 	 */
-	Mono<Set<TypedTuple<V>>> reverseRangeWithScores(K key, Range<Long> range);
+	Flux<TypedTuple<V>> reverseRangeWithScores(K key, Range<Long> range);
 
 	/**
 	 * Get elements where score is between {@code min} and {@code max} from sorted set ordered from high to low.
@@ -187,7 +189,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrange">Redis Documentation: ZREVRANGE</a>
 	 */
-	Mono<Set<V>> reverseRangeByScore(K key, Range<Double> range);
+	Flux<V> reverseRangeByScore(K key, Range<Double> range);
 
 	/**
 	 * Get set of {@link Tuple} where score is between {@code min} and {@code max} from sorted set ordered from high to
@@ -198,7 +200,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	Mono<Set<TypedTuple<V>>> reverseRangeByScoreWithScores(K key, Range<Double> range);
+	Flux<TypedTuple<V>> reverseRangeByScoreWithScores(K key, Range<Double> range);
 
 	/**
 	 * Get elements in range from {@code start} to {@code end} where score is between {@code min} and {@code max} from
@@ -210,7 +212,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	Mono<Set<V>> reverseRangeByScore(K key, Range<Double> range, Limit limit);
+	Flux<V> reverseRangeByScore(K key, Range<Double> range, Limit limit);
 
 	/**
 	 * Get set of {@link Tuple} in range from {@code start} to {@code end} where score is between {@code min} and
@@ -222,7 +224,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebyscore">Redis Documentation: ZREVRANGEBYSCORE</a>
 	 */
-	Mono<Set<TypedTuple<V>>> reverseRangeByScoreWithScores(K key, Range<Double> range, Limit limit);
+	Flux<TypedTuple<V>> reverseRangeByScoreWithScores(K key, Range<Double> range, Limit limit);
 
 	/**
 	 * Count number of elements within sorted set with scores between {@code min} and {@code max}.
@@ -325,7 +327,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @param range must not be {@literal null}.
 	 * @see <a href="http://redis.io/commands/zrangebylex">Redis Documentation: ZRANGEBYLEX</a>
 	 */
-	Mono<Set<V>> rangeByLex(K key, Range<String> range);
+	Flux<V> rangeByLex(K key, Range<String> range);
 
 	/**
 	 * Get all elements {@literal n} elements, where {@literal n = } {@link Limit#getCount()}, starting at
@@ -338,7 +340,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrangebylex">Redis Documentation: ZRANGEBYLEX</a>
 	 */
-	Mono<Set<V>> rangeByLex(K key, Range<String> range, Limit limit);
+	Flux<V> rangeByLex(K key, Range<String> range, Limit limit);
 
 	/**
 	 * Get all elements with reverse lexicographical ordering from {@literal ZSET} at {@code key} with a value between
@@ -348,7 +350,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @param range must not be {@literal null}.
 	 * @see <a href="http://redis.io/commands/zrevrangebylex">Redis Documentation: ZREVRANGEBYLEX</a>
 	 */
-	Mono<Set<V>> reverseRangeByLex(K key, Range<String> range);
+	Flux<V> reverseRangeByLex(K key, Range<String> range);
 
 	/**
 	 * Get all elements {@literal n} elements, where {@literal n = } {@link Limit#getCount()}, starting at
@@ -361,7 +363,7 @@ public interface ReactiveZSetOperations<K, V> {
 	 * @return
 	 * @see <a href="http://redis.io/commands/zrevrangebylex">Redis Documentation: ZREVRANGEBYLEX</a>
 	 */
-	Mono<Set<V>> reverseRangeByLex(K key, Range<String> range, Limit limit);
+	Flux<V> reverseRangeByLex(K key, Range<String> range, Limit limit);
 
 	/**
 	 * Removes the given {@literal key}.