From fe6356cee695e536753d062307d241ae423566bd Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 21 May 2020 10:53:51 +0200 Subject: [PATCH] GH-1967 Fixed support for converting collection types The main fix is actually in spring-cloud-function https://github.com/spring-cloud/spring-cloud-function/commit/417e54e0afe2d8d3ebdda1fed363a5a52a44ae5c The changes here are mainly polishing and adjusting around function changes Resolves #1967 --- .../ObjectStringMessageConverter.java | 8 ++- .../function/FunctionConfiguration.java | 4 +- .../function/FunctionBatchingTests.java | 60 ++++++++++++++++++- 3 files changed, 66 insertions(+), 6 deletions(-) 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 d47d9de35..76b61d4fd 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 @@ -23,6 +23,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; +import org.springframework.util.Assert; import org.springframework.util.MimeType; /** @@ -43,6 +44,7 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter { setStrictContentTypeMatch(true); } + @Override protected boolean supports(Class clazz) { return true; } @@ -67,8 +69,9 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter { return super.supportsMimeType(headers); } - protected Object convertFromInternal(Message message, Class targetClass, - Object conversionHint) { + @Override + protected Object convertFromInternal(Message message, Class targetClass, Object conversionHint) { + Assert.isTrue(String.class.isAssignableFrom(targetClass) || targetClass == Object.class, "This converter can only convert byte[] to String"); if (message.getPayload() != null) { if (message.getPayload() instanceof byte[]) { if (byte[].class.isAssignableFrom(targetClass)) { @@ -92,6 +95,7 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter { return null; } + @Override protected Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint) { if (payload != null) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index ff4474b60..2ab8dd3ae 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -523,7 +523,7 @@ public class FunctionConfiguration { * not attempt any conversion and sends a raw Message. */ @SuppressWarnings("rawtypes") - private static class FunctionWrapper implements Function, Object> { + private static class FunctionWrapper implements Function { private final Function function; private final ConsumerProperties consumerProperties; @@ -551,7 +551,7 @@ public class FunctionConfiguration { @SuppressWarnings("unchecked") @Override - public Object apply(Message message) { + public Object apply(Message message) { if (message != null && consumerProperties != null) { Map headersMap = (Map) ReflectionUtils .getField(this.headersField, message.getHeaders()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java index 82dc11e7c..d0284fb06 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.List; import java.util.function.Function; +import org.assertj.core.util.Arrays; import org.junit.Test; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -94,6 +95,48 @@ public class FunctionBatchingTests { }); } + @Test + public void testListStringPayloadConfigurationTextPlain() { + TestChannelBinderConfiguration.applicationContextRunner(ListStringPayloadConfiguration.class) + .withPropertyValues("spring.jmx.enabled=false", + "spring.cloud.stream.function.definition=func", + "spring.cloud.stream.bindings.func-in-0.content-type=text/plain") + .run(context -> { + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List bytes = Arrays.asList(new Object[] {"abc".getBytes(), "xyz".getBytes()}); + Message inputMessage = MessageBuilder.withPayload(bytes).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(new String(outputMessage.getPayload())).isEqualTo("[abc, xyz]"); + context.stop(); + }); + } + + @Test + public void testListObjectPayloadObjectConfigurationTextPlain() { + TestChannelBinderConfiguration.applicationContextRunner(ListObjectPayloadConfiguration.class) + .withPropertyValues("spring.jmx.enabled=false", + "spring.cloud.stream.function.definition=func", + "spring.cloud.stream.bindings.func-in-0.content-type=text/plain") + .run(context -> { + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List bytes = Arrays.asList(new Object[] {"abc".getBytes(), "xyz".getBytes()}); + Message inputMessage = MessageBuilder.withPayload(bytes).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(new String(outputMessage.getPayload())).isEqualTo("[abc, xyz]"); + context.stop(); + }); + } + @Test public void testSimpleBatchConfiguration() { TestChannelBinderConfiguration.applicationContextRunner(SimpleBatchConfiguration.class) @@ -172,6 +215,21 @@ public class FunctionBatchingTests { } + @EnableAutoConfiguration + public static class ListStringPayloadConfiguration { + @Bean + public Function, String> func() { + return x -> x.toString(); + } + } + + @EnableAutoConfiguration + public static class ListObjectPayloadConfiguration { + @Bean + public Function, String> func() { + return x -> x.toString(); + } + } @EnableAutoConfiguration public static class ListPayloadNotBatchConfiguration { @@ -192,9 +250,7 @@ public class FunctionBatchingTests { public void setName(String name) { this.name = name; } - } - } @EnableAutoConfiguration