DATAREDIS-1233 - Introduce and use Junit5 based ParameterizedRedisTest.

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

@@ -87,7 +87,13 @@ class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommands imple
 			}
 
 			return sUnion(Mono.just(SUnionCommand.keys(command.getKeys()))).next().flatMap(values -> {
-				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().toStream().toArray(ByteBuffer[]::new));
+
+				Mono<Long> result = values.getOutput().collectList().flatMap(it -> {
+
+					ByteBuffer[] members = it.toArray(new ByteBuffer[0]);
+					return cmd.sadd(command.getKey(), members);
+				});
+
 				return result.map(value -> new NumericResponse<>(command, value));
 			});
 		}));
@@ -147,7 +153,13 @@ class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommands imple
 			}
 
 			return sInter(Mono.just(SInterCommand.keys(command.getKeys()))).next().flatMap(values -> {
-				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().toStream().toArray(ByteBuffer[]::new));
+
+				Mono<Long> result = values.getOutput().collectList().flatMap(it -> {
+
+					ByteBuffer[] members = it.toArray(new ByteBuffer[0]);
+					return cmd.sadd(command.getKey(), members);
+				});
+
 				return result.map(value -> new NumericResponse<>(command, value));
 			});
 		}));
@@ -209,7 +221,13 @@ class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommands imple
 			}
 
 			return sDiff(Mono.just(SDiffCommand.keys(command.getKeys()))).next().flatMap(values -> {
-				Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().toStream().toArray(ByteBuffer[]::new));
+
+				Mono<Long> result = values.getOutput().collectList().flatMap(it -> {
+
+					ByteBuffer[] members = it.toArray(new ByteBuffer[0]);
+					return cmd.sadd(command.getKey(), members);
+				});
+
 				return result.map(value -> new NumericResponse<>(command, value));
 			});
 		}));

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

@@ -492,7 +492,11 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 
 		if (StringUtils.hasText(node.getId())) {
 			return getConnection().cast(StatefulRedisClusterConnection.class)
-					.map(it -> it.getConnection(node.getId()).reactive());
+					.flatMap(it -> {
+						StatefulRedisClusterConnection<ByteBuffer, ByteBuffer> connection = it;
+						return Mono.fromCompletionStage(connection.getConnectionAsync(node.getId()))
+								.map(StatefulRedisConnection::reactive);
+					});
 		}
 
 		return getConnection().flatMap(it -> Mono.fromCompletionStage(it.getConnectionAsync(node.getHost(), node.getPort()))

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

@@ -660,7 +660,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	@SuppressWarnings("unchecked")
 	public <HK, HV> ReactiveStreamOperations<K, HK, HV> opsForStream(
 			RedisSerializationContext<K, ?> serializationContext) {
-		return opsForStream(serializationContext, (HashMapper) new ObjectHashMapper());
+		return opsForStream(serializationContext, (HashMapper) ObjectHashMapper.getSharedInstance());
 	}
 
 	protected <HK, HV> ReactiveStreamOperations<K, HK, HV> opsForStream(

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

@@ -106,7 +106,8 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 	private final ValueOperations<K, V> valueOps = new DefaultValueOperations<>(this);
 	private final ListOperations<K, V> listOps = new DefaultListOperations<>(this);
 	private final SetOperations<K, V> setOps = new DefaultSetOperations<>(this);
-	private final StreamOperations<K, ?, ?> streamOps = new DefaultStreamOperations<>(this, new ObjectHashMapper());
+	private final StreamOperations<K, ?, ?> streamOps = new DefaultStreamOperations<>(this,
+			ObjectHashMapper.getSharedInstance());
 	private final ZSetOperations<K, V> zSetOps = new DefaultZSetOperations<>(this);
 	private final GeoOperations<K, V> geoOps = new DefaultGeoOperations<>(this);
 	private final HyperLogLogOperations<K, V> hllOps = new DefaultHyperLogLogOperations<>(this);

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

@@ -39,7 +39,7 @@ import org.springframework.util.Assert;
  * This utility can use generic a {@link HashMapper} or adapt specifically to {@link ObjectHashMapper}'s requirement to
  * convert incoming data into byte arrays. This class can be subclassed to override template methods for specific object
  * mapping strategies.
- * 
+ *
  * @author Mark Paluch
  * @since 2.2
  * @see ObjectHashMapper
@@ -47,14 +47,21 @@ import org.springframework.util.Assert;
  */
 class StreamObjectMapper {
 
-	private final DefaultConversionService conversionService = new DefaultConversionService();
-	private final RedisCustomConversions customConversions = new RedisCustomConversions();
+	private final static RedisCustomConversions customConversions = new RedisCustomConversions();
+	private final static ConversionService conversionService;
+
 	private final HashMapper<Object, Object, Object> mapper;
 	private final @Nullable HashMapper<Object, Object, Object> objectHashMapper;
 
+	static {
+		DefaultConversionService cs = new DefaultConversionService();
+		customConversions.registerConvertersIn(cs);
+		conversionService = cs;
+	}
+
 	/**
 	 * Creates a new {@link StreamObjectMapper}.
-	 * 
+	 *
 	 * @param mapper the configured {@link HashMapper}.
 	 */
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -62,7 +69,6 @@ class StreamObjectMapper {
 
 		Assert.notNull(mapper, "HashMapper must not be null");
 
-		this.customConversions.registerConvertersIn(conversionService);
 		this.mapper = (HashMapper) mapper;
 
 		if (mapper instanceof ObjectHashMapper) {
@@ -92,7 +98,7 @@ class StreamObjectMapper {
 
 	/**
 	 * Convert the given {@link Record} into a {@link MapRecord}.
-	 * 
+	 *
 	 * @param provider provider for {@link HashMapper} to apply mapping for {@link ObjectRecord}.
 	 * @param source the source value.
 	 * @return the converted {@link MapRecord}.
@@ -121,7 +127,7 @@ class StreamObjectMapper {
 
 	/**
 	 * Convert the given {@link Record} into an {@link ObjectRecord}.
-	 * 
+	 *
 	 * @param provider provider for {@link HashMapper} to apply mapping for {@link ObjectRecord}.
 	 * @param source the source value.
 	 * @param targetType the desired target type.
@@ -135,7 +141,7 @@ class StreamObjectMapper {
 	/**
 	 * Map a {@link List} of {@link MapRecord}s to a {@link List} of {@link ObjectRecord}. Optimizes for empty,
 	 * single-element and multi-element list transformation.l
-	 * 
+	 *
 	 * @param records the {@link MapRecord} that should be mapped.
 	 * @param hashMapperProvider the provider to obtain the actual {@link HashMapper} from. Must not be {@literal null}.
 	 * @param targetType the requested {@link Class target type}.
@@ -175,7 +181,7 @@ class StreamObjectMapper {
 
 	/**
 	 * Returns the actual {@link HashMapper}. Can be overridden by subclasses.
-	 * 
+	 *
 	 * @param conversionService the used {@link ConversionService}.
 	 * @param targetType the target type.
 	 * @return obtain the {@link HashMapper} for a certain type.
@@ -200,6 +206,6 @@ class StreamObjectMapper {
 	 * @return used {@link ConversionService}.
 	 */
 	ConversionService getConversionService() {
-		return this.conversionService;
+		return conversionService;
 	}
 }

src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java

@@ -71,6 +71,8 @@ import org.springframework.util.Assert;
  */
 public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]> {
 
+	@Nullable private volatile static ObjectHashMapper sharedInstance;
+
 	private final RedisConverter converter;
 
 	/**
@@ -120,6 +122,31 @@ public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]> {
 		converter = mappingConverter;
 	}
 
+	/**
+	 * Return a shared default {@value ObjectHashMapper} instance, lazily building it once needed.
+	 * <p>
+	 * <b>NOTE:</b> We highly recommend constructing individual {@link ObjectHashMapper} instances for customization
+	 * purposes. This accessor is only meant as a fallback for code paths which need simple type coercion but cannot
+	 * access a longer-lived {@link ObjectHashMapper} instance any other way.
+	 *
+	 * @return the shared {@link ObjectHashMapper} instance (never {@literal null}).
+	 * @since 2.4
+	 */
+	public static ObjectHashMapper getSharedInstance() {
+
+		ObjectHashMapper cs = sharedInstance;
+		if (cs == null) {
+			synchronized (ObjectHashMapper.class) {
+				cs = sharedInstance;
+				if (cs == null) {
+					cs = new ObjectHashMapper();
+					sharedInstance = cs;
+				}
+			}
+		}
+		return cs;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.hash.HashMapper#toHash(java.lang.Object)

src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java

@@ -749,7 +749,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 
 				hashKeySerializer(RedisSerializer.byteArray());
 				hashValueSerializer(RedisSerializer.byteArray());
-				return (StreamMessageListenerContainerOptionsBuilder) objectMapper(new ObjectHashMapper());
+				return (StreamMessageListenerContainerOptionsBuilder) objectMapper(ObjectHashMapper.getSharedInstance());
 			}
 
 			return (StreamMessageListenerContainerOptionsBuilder) this;

src/main/java/org/springframework/data/redis/stream/StreamReceiver.java

@@ -423,7 +423,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 
 				hashKeySerializer(SerializationPair.raw());
 				hashValueSerializer(SerializationPair.raw());
-				return (StreamReceiverOptionsBuilder) objectMapper(new ObjectHashMapper());
+				return (StreamReceiverOptionsBuilder) objectMapper(ObjectHashMapper.getSharedInstance());
 			}
 
 			return (StreamReceiverOptionsBuilder) this;