diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java index 9a261a95e..387ab1ef9 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -34,7 +34,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -73,7 +73,7 @@ public class TextPlainToJsonConversionTest { * @since 2.0: Conversion from text/plain -> json is no longer supported. Strict contentType only. * @throws Exception */ - @Test(expected = MessageConversionException.class) + @Test(expected = MessagingException.class) public void testTextPlainToJsonConversionOnInput() throws Exception { testProcessor.input().send(MessageBuilder.withPayload("{\"name\":\"Bar\"}") .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build()); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 0e5e1a6f1..318584f54 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.binding; import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; import java.util.Map; import org.apache.commons.logging.Log; @@ -52,6 +53,7 @@ import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; @@ -282,6 +284,10 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig headersMap.put(MessageHeaders.CONTENT_TYPE, MimeType.valueOf((String)message.getHeaders().get(MessageHeaders.CONTENT_TYPE))); } + if (message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString().startsWith("text") && message.getPayload() instanceof byte[]) { + message = MessageBuilder.withPayload(new String((byte[])message.getPayload(), StandardCharsets.UTF_8)).copyHeaders(message.getHeaders()).build(); + } + return message; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java index d0ccce393..f70445e31 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -17,7 +17,9 @@ package org.springframework.cloud.stream.converter; import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; @@ -31,13 +33,14 @@ import org.springframework.util.MimeType; * {@link Object#toString()}. * * @author Marius Bogoevici + * @author Oleg Zhurakousky * * @since 1.2 */ public class ObjectStringMessageConverter extends AbstractMessageConverter { public ObjectStringMessageConverter() { - super(new MimeType("text", "plain", Charset.forName("UTF-8"))); + super(new MimeType("text", "*", Charset.forName("UTF-8"))); setStrictContentTypeMatch(true); } @@ -48,16 +51,40 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter { @Override protected boolean canConvertFrom(Message message, Class targetClass) { // only supports the conversion to String - return supportsMimeType(message.getHeaders()) && String.class == targetClass; + return supportsMimeType(message.getHeaders()); + } + + @Override + protected boolean supportsMimeType(@Nullable MessageHeaders headers) { + MimeType mimeType = getMimeType(headers); + if (mimeType != null && !super.supportsMimeType(headers)) { + for (MimeType current : getSupportedMimeTypes()) { + if (current.getType().equals(mimeType.getType())) { + return true; + } + } + } + + return false; } protected Object convertFromInternal(Message message, Class targetClass, Object conversionHint) { if (message.getPayload() != null) { if (message.getPayload() instanceof byte[]) { - return new String((byte[]) message.getPayload(), Charset.forName("UTF-8")); + if (byte[].class.isAssignableFrom(targetClass)) { + return message.getPayload(); + } + else { + return new String((byte[]) message.getPayload(), StandardCharsets.UTF_8); + } } else { - return message.getPayload().toString(); + if (byte[].class.isAssignableFrom(targetClass)) { + return message.getPayload().toString().getBytes(StandardCharsets.UTF_8); + } + else { + return message.getPayload(); + } } } return null; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java index deb49369b..704fb8f34 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java @@ -47,6 +47,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.converter.AbstractMessageConverter; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.handler.annotation.Payload; @@ -217,6 +218,22 @@ public class ContentTypeTckTests { assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } + @Test + public void typelessToPojoInboundContentTypeBinding() { + ApplicationContext context = new SpringApplicationBuilder(TypelessToPojoStreamListener.class) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.bindings.input.contentType=text/plain", "--spring.jmx.enabled=false"); + InputDestination source = context.getBean(InputDestination.class); + OutputDestination target = context.getBean(OutputDestination.class); + String jsonPayload = "{\"name\":\"oleg\"}"; + source.send(new GenericMessage<>(jsonPayload.getBytes())); + Message outputMessage = target.receive(); + assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); + assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); + } + + + @Test public void stringToPojoInboundContentTypeHeader() { ApplicationContext context = new SpringApplicationBuilder(StringToPojoStreamListener.class) @@ -433,7 +450,7 @@ public class ContentTypeTckTests { TestChannelBinder binder = context.getBean(TestChannelBinder.class); String jsonPayload = "{\"name\":\"oleg\"}"; source.send(new GenericMessage<>(jsonPayload.getBytes())); - assertTrue(binder.getLastError().getPayload() instanceof MessageConversionException); + assertTrue(binder.getLastError().getPayload() instanceof MessagingException); } @Test @@ -535,6 +552,18 @@ public class ContentTypeTckTests { } } + @EnableBinding(Processor.class) + @Import(TestChannelBinderConfiguration.class) + public static class TypelessToPojoStreamListener { + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public Person echo(Object value) throws Exception { + ObjectMapper mapper = new ObjectMapper(); + //assume it is string because CT is text/plain + return mapper.readValue((String)value, Person.class); + } + } + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class ByteArrayToByteArrayStreamListener {