From df32fd6954148f2a6d6ad91db039c425aee18cdc Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 2 Mar 2020 11:47:30 +0100 Subject: [PATCH] Add MessageConverter filtering logic Aligned ApplicationJsonMessageMarshallingConverter with the same from s-c-Stream --- ...cationJsonMessageMarshallingConverter.java | 26 ++++++++++++++++++- ...ntextFunctionCatalogAutoConfiguration.java | 19 +++++++++++--- 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ApplicationJsonMessageMarshallingConverter.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ApplicationJsonMessageMarshallingConverter.java index 6390a3e40..b2369e9eb 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ApplicationJsonMessageMarshallingConverter.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ApplicationJsonMessageMarshallingConverter.java @@ -17,6 +17,7 @@ package org.springframework.cloud.function.context.config; import java.io.IOException; +import java.lang.reflect.Field; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.nio.charset.StandardCharsets; @@ -36,6 +37,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MappingJackson2MessageConverter; import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.util.MimeType; +import org.springframework.util.ReflectionUtils; /** * Variation of {@link MappingJackson2MessageConverter} to support marshalling and @@ -48,12 +51,16 @@ import org.springframework.messaging.converter.MessageConversionException; */ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter { + private final Field headersField; + private final Map typeCache = new ConcurrentHashMap<>(); ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) { if (objectMapper != null) { this.setObjectMapper(objectMapper); } + this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers"); + this.headersField.setAccessible(true); } @Override @@ -100,7 +107,6 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC else if (conversionHint instanceof ParameterizedType) { result = convertParameterizedType(message, (Type) conversionHint); } - if (result == null) { if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) { @@ -146,6 +152,11 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC else if (value instanceof String) { return objectMapper.readValue((String) value, typeToUse.getContentType()); } + else { + // fall back to simple type-conversion + // see https://github.com/spring-cloud/spring-cloud-stream/issues/1898 + return objectMapper.convertValue(value, typeToUse.getContentType()); + } } catch (Exception e) { logger.error("Failed to convert payload " + value, e); @@ -163,4 +174,17 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC } } + @SuppressWarnings("unchecked") + @Override + @Nullable + protected MimeType getMimeType(@Nullable MessageHeaders headers) { + Object contentType = headers.get(MessageHeaders.CONTENT_TYPE); + if (contentType instanceof byte[]) { + contentType = new String((byte[]) contentType, StandardCharsets.UTF_8); + contentType = ((String) contentType).replace("\"", ""); + Map headersMap = (Map) ReflectionUtils.getField(this.headersField, headers); + headersMap.put(MessageHeaders.CONTENT_TYPE, contentType); + } + return super.getMimeType(headers); + } } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java index 0a366f932..ca400a791 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java @@ -18,10 +18,10 @@ package org.springframework.cloud.function.context.config; import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.gson.Gson; @@ -70,14 +70,16 @@ public class ContextFunctionCatalogAutoConfiguration { static final String PREFERRED_MAPPER_PROPERTY = "spring.http.converters.preferred-json-mapper"; @Bean - public FunctionRegistry functionCatalog(Map messageConverters, @Nullable ObjectMapper objectMapper) { + public FunctionRegistry functionCatalog(List messageConverters, @Nullable ObjectMapper objectMapper) { ConversionService conversionService = new DefaultConversionService(); CompositeMessageConverter messageConverter = null; + messageConverters = messageConverters.stream() + .filter(c -> isConverterEligible(c)).collect(Collectors.toList()); List mcList = new ArrayList<>(); boolean addDefaultConverters = true; if (!CollectionUtils.isEmpty(messageConverters)) { - for (MessageConverter mc : messageConverters.values()) { + for (MessageConverter mc : messageConverters) { if (mc instanceof CompositeMessageConverter) { mcList.addAll(((CompositeMessageConverter) mc).getConverters()); addDefaultConverters = false; @@ -104,6 +106,17 @@ public class ContextFunctionCatalogAutoConfiguration { return new RoutingFunction(functionCatalog, functionInspector, functionProperties); } + private boolean isConverterEligible(Object messageConverter) { + String messageConverterName = messageConverter.getClass().getName(); + if (messageConverterName.startsWith("org.springframework.cloud.")) { + return true; + } + else if (!messageConverterName.startsWith("org.springframework.")) { + return true; + } + return false; + } + @Configuration(proxyBeanMethods = false) @ComponentScan(basePackages = "${spring.cloud.function.scan.packages:functions}", // includeFilters = @Filter(type = FilterType.ASSIGNABLE_TYPE,