SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2538 Remove ApplicationJsonMessageMarshallingConverter

Browse Source 
Resolves #2538
This commit is contained in:
Oleg Zhurakousky
2022-10-17 16:13:00 +02:00
parent e91c70154a
commit a5aa6d152c
8 changed files with 71 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
core/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
View File

@@ -379,7 +379,7 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
bindingServiceProperties.afterPropertiesSet();
MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer(
bindingServiceProperties,
new CompositeMessageConverterFactory(null, null).getMessageConverterForAllRegistered());
new CompositeMessageConverterFactory(null, null, null).getMessageConverterForAllRegistered());
messageConverterConfigurer.setBeanFactory(applicationContext.getBeanFactory());
return messageConverterConfigurer;
}

1
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -54,7 +54,6 @@ import org.springframework.util.MimeType;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
//import org.springframework.util.StringUtils;
/**
* A {@link MessageChannelConfigurer} that sets data types and message converters based on

6
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ContentTypeConfiguration.java
View File

@@ -23,11 +23,13 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.cloud.function.json.JsonMapper;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Role;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.MessageConverter;
@@ -43,13 +45,13 @@ class ContentTypeConfiguration {
@Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)
public CompositeMessageConverter configurableCompositeMessageConverter(
ObjectProvider<ObjectMapper> objectMapperObjectProvider,
List<MessageConverter> customMessageConverters) {
List<MessageConverter> customMessageConverters, @Nullable JsonMapper jsonMapper) {
customMessageConverters = customMessageConverters.stream()
.filter(c -> isConverterEligible(c)).collect(Collectors.toList());
CompositeMessageConverterFactory factory =
new CompositeMessageConverterFactory(customMessageConverters, objectMapperObjectProvider.getIfAvailable(ObjectMapper::new));
new CompositeMessageConverterFactory(customMessageConverters, objectMapperObjectProvider.getIfAvailable(ObjectMapper::new), jsonMapper);
return factory.getMessageConverterForAllRegistered();
}

172
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
View File

@@ -1,172 +0,0 @@
/*
* Copyright 2018-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.cloud.stream.converter;
import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.core.MethodParameter;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.converter.MessageConversionException;
/**
* Variation of {@link MappingJackson2MessageConverter} to support marshalling and
* unmarshalling of Messages's payload from 'String' or 'byte[]' to an instance of a
* 'targetClass' and and back to 'byte[]'.
*
* @author Oleg Zhurakousky
* @author Gary Russell
* @since 2.0
* @deprecated since 3.2 as we are no longer needed since functional-based programming model is no longer using it.
*/
class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter {
private final Map<Type, JavaType> typeCache = new ConcurrentHashMap<>();
ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) {
if (objectMapper != null) {
this.setObjectMapper(objectMapper);
}
}
@Override
protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers,
@Nullable Object conversionHint) {
if (payload instanceof byte[]) {
return payload;
}
else if (payload instanceof String) {
return ((String) payload).getBytes(StandardCharsets.UTF_8);
}
else {
return super.convertToInternal(payload, headers, conversionHint);
}
}
@Override
protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @Nullable Object hint) {
if (message.getPayload().getClass().getName().startsWith("org.springframework.kafka.support.KafkaNull")) {
return null;
}
Object conversionHint = hint;
Object result = null;
if (conversionHint instanceof MethodParameter) {
Class<?> conversionHintType = ((MethodParameter) conversionHint)
.getParameterType();
if (Message.class.isAssignableFrom(conversionHintType)) {
/*
* Ensures that super won't attempt to create Message as a result of
* conversion and stays at payload conversion only. The Message will
* eventually be created in
* MessageMethodArgumentResolver.resolveArgument(..)
*/
conversionHint = null;
}
else if (((MethodParameter) conversionHint)
.getGenericParameterType() instanceof ParameterizedType) {
ParameterizedTypeReference<Object> forType = ParameterizedTypeReference
.forType(((MethodParameter) conversionHint)
.getGenericParameterType());
result = convertParameterizedType(message, forType.getType());
}
}
else if (conversionHint instanceof ParameterizedTypeReference) {
result = convertParameterizedType(message, ((ParameterizedTypeReference<?>) conversionHint).getType());
}
else if (conversionHint instanceof ParameterizedType) {
result = convertParameterizedType(message, (Type) conversionHint);
}
if (result == null) {
if (message.getPayload() instanceof byte[]
&& String.class.isAssignableFrom(targetClass)) {
result = new String((byte[]) message.getPayload(),
StandardCharsets.UTF_8);
}
else {
result = super.convertFromInternal(message, targetClass, conversionHint);
}
}
return result;
}
private Object convertParameterizedType(Message<?> message, Type conversionHint) {
ObjectMapper objectMapper = this.getObjectMapper();
Object payload = message.getPayload();
try {
JavaType type = this.typeCache.get(conversionHint);
if (type == null) {
conversionHint = FunctionTypeUtils.isMessage(conversionHint)
? FunctionTypeUtils.getImmediateGenericType(conversionHint, 0)
: conversionHint;
type = objectMapper.getTypeFactory()
.constructType(conversionHint);
this.typeCache.put(conversionHint, type);
}
if (payload instanceof byte[]) {
return objectMapper.readValue((byte[]) payload, type);
}
else if (payload instanceof String) {
return objectMapper.readValue((String) payload, type);
}
else {
final JavaType typeToUse = type;
if (payload instanceof Collection) {
List<Object> collection = new ArrayList<>();
for (Object value : ((Collection<?>) payload)) {
try {
if (value instanceof byte[]) {
collection.add(objectMapper.readValue((byte[]) value, typeToUse.getContentType()));
}
else if (value instanceof String) {
collection.add(objectMapper.readValue((String) value, typeToUse.getContentType()));
}
else {
// fall back to simple type-conversion
// see https://github.com/spring-cloud/spring-cloud-stream/issues/1898
collection.add(objectMapper.convertValue(value, typeToUse.getContentType()));
}
}
catch (Exception e) {
throw new MessageConversionException("Failed to convert payload " + value, e);
}
}
return collection;
}
return null;
}
}
catch (IOException e) {
throw new MessageConversionException("Cannot parse payload ", e);
}
}
}

69
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2017 the original author or authors.
* Copyright 2015-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.
@@ -16,6 +16,8 @@
package org.springframework.cloud.stream.converter;
import java.io.ByteArrayOutputStream;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
@@ -23,10 +25,15 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.core.JsonEncoding;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.function.context.config.JsonMessageConverter;
import org.springframework.cloud.function.json.JacksonMapper;
import org.springframework.cloud.function.json.JsonMapper;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageHeaders;
@@ -55,8 +62,10 @@ public class CompositeMessageConverterFactory {
private final List<MessageConverter> converters;
private final JsonMapper jsonMapper;
public CompositeMessageConverterFactory() {
this(Collections.<MessageConverter>emptyList(), new ObjectMapper());
this(Collections.<MessageConverter>emptyList(), new ObjectMapper(), null);
}
/**
@@ -65,8 +74,9 @@ public class CompositeMessageConverterFactory {
*/
public CompositeMessageConverterFactory(
List<? extends MessageConverter> customConverters,
ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
ObjectMapper objectMapper, JsonMapper jsonMapper) {
this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
this.jsonMapper = jsonMapper == null ? new JacksonMapper(objectMapper) : jsonMapper;
if (!CollectionUtils.isEmpty(customConverters)) {
this.converters = new ArrayList<>(customConverters);
}
@@ -95,10 +105,36 @@ public class CompositeMessageConverterFactory {
}
private void initDefaultConverters() {
ApplicationJsonMessageMarshallingConverter applicationJsonConverter = new ApplicationJsonMessageMarshallingConverter(
this.objectMapper);
applicationJsonConverter.setStrictContentTypeMatch(true);
this.converters.add(applicationJsonConverter);
this.converters.add(new JsonMessageConverter(this.jsonMapper) {
@Override
protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers,
@Nullable Object conversionHint) {
/*
* We must revisit this. This is a copy from ApplicationMarshallingMessageConverter which derived from an older class etc. . .
* This attempts to use JSON conversion to convert something that is not json in the first place.
* For example Integer payload with application/json CT should actually fail since Integer is not a JSON.
* This is !!!!wrong!!!!! and ONLY remains here for backward compatibility.
*/
if (payload instanceof String) {
return ((String) payload).getBytes(StandardCharsets.UTF_8);
}
try {
if (byte[].class == getSerializedPayloadClass()) {
ByteArrayOutputStream out = new ByteArrayOutputStream(1024);
JsonEncoding encoding = getJsonEncoding(getMimeType(headers));
try (JsonGenerator generator = objectMapper.getFactory().createGenerator(out, encoding)) {
objectMapper.writeValue(generator, payload);
payload = out.toByteArray();
return payload;
}
}
}
catch (Exception e) {
logger.debug("Failed to convert to byte[]", e);
}
return super.convertToInternal(payload, headers, conversionHint);
}
});
this.converters.add(new ByteArrayMessageConverter() {
@Override
protected boolean supports(Class<?> clazz) {
@@ -111,6 +147,23 @@ public class CompositeMessageConverterFactory {
this.converters.add(new ObjectStringMessageConverter());
}
/**
* Determine the JSON encoding to use for the given content type.
* @param contentType the MIME type from the MessageHeaders, if any
* @return the JSON encoding to use (never {@code null})
*/
private JsonEncoding getJsonEncoding(@Nullable MimeType contentType) {
if (contentType != null && contentType.getCharset() != null) {
Charset charset = contentType.getCharset();
for (JsonEncoding encoding : JsonEncoding.values()) {
if (charset.name().equals(encoding.getJavaName())) {
return encoding;
}
}
}
return JsonEncoding.UTF8;
}
/**
* Creation method.
* @param mimeType the target MIME type

6
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java
View File

@@ -44,14 +44,14 @@ import static org.assertj.core.api.Assertions.fail;
*/
public class MessageConverterConfigurerTests {
// @Test
// @Test
void testConfigureOutputChannelWithBadContentType() {
BindingServiceProperties props = new BindingServiceProperties();
BindingProperties bindingProps = new BindingProperties();
bindingProps.setContentType("application/json");
props.setBindings(Collections.singletonMap("foo", bindingProps));
CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(
Collections.<MessageConverter>emptyList(), null);
Collections.<MessageConverter>emptyList(), null, null);
MessageConverterConfigurer configurer = new MessageConverterConfigurer(props,
converterFactory.getMessageConverterForAllRegistered());
QueueChannel out = new QueueChannel();
@@ -86,7 +86,7 @@ public class MessageConverterConfigurerTests {
};
CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(
Collections.<MessageConverter>singletonList(converter), null);
Collections.<MessageConverter>singletonList(converter), null, null);
MessageConverterConfigurer configurer = new MessageConverterConfigurer(props,
converterFactory.getMessageConverterForAllRegistered());
QueueChannel out = new QueueChannel();

81
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java
View File

@@ -1,81 +0,0 @@
/*
* Copyright 2021-2021 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.cloud.stream.converter;
import java.util.Collections;
import java.util.Map;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.json.JsonMapper;
import org.junit.jupiter.api.Test;
import org.springframework.core.ResolvableType;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.support.GenericMessage;
import static org.junit.Assert.fail;
@SuppressWarnings("deprecation")
public class ApplicationJsonMessageMarshallingConverterTests {
@Test
void badJson() {
try {
ApplicationJsonMessageMarshallingConverter converter = new ApplicationJsonMessageMarshallingConverter(
initObjectMapper());
converter.convertFromInternal(
new GenericMessage<>("{ notjson }".getBytes()), JsonNode.class, null);
fail();
}
catch (MessageConversionException e) {
// Good
}
catch (Throwable t) {
fail();
}
}
@Test
void errorPropagationTestOnCollection() {
ApplicationJsonMessageMarshallingConverter converter = new ApplicationJsonMessageMarshallingConverter(
JsonMapper.builder().build());
try {
converter.fromMessage(new GenericMessage<>(Collections.singletonList("{ \"field1\": 1 }")), Map.class,
ResolvableType.forClassWithGenerics(Map.class, String.class, String.class).getType());
fail();
}
catch (MessageConversionException e) {
// good
}
catch (Throwable t) {
fail();
}
}
private ObjectMapper initObjectMapper() {
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
return objectMapper;
}
}

4
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -538,12 +538,12 @@ public class ImplicitFunctionBindingTests {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierWithExplicitPollerConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false", "--spring.cloud.stream.poller.fixed-delay=2000")) {
.run("--spring.jmx.enabled=false", "--spring.cloud.stream.poller.fixed-delay=1500")) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
PollerMetadata pollerMetadata = context.getBean(PollerMetadata.class);
assertThat(((PeriodicTrigger) pollerMetadata.getTrigger()).getPeriod()).isEqualTo(2000);
assertThat(((PeriodicTrigger) pollerMetadata.getTrigger()).getPeriodDuration()).isEqualTo(Duration.ofMillis(1500));
Message<byte[]> outputMessage = outputDestination.receive(6000);
assertThat(outputMessage.getPayload()).isEqualTo("hello".getBytes());