Retain target type hint when deserializing Stream records

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-redis</artifactId>
-	<version>2.7.0-SNAPSHOT</version>
+	<version>2.7.0-GH-2198-SNAPSHOT</version>
 
 	<name>Spring Data Redis</name>
 

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

@@ -17,9 +17,9 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
@@ -29,6 +29,7 @@
 import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.data.redis.core.convert.RedisCustomConversions;
 import org.springframework.data.redis.hash.HashMapper;
+import org.springframework.data.redis.hash.HashObjectReader;
 import org.springframework.data.redis.hash.ObjectHashMapper;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -72,25 +73,7 @@ class StreamObjectMapper {
 		this.mapper = (HashMapper) mapper;
 
 		if (mapper instanceof ObjectHashMapper) {
-
-			ObjectHashMapper ohm = (ObjectHashMapper) mapper;
-			this.objectHashMapper = new HashMapper<Object, Object, Object>() {
-
-				@Override
-				public Map<Object, Object> toHash(Object object) {
-					return (Map) ohm.toHash(object);
-				}
-
-				@Override
-				public Object fromHash(Map<Object, Object> hash) {
-
-					Map<byte[], byte[]> map = hash.entrySet().stream()
-							.collect(Collectors.toMap(e -> conversionService.convert(e.getKey(), byte[].class),
-									e -> conversionService.convert(e.getValue(), byte[].class)));
-
-					return ohm.fromHash(map);
-				}
-			};
+			this.objectHashMapper = new BinaryObjectHashMapperAdapter((ObjectHashMapper) mapper);
 		} else {
 			this.objectHashMapper = null;
 		}
@@ -174,9 +157,27 @@ static <K, V, HK, HV> List<ObjectRecord<K, V>> toObjectRecords(@Nullable List<Ma
 		return transformed;
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	final <V, HK, HV> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType) {
-		return (HashMapper) doGetHashMapper(conversionService, targetType);
+
+		HashMapper hashMapper = doGetHashMapper(conversionService, targetType);
+
+		if (hashMapper instanceof HashObjectReader) {
+
+			return new HashMapper<V, HK, HV>() {
+				@Override
+				public Map<HK, HV> toHash(V object) {
+					return hashMapper.toHash(object);
+				}
+
+				@Override
+				public V fromHash(Map<HK, HV> hash) {
+					return ((HashObjectReader<HK, HV>) hashMapper).fromHash(targetType, hash);
+				}
+			};
+		}
+
+		return hashMapper;
 	}
 
 	/**
@@ -208,4 +209,46 @@ boolean isSimpleType(Class<?> targetType) {
 	ConversionService getConversionService() {
 		return conversionService;
 	}
+
+	private static class BinaryObjectHashMapperAdapter
+			implements HashMapper<Object, Object, Object>, HashObjectReader<Object, Object> {
+
+		private final ObjectHashMapper ohm;
+
+		public BinaryObjectHashMapperAdapter(ObjectHashMapper ohm) {
+			this.ohm = ohm;
+		}
+
+		@Override
+		@SuppressWarnings({ "unchecked", "rawtypes" })
+		public Map<Object, Object> toHash(Object object) {
+			return (Map) ohm.toHash(object);
+		}
+
+		@Override
+		public Object fromHash(Map<Object, Object> hash) {
+			return ohm.fromHash(toMap(hash));
+		}
+
+		@Override
+		public <R> R fromHash(Class<R> type, Map<Object, Object> hash) {
+			return ohm.fromHash(type, toMap(hash));
+		}
+
+		private static Map<byte[], byte[]> toMap(Map<Object, Object> hash) {
+
+			Map<byte[], byte[]> target = new LinkedHashMap<>(hash.size());
+
+			for (Map.Entry<Object, Object> entry : hash.entrySet()) {
+				target.put(toBytes(entry.getKey()), toBytes(entry.getValue()));
+			}
+
+			return target;
+		}
+
+		@Nullable
+		private static byte[] toBytes(Object value) {
+			return value instanceof byte[] ? (byte[]) value : conversionService.convert(value, byte[].class);
+		}
+	}
 }

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

@@ -21,14 +21,16 @@
 
 import org.apache.commons.beanutils.BeanUtils;
 
+import org.springframework.util.Assert;
+
 /**
  * HashMapper based on Apache Commons BeanUtils project. Does NOT supports nested properties.
  *
  * @author Costin Leau
  * @author Christoph Strobl
  * @author Mark Paluch
  */
-public class BeanUtilsHashMapper<T> implements HashMapper<T, String, String> {
+public class BeanUtilsHashMapper<T> implements HashMapper<T, String, String>, HashObjectReader<String, String> {
 
 	private final Class<T> type;
 
@@ -47,8 +49,20 @@ public BeanUtilsHashMapper(Class<T> type) {
 	 */
 	@Override
 	public T fromHash(Map<String, String> hash) {
+		return fromHash(type, hash);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.hash.HashMapper#fromHash(java.lang.Class, java.util.Map)
+	 */
+	@Override
+	public <R> R fromHash(Class<R> type, Map<String, String> hash) {
+
+		Assert.notNull(type, "Type must not be null");
+		Assert.notNull(hash, "Hash must not be null");
 
-		T instance = org.springframework.beans.BeanUtils.instantiateClass(type);
+		R instance = org.springframework.beans.BeanUtils.instantiateClass(type);
 
 		try {
 

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

@@ -26,6 +26,7 @@
  * @param <V> Redis Hash value type
  * @author Costin Leau
  * @author Mark Paluch
+ * @see HashObjectReader
  */
 public interface HashMapper<T, K, V> {
 

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.hash;
+
+import java.util.Map;
+
+/**
+ * Core mapping contract to materialize an object using particular Java class from a Redis Hash.
+ *
+ * @param <K> Redis Hash field type
+ * @param <V> Redis Hash value type
+ * @author Mark Paluch
+ * @since 2.7
+ * @see HashMapper
+ */
+public interface HashObjectReader<K, V> {
+
+	/**
+	 * Materialize an object of the {@link Class type} from a {@code hash}.
+	 *
+	 * @param hash must not be {@literal null}.
+	 * @return the materialized object from the given {@code hash}.
+	 */
+	<R> R fromHash(Class<R> type, Map<K, V> hash);
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,10 +48,13 @@
 import com.fasterxml.jackson.databind.JsonDeserializer;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectMapper.DefaultTyping;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.deser.BeanDeserializerFactory;
+import com.fasterxml.jackson.databind.deser.Deserializers;
 import com.fasterxml.jackson.databind.deser.std.UntypedObjectDeserializer;
 import com.fasterxml.jackson.databind.jsontype.PolymorphicTypeValidator;
 import com.fasterxml.jackson.databind.jsontype.TypeDeserializer;
@@ -148,7 +151,7 @@
  * @author Mark Paluch
  * @since 1.8
  */
-public class Jackson2HashMapper implements HashMapper<Object, String, Object> {
+public class Jackson2HashMapper implements HashMapper<Object, String, Object>, HashObjectReader<String, Object> {
 
 	private final HashMapperModule HASH_MAPPER_MODULE = new HashMapperModule();
 
@@ -168,32 +171,70 @@ public Jackson2HashMapper(boolean flatten) {
 			@Override
 			protected TypeResolverBuilder<?> _constructDefaultTypeResolverBuilder(DefaultTyping applicability,
 					PolymorphicTypeValidator ptv) {
+
 				return new DefaultTypeResolverBuilder(applicability, ptv) {
+
+					Map<Class, Boolean> serializerPresentCache = new HashMap<>();
+
 					public boolean useForType(JavaType t) {
 
 						if (t.isPrimitive()) {
 							return false;
 						}
 
 						if (EVERYTHING.equals(_appliesFor)) {
+
+							while (t.isArrayType()) {
+								t = t.getContentType();
+							}
+							while (t.isReferenceType()) {
+								t = t.getReferencedType();
+							}
+
+							/* 
+							 * check for registered serializers and make uses of those.
+							 */
+							if (serializerPresentCache.computeIfAbsent(t.getRawClass(), this::hasConfiguredSerializer)) {
+								return false;
+							}
+
 							return !TreeNode.class.isAssignableFrom(t.getRawClass());
 						}
 
 						return super.useForType(t);
 					}
+
+					private Boolean hasConfiguredSerializer(Class key) {
+
+						if (!(_deserializationContext.getFactory() instanceof BeanDeserializerFactory)) {
+							return false;
+						}
+
+						Iterator<Deserializers> deserializers = ((BeanDeserializerFactory) _deserializationContext.getFactory())
+								.getFactoryConfig().deserializers().iterator();
+						while (deserializers.hasNext()) {
+							Deserializers next = deserializers.next();
+							if (next.hasDeserializerFor(_deserializationConfig, key)) {
+								return true;
+							}
+						}
+						return false;
+					}
 				};
 			}
 		}.findAndRegisterModules(), flatten);
 
 		typingMapper.activateDefaultTyping(typingMapper.getPolymorphicTypeValidator(), DefaultTyping.EVERYTHING,
 				As.PROPERTY);
 		typingMapper.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
+		typingMapper.configure(MapperFeature.USE_BASE_TYPE_AS_DEFAULT_IMPL, true);
 
 		// Prevent splitting time types into arrays. E
 		typingMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
 		typingMapper.setSerializationInclusion(Include.NON_NULL);
 		typingMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 		typingMapper.registerModule(HASH_MAPPER_MODULE);
+		typingMapper.findAndRegisterModules();
 	}
 
 	/**
@@ -209,7 +250,7 @@ public Jackson2HashMapper(ObjectMapper mapper, boolean flatten) {
 		this.flatten = flatten;
 
 		this.untypedMapper = new ObjectMapper();
-		untypedMapper.findAndRegisterModules();
+		this.untypedMapper.findAndRegisterModules();
 		this.untypedMapper.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
 		this.untypedMapper.setSerializationInclusion(Include.NON_NULL);
 	}
@@ -232,16 +273,25 @@ public Map<String, Object> toHash(Object source) {
 	 */
 	@Override
 	public Object fromHash(Map<String, Object> hash) {
+		return fromHash(Object.class, hash);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.hash.HashMapper#fromHash(Class, java.util.Map)
+	 */
+	@Override
+	public <R> R fromHash(Class<R> type, Map<String, Object> hash) {
 
 		try {
 
 			if (flatten) {
 
-				return typingMapper.reader().forType(Object.class)
+				return typingMapper.reader().forType(type)
 						.readValue(untypedMapper.writeValueAsBytes(doUnflatten(hash)));
 			}
 
-			return typingMapper.treeToValue(untypedMapper.valueToTree(hash), Object.class);
+			return typingMapper.treeToValue(untypedMapper.valueToTree(hash), type);
 
 		} catch (IOException e) {
 			throw new MappingException(e.getMessage(), e);

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

@@ -69,7 +69,7 @@
  * @author Mark Paluch
  * @since 1.8
  */
-public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]> {
+public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]>, HashObjectReader<byte[], byte[]> {
 
 	@Nullable private volatile static ObjectHashMapper sharedInstance;
 
@@ -169,12 +169,20 @@ public Map<byte[], byte[]> toHash(Object source) {
 	 */
 	@Override
 	public Object fromHash(Map<byte[], byte[]> hash) {
+		return fromHash(Object.class, hash);
+	}
 
-		if (hash == null || hash.isEmpty()) {
-			return null;
-		}
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.hash.HashMapper#fromHash(java.lang.Class, java.util.Map)
+	 */
+	@Override
+	public <R> R fromHash(Class<R> type, Map<byte[], byte[]> hash) {
+
+		Assert.notNull(type, "Type must not be null");
+		Assert.notNull(hash, "Hash must not be null");
 
-		return converter.read(Object.class, new RedisData(hash));
+		return converter.read(type, new RedisData(hash));
 	}
 
 	/**