DATAREDIS-1223 - Upgrade to Lettuce 6.0 GA.

Adopt HashCommands to changed HGETALL method signature returning Flux<KeyValue> instead of Mono<Map>.

Author: mp911de

pom.xml

@@ -22,7 +22,7 @@
 		<beanutils>1.9.2</beanutils>
 		<xstream>1.4.12</xstream>
 		<pool>2.7.0</pool>
-		<lettuce>6.0.0.RC2</lettuce>
+		<lettuce>6.0.0.RELEASE</lettuce>
 		<jedis>3.3.0</jedis>
 		<multithreadedtc>1.01</multithreadedtc>
 		<netty>4.1.51.Final</netty>

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

@@ -28,6 +28,7 @@
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.redis.connection.ReactiveHashCommands;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.BooleanResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
@@ -209,9 +210,9 @@ public Flux<CommandResponse<KeyCommand, Flux<Map.Entry<ByteBuffer, ByteBuffer>>>
 
 			Assert.notNull(command.getKey(), "Key must not be null!");
 
-			Mono<Map<ByteBuffer, ByteBuffer>> result = cmd.hgetall(command.getKey());
+			Flux<KeyValue<ByteBuffer, ByteBuffer>> result = cmd.hgetall(command.getKey());
 
-			return Mono.just(new CommandResponse<>(command, result.flatMapMany(v -> Flux.fromStream(v.entrySet().stream()))));
+			return Mono.just(new CommandResponse<>(command, result.map(LettuceReactiveHashCommands::toEntry)));
 		}));
 	}
 
@@ -268,4 +269,25 @@ public Flux<NumericResponse<HStrLenCommand, Long>> hStrLen(Publisher<HStrLenComm
 			return cmd.hstrlen(command.getKey(), command.getField()).map(value -> new NumericResponse<>(command, value));
 		}));
 	}
+
+	private static Map.Entry<ByteBuffer, ByteBuffer> toEntry(KeyValue<ByteBuffer, ByteBuffer> kv) {
+
+		return new Entry<ByteBuffer, ByteBuffer>() {
+
+			@Override
+			public ByteBuffer getKey() {
+				return kv.getKey();
+			}
+
+			@Override
+			public ByteBuffer getValue() {
+				return kv.getValue();
+			}
+
+			@Override
+			public ByteBuffer setValue(ByteBuffer value) {
+				throw new UnsupportedOperationException("Cannot set value for entry");
+			}
+		};
+	}
 }