Cherry-picked GH-1527 (777eec819e)

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());

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;
 		}
 	}

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;

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<byte[]> 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 {