Json Type Determination Using Methods

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java

@@ -20,6 +20,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
+import java.util.function.BiFunction;
 
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.header.Headers;
@@ -92,20 +93,32 @@ public class JsonDeserializer<T> implements Deserializer<T> {
 	 */
 	public static final String USE_TYPE_INFO_HEADERS = "spring.json.use.type.headers";
 
+	/**
+	 * A method name to determine the {@link JavaType} to deserialize the key to.
+	 */
+	public static final String KEY_TYPE_METHOD = "spring.json.key.type.method";
+
+	/**
+	 * A method name to determine the {@link JavaType} to deserialize the key to.
+	 */
+	public static final String VALUE_TYPE_METHOD = "spring.json.value.type.method";
+
 	protected final ObjectMapper objectMapper; // NOSONAR
 
 	protected JavaType targetType; // NOSONAR
 
 	protected Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); // NOSONAR
 
-	private volatile ObjectReader reader;
+	private ObjectReader reader;
 
 	private boolean typeMapperExplicitlySet = false;
 
 	private boolean removeTypeHeaders = true;
 
 	private boolean useTypeHeaders = true;
 
+	private BiFunction<byte[], Headers, JavaType> typeFunction;
+
 	/**
 	 * Construct an instance with a default {@link ObjectMapper}.
 	 */
@@ -242,19 +255,6 @@ public JsonDeserializer(@Nullable JavaType targetType, ObjectMapper objectMapper
 		initialize(targetType, useHeadersIfPresent);
 	}
 
-	private void initialize(@Nullable JavaType type, boolean useHeadersIfPresent) {
-		this.targetType = type;
-		Assert.isTrue(this.targetType != null || useHeadersIfPresent,
-				"'targetType' cannot be null if 'useHeadersIfPresent' is false");
-
-		if (this.targetType != null) {
-			this.reader = this.objectMapper.readerFor(this.targetType);
-		}
-
-		addTargetPackageToTrusted();
-		this.typeMapper.setTypePrecedence(useHeadersIfPresent ? TypePrecedence.TYPE_ID : TypePrecedence.INFERRED);
-	}
-
 	public Jackson2JavaTypeMapper getTypeMapper() {
 		return this.typeMapper;
 	}
@@ -307,6 +307,16 @@ public void setUseTypeHeaders(boolean useTypeHeaders) {
 		}
 	}
 
+	/**
+	 * Set a {@link BiFunction} that receives the data to be deserialized and the headers
+	 * and returns a JavaType.
+	 * @param typeFunction the function.
+	 * @since 2.5
+	 */
+	public void setTypeFunction(BiFunction<byte[], Headers, JavaType> typeFunction) {
+		this.typeFunction = typeFunction;
+	}
+
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {
 		setUseTypeMapperForKey(isKey);
@@ -325,6 +335,17 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 		if (configs.containsKey(REMOVE_TYPE_INFO_HEADERS)) {
 			this.removeTypeHeaders = Boolean.parseBoolean(configs.get(REMOVE_TYPE_INFO_HEADERS).toString());
 		}
+		if (isKey && configs.containsKey(KEY_TYPE_METHOD)) {
+			setUpTypeFuntion((String) configs.get(KEY_TYPE_METHOD));
+		}
+		else if (!isKey && configs.containsKey(VALUE_TYPE_METHOD)) {
+			setUpTypeFuntion((String) configs.get(VALUE_TYPE_METHOD));
+		}
+	}
+
+	private void setUpTypeFuntion(String method) {
+		this.typeFunction = SerializationUtils.propertyToMethodInvokingFunction(method, byte[].class,
+				getClass().getClassLoader());
 	}
 
 	private void setUpTypePrecedence(Map<String, ?> configs) {
@@ -355,6 +376,19 @@ else if (!isKey && configs.containsKey(VALUE_DEFAULT_TYPE)) {
 		}
 	}
 
+	private void initialize(@Nullable JavaType type, boolean useHeadersIfPresent) {
+		this.targetType = type;
+		Assert.isTrue(this.targetType != null || useHeadersIfPresent,
+				"'targetType' cannot be null if 'useHeadersIfPresent' is false");
+
+		if (this.targetType != null) {
+			this.reader = this.objectMapper.readerFor(this.targetType);
+		}
+
+		addTargetPackageToTrusted();
+		this.typeMapper.setTypePrecedence(useHeadersIfPresent ? TypePrecedence.TYPE_ID : TypePrecedence.INFERRED);
+	}
+
 	private JavaType setupTargetType(Map<String, ?> configs, String key) throws ClassNotFoundException, LinkageError {
 		if (configs.get(key) instanceof Class) {
 			return TypeFactory.defaultInstance().constructType((Class<?>) configs.get(key));
@@ -401,11 +435,15 @@ public T deserialize(String topic, Headers headers, byte[] data) {
 			return null;
 		}
 		ObjectReader deserReader = null;
-		if (this.typeMapper.getTypePrecedence().equals(TypePrecedence.TYPE_ID)) {
-			JavaType javaType = this.typeMapper.toJavaType(headers);
-			if (javaType != null) {
-				deserReader = this.objectMapper.readerFor(javaType);
-			}
+		JavaType javaType = null;
+		if (this.typeFunction != null) {
+			javaType = this.typeFunction.apply(data, headers);
+		}
+		if (javaType == null && this.typeMapper.getTypePrecedence().equals(TypePrecedence.TYPE_ID)) {
+			javaType = this.typeMapper.toJavaType(headers);
+		}
+		if (javaType != null) {
+			deserReader = this.objectMapper.readerFor(javaType);
 		}
 		if (this.removeTypeHeaders) {
 			this.typeMapper.removeHeaders(headers);
@@ -428,9 +466,16 @@ public T deserialize(String topic, @Nullable byte[] data) {
 		if (data == null) {
 			return null;
 		}
-		Assert.state(this.reader != null, "No headers available and no default type provided");
+		ObjectReader localReader = this.reader;
+		if (this.typeFunction != null) {
+			JavaType javaType = this.typeFunction.apply(data, null);
+			if (javaType != null) {
+				localReader = this.objectMapper.readerFor(javaType);
+			}
+		}
+		Assert.state(localReader != null, "No headers available and no default type provided");
 		try {
-			return this.reader.readValue(data);
+			return localReader.readValue(data);
 		}
 		catch (IOException e) {
 			throw new SerializationException("Can't deserialize data [" + Arrays.toString(data) +
@@ -490,4 +535,28 @@ public JsonDeserializer<T> typeMapper(Jackson2JavaTypeMapper mapper) {
 		return this;
 	}
 
+	/**
+	 * Add trusted packages to the default type mapper.
+	 * @param packages the packages.
+	 * @return the deserializer.
+	 * @since 2,5
+	 */
+	public JsonDeserializer<T> trustedPackages(String... packages) {
+		Assert.isTrue(!this.typeMapperExplicitlySet, "When using a custom type mapper, set the trusted packages there");
+		this.typeMapper.addTrustedPackages(packages);
+		return this;
+	}
+
+	/**
+	 * Set a {@link BiFunction} that receives the data to be deserialized and the headers
+	 * and returns a JavaType.
+	 * @param typeFunction the function.
+	 * @return the deserializer.
+	 * @since 2.5
+	 */
+	public JsonDeserializer<T> typeFunction(BiFunction<byte[], Headers, JavaType> typeFunction) {
+		setTypeFunction(typeFunction);
+		return this;
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ParseStringDeserializer.java

@@ -16,8 +16,6 @@
 
 package org.springframework.kafka.support.serializer;
 
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
@@ -28,7 +26,6 @@
 import org.apache.kafka.common.serialization.Deserializer;
 
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
 
 /**
  * Generic {@link org.apache.kafka.common.serialization.Deserializer Deserializer} for deserialization of entity from
@@ -56,6 +53,7 @@ public class ParseStringDeserializer<T> implements Deserializer<T> {
 		throw new IllegalStateException("A parser must be provided either via a constructor or consumer properties");
 	};
 
+	@SuppressWarnings("unchecked")
 	private BiFunction<String, Headers, T> parser = (BiFunction<String, Headers, T>) NO_PARSER;
 
 	private Charset charset = StandardCharsets.UTF_8;
@@ -84,64 +82,16 @@ public ParseStringDeserializer(BiFunction<String, Headers, T> parser) {
 		this.parser = parser;
 	}
 
-	@SuppressWarnings("unchecked")
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {
 		if (NO_PARSER.equals(this.parser)) {
 			String parserMethod = (String) configs.get(isKey ? KEY_PARSER : VALUE_PARSER);
 			Assert.state(parserMethod != null,
 					"A parser must be provided either via a constructor or consumer properties");
-			int lastDotPosn = parserMethod.lastIndexOf(".");
-			Assert.state(lastDotPosn > 1,
-					"the parser method needs to be a class name followed by the method name, separated by '.'");
-			Class<?> clazz;
-			try {
-				clazz = ClassUtils.forName(parserMethod.substring(0, lastDotPosn),
-						getClass().getClassLoader());
-			}
-			catch (ClassNotFoundException | LinkageError e) {
-				throw new IllegalStateException(e);
-			}
-			parserMethod = parserMethod.substring(lastDotPosn + 1);
-			Method method;
-			try {
-				method = clazz.getDeclaredMethod(parserMethod, String.class, Headers.class);
-			}
-			catch (@SuppressWarnings("unused") NoSuchMethodException e) {
-				try {
-					method = clazz.getDeclaredMethod(parserMethod, String.class);
-				}
-				catch (NoSuchMethodException e1) {
-					throw new IllegalStateException("the parser method must take '(String, Headers)' or '(String)'");
-				}
-				catch (SecurityException e1) {
-					throw new IllegalStateException(e1);
-				}
-			}
-			catch (SecurityException e) {
-				throw new IllegalStateException(e);
-			}
-			Method parseMethod = method;
-			if (method.getParameters().length > 1) {
-				this.parser = (str, headers) -> {
-					try {
-						return (T) parseMethod.invoke(null, str, headers);
-					}
-					catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
-						throw new IllegalStateException(e);
-					}
-				};
-			}
-			else {
-				this.parser = (str, headers) -> {
-					try {
-						return (T) parseMethod.invoke(null, str);
-					}
-					catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
-						throw new IllegalStateException(e);
-					}
-				};
-			}
+			BiFunction<String, Headers, T> parser;
+			parser = SerializationUtils.propertyToMethodInvokingFunction(parserMethod, String.class,
+					getClass().getClassLoader());
+			this.parser = parser;
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2020 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.kafka.support.serializer;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.function.BiFunction;
+
+import org.apache.kafka.common.header.Headers;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Utilities for serialization.
+ *
+ * @author Gary Russell
+ * @since 2.5
+ *
+ */
+public final class SerializationUtils {
+
+	private SerializationUtils() {
+	}
+
+	/**
+	 * Convert a property value (FQCN.methodName) to a {@link BiFunction} that takes a
+	 * payload and headers and returns some value. The method must have parameters
+	 * {@code (P, Headers)} or {@code (P)} and be declared as static.
+	 * @param <P> The {@link BiFunction} first parameter type.
+	 * @param <T> The {@link BiFunction} return type.
+	 * @param methodProperty the method name property.
+	 * @param payloadType the {@link BiFunction} first parameter type.
+	 * @param classLoader the class loader.
+	 * @return the function.
+	 */
+	@SuppressWarnings("unchecked")
+	public static <P, T> BiFunction<P, Headers, T> propertyToMethodInvokingFunction(String methodProperty,
+			Class<P> payloadType, ClassLoader classLoader) {
+
+		int lastDotPosn = methodProperty.lastIndexOf(".");
+		Assert.state(lastDotPosn > 1,
+				"the method property needs to be a class name followed by the method name, separated by '.'");
+		BiFunction<P, Headers, T> function;
+		Class<?> clazz;
+		try {
+			clazz = ClassUtils.forName(methodProperty.substring(0, lastDotPosn), classLoader);
+		}
+		catch (ClassNotFoundException | LinkageError e) {
+			throw new IllegalStateException(e);
+		}
+		String methodName = methodProperty.substring(lastDotPosn + 1);
+		Method method;
+		try {
+			method = clazz.getDeclaredMethod(methodName, payloadType, Headers.class);
+		}
+		catch (@SuppressWarnings("unused") NoSuchMethodException e) {
+			try {
+				method = clazz.getDeclaredMethod(methodName, payloadType);
+			}
+			catch (@SuppressWarnings("unused") NoSuchMethodException e1) {
+				throw new IllegalStateException("the parser method must take '(String, Headers)' or '(String)'");
+			}
+			catch (SecurityException e1) {
+				throw new IllegalStateException(e1);
+			}
+		}
+		catch (SecurityException e) {
+			throw new IllegalStateException(e);
+		}
+		Method parseMethod = method;
+		if (method.getParameters().length > 1) {
+			function = (str, headers) -> {
+				try {
+					return (T) parseMethod.invoke(null, str, headers);
+				}
+				catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
+					throw new IllegalStateException(e);
+				}
+			};
+		}
+		else {
+			function = (str, headers) -> {
+				try {
+					return (T) parseMethod.invoke(null, str);
+				}
+				catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
+					throw new IllegalStateException(e);
+				}
+			};
+		}
+		return function;
+	}
+
+}