diff --git a/core/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/core/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index c96bd78f1..df82eb3cb 100644 --- a/core/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/core/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -379,7 +379,7 @@ public abstract class AbstractBinderTests objectMapperObjectProvider, - List customMessageConverters) { + List 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(); } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java deleted file mode 100644 index 5d368d75c..000000000 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java +++ /dev/null @@ -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 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 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 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); - } - } -} diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java index c49ed571d..8ac0e1395 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java @@ -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 converters; + private final JsonMapper jsonMapper; + public CompositeMessageConverterFactory() { - this(Collections.emptyList(), new ObjectMapper()); + this(Collections.emptyList(), new ObjectMapper(), null); } /** @@ -65,8 +74,9 @@ public class CompositeMessageConverterFactory { */ public CompositeMessageConverterFactory( List 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 diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java index cdd87b15d..4a20ffa5c 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java @@ -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.emptyList(), null); + Collections.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.singletonList(converter), null); + Collections.singletonList(converter), null, null); MessageConverterConfigurer configurer = new MessageConverterConfigurer(props, converterFactory.getMessageConverterForAllRegistered()); QueueChannel out = new QueueChannel(); diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java deleted file mode 100644 index b62dc930c..000000000 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java +++ /dev/null @@ -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; - } - -} diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index b3a70b35d..07da13232 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -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 outputMessage = outputDestination.receive(6000); assertThat(outputMessage.getPayload()).isEqualTo("hello".getBytes());