Reader should provide target Object type to allow easy customization.

This commit makes sure to extract and pass on the target type, otherwise it would be Object.class all the time.

Original Pull Request: #2332

src/main/java/org/springframework/data/redis/serializer/GenericJackson2JsonRedisSerializer.java

@@ -16,8 +16,11 @@
 package org.springframework.data.redis.serializer;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.function.Supplier;
 
 import org.springframework.cache.support.NullValue;
+import org.springframework.data.util.Lazy;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -25,14 +28,19 @@ import org.springframework.util.StringUtils;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.annotation.JsonTypeInfo.As;
 import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectMapper.DefaultTyping;
+import com.fasterxml.jackson.databind.ObjectReader;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.jsontype.PolymorphicTypeValidator;
 import com.fasterxml.jackson.databind.jsontype.TypeSerializer;
 import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.node.TextNode;
 import com.fasterxml.jackson.databind.ser.SerializerFactory;
 import com.fasterxml.jackson.databind.ser.std.StdSerializer;
+import com.fasterxml.jackson.databind.type.TypeFactory;
 
 /**
  * Generic Jackson 2-based {@link RedisSerializer} that maps {@link Object objects} to JSON using dynamic typing.
@@ -47,12 +55,37 @@ import com.fasterxml.jackson.databind.ser.std.StdSerializer;
  */
 public class GenericJackson2JsonRedisSerializer implements RedisSerializer<Object> {
 
-	private final ObjectMapper mapper;
+	private ObjectMapper mapper;
 
 	private final JacksonObjectReader reader;
 
 	private final JacksonObjectWriter writer;
 
+	private boolean internalReader = false;
+
+	private final TypeResolver typeResolver;
+
+	private Lazy<Boolean> defaultTypingEnabled = Lazy
+			.of(() -> mapper.getSerializationConfig().getDefaultTyper(null) != null);
+
+	private Lazy<String> typeHintPropertyName;
+
+	{
+		typeHintPropertyName = Lazy.of(() -> {
+			if (defaultTypingEnabled.get()) {
+				return null;
+			}
+
+			return mapper.getDeserializationConfig().getDefaultTyper(null)
+					.buildTypeDeserializer(mapper.getDeserializationConfig(), mapper.getTypeFactory().constructType(Object.class),
+							Collections.emptyList())
+					.getPropertyName();
+
+		}).or("@class");
+
+		typeResolver = new TypeResolver(Lazy.of(() -> mapper.getTypeFactory()), typeHintPropertyName);
+	}
+
 	/**
 	 * Creates {@link GenericJackson2JsonRedisSerializer} and configures {@link ObjectMapper} for default typing.
 	 */
@@ -71,6 +104,7 @@ public class GenericJackson2JsonRedisSerializer implements RedisSerializer<Objec
 	 */
 	public GenericJackson2JsonRedisSerializer(@Nullable String classPropertyTypeName) {
 		this(classPropertyTypeName, JacksonObjectReader.create(), JacksonObjectWriter.create());
+		this.internalReader = true;
 	}
 
 	/**
@@ -100,6 +134,10 @@ public class GenericJackson2JsonRedisSerializer implements RedisSerializer<Objec
 		} else {
 			mapper.activateDefaultTyping(mapper.getPolymorphicTypeValidator(), DefaultTyping.EVERYTHING, As.PROPERTY);
 		}
+
+		if (classPropertyTypeName != null) {
+			typeHintPropertyName = Lazy.of(classPropertyTypeName);
+		}
 	}
 
 	/**
@@ -111,6 +149,7 @@ public class GenericJackson2JsonRedisSerializer implements RedisSerializer<Objec
 	 */
 	public GenericJackson2JsonRedisSerializer(ObjectMapper mapper) {
 		this(mapper, JacksonObjectReader.create(), JacksonObjectWriter.create());
+		this.internalReader = true;
 	}
 
 	/**
@@ -188,12 +227,54 @@ public class GenericJackson2JsonRedisSerializer implements RedisSerializer<Objec
 		}
 
 		try {
-			return (T) reader.read(mapper, source, mapper.getTypeFactory().constructType(type));
+			return (T) reader.read(mapper, source, resolveType(source, type));
 		} catch (Exception ex) {
 			throw new SerializationException("Could not read JSON: " + ex.getMessage(), ex);
 		}
 	}
 
+	protected JavaType resolveType(byte[] source, Class<?> type) {
+
+		if (internalReader || !type.equals(Object.class) || !defaultTypingEnabled.get()) {
+			return typeResolver.constructType(type);
+		}
+
+		return typeResolver.resolveType(source, type);
+	}
+
+	private static class TypeResolver {
+
+		private final ObjectReader objectReader = new ObjectMapper().reader();
+
+		private final Supplier<TypeFactory> typeFactory;
+		private Supplier<String> hintName;
+
+		public TypeResolver(Supplier<TypeFactory> typeFactory, Supplier<String> hintName) {
+
+			this.typeFactory = typeFactory;
+			this.hintName = hintName;
+		}
+
+		protected JavaType constructType(Class<?> type) {
+			return typeFactory.get().constructType(type);
+		}
+
+		protected JavaType resolveType(byte[] source, Class<?> type) {
+
+			try {
+				TextNode typeName = (TextNode) objectReader.readValue(source, JsonNode.class).get(hintName.get());
+				if (typeName != null) {
+					return typeFactory.get().constructFromCanonical(typeName.textValue());
+				}
+			} catch (IOException e) {
+				// TODO: logging?
+			}
+
+			return constructType(type);
+		}
+
+	}
+
 	/**
 	 * {@link StdSerializer} adding class information required by default typing. This allows de-/serialization of
 	 * {@link NullValue}.