From 05a8eef31ccb2d45c9b9ab88f29ad54ffdd31961 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 26 Feb 2018 15:47:11 -0500 Subject: [PATCH] GH-1247: Polled Consumer Simple Type Conversion Fixes https://github.com/spring-cloud/spring-cloud-stream/issues/1247 The `DefaultPollableMessageSource` was incorrectly setting the target type to `byte[]`. Use `Object` or, if the generic type of the `ParamterizedTypeReference` is a simple `Class` use it as the target object. This allows `ParameterizedTypeReference()` with contentType `text/plain` to work. --- .../binder/DefaultPollableMessageSource.java | 12 ++- .../binding/MessageConverterConfigurer.java | 38 ++++---- .../config/BindingServiceProperties.java | 18 ++-- .../stream/binder/PollableConsumerTests.java | 89 +++++++++++++------ 4 files changed, 98 insertions(+), 59 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index 7e03ab07e..0c602f51b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -40,6 +40,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.MessageBuilder; @@ -65,7 +66,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life private final List interceptors = new ArrayList<>(); private final MessagingTemplate messagingTemplate = new MessagingTemplate(); - + private final SmartMessageConverter messageConverter; private MessageSource source; @@ -81,7 +82,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life private BiConsumer> attributesProvider; private boolean running; - + public DefaultPollableMessageSource(SmartMessageConverter messageConverter) { this.messageConverter = messageConverter; } @@ -204,7 +205,12 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life return false; } if (type != null && this.messageConverter != null) { - Object payload = this.messageConverter.fromMessage(message, byte[].class, type); + Class targetType = type == null ? Object.class : + type.getType() instanceof Class ? (Class) type.getType() : Object.class; + Object payload = this.messageConverter.fromMessage(message, targetType, type); + if (payload == null) { + throw new MessageConversionException(message, "No converter could convert Message"); + } message = MessageBuilder.withPayload(payload) .copyHeaders(message.getHeaders()) .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 938601100..1ff55d750 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 @@ -89,7 +89,7 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig private final Map partitionKeyExtractors; private final Map partitionSelectors; - + private final Field headersField; public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties, @@ -258,7 +258,7 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig return Math.abs(hashCode); } } - + /** * Primary purpose of this interceptor is to enhance/enrich Message that sent to the *inbound* * channel with 'contentType' header for cases where 'contentType' is not present in the Message @@ -273,26 +273,26 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig } @Override - public Message doPreSend(Message message, MessageChannel channel) { + public Message doPreSend(Message message, MessageChannel channel) { @SuppressWarnings("unchecked") Map headersMap = (Map) ReflectionUtils.getField(MessageConverterConfigurer.this.headersField, message.getHeaders()); - + /* - * NOTE: The below code for BINDER_ORIGINAL_CONTENT_TYPE is to support legacy message format established + * NOTE: The below code for BINDER_ORIGINAL_CONTENT_TYPE is to support legacy message format established * in 1.x version of the framework and should/will no longer be supported in 3.x */ Object ct = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE); MimeType contentType = ct instanceof String ? MimeType.valueOf((String)ct) : (ct == null ? this.mimeType : (MimeType)ct); headersMap.remove(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE); // == end legacy note - + if (!message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) { headersMap.put(MessageHeaders.CONTENT_TYPE, contentType); } else if (message.getHeaders().get(MessageHeaders.CONTENT_TYPE) instanceof String) { headersMap.put(MessageHeaders.CONTENT_TYPE, MimeType.valueOf((String)message.getHeaders().get(MessageHeaders.CONTENT_TYPE))); } - + return message; } } @@ -313,7 +313,7 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig } @Override - public Message doPreSend(Message message, MessageChannel channel) { + public Message doPreSend(Message message, MessageChannel channel) { // ===== 1.3 backward compatibility code part-1 === String oct = message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE) ? message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString() : null; String ct = oct; @@ -321,34 +321,34 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig ct = JavaClassMimeTypeUtils.mimeTypeFromObject(message.getPayload(), ObjectUtils.nullSafeToString(oct)).toString(); } // ===== END 1.3 backward compatibility code part-1 === - + if (!message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) { @SuppressWarnings("unchecked") Map headersMap = (Map) ReflectionUtils.getField(MessageConverterConfigurer.this.headersField, message.getHeaders()); headersMap.put(MessageHeaders.CONTENT_TYPE, this.mimeType); } - + @SuppressWarnings("unchecked") - Message outboundMessage = message.getPayload() instanceof byte[] + Message outboundMessage = message.getPayload() instanceof byte[] ? (Message)message : (Message) this.messageConverter.toMessage(message.getPayload(), message.getHeaders()); if (outboundMessage == null) { throw new IllegalStateException("Failed to convert message: '" + message + "' to outbound message."); } - + /// ===== 1.3 backward compatibility code part-2 === - if (ct != null && !ct.equals(oct) && oct != null) { + if (ct != null && !ct.equals(oct) && oct != null) { @SuppressWarnings("unchecked") Map headersMap = (Map) ReflectionUtils.getField(MessageConverterConfigurer.this.headersField, message.getHeaders()); headersMap.put(MessageHeaders.CONTENT_TYPE, MimeType.valueOf(ct)); headersMap.put(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, MimeType.valueOf(oct)); } // ===== END 1.3 backward compatibility code part-2 === - return outboundMessage; + return outboundMessage; } } - + /** - * + * */ private abstract class AbstractContentTypeInterceptor extends ChannelInterceptorAdapter { final MimeType mimeType; @@ -356,17 +356,17 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig private AbstractContentTypeInterceptor(String contentType) { this.mimeType = MessageConverterUtils.getMimeType(contentType); } - + @Override public Message preSend(Message message, MessageChannel channel) { return message instanceof ErrorMessage ? message : this.doPreSend(message, channel); } - + protected abstract Message doPreSend(Message message, MessageChannel channel); } /** - * + * */ protected final class PartitioningInterceptor extends ChannelInterceptorAdapter { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index 343c7299b..51fdbe753 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -52,7 +52,7 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia private static final int DEFAULT_BINDING_RETRY_INTERVAL = 30; /** - * The instance id of the application: a number from 0 to instanceCount-1. Used for partitioning and with Kafka. + * The instance id of the application: a number from 0 to instanceCount-1. Used for partitioning and with Kafka. */ @Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}") private int instanceIndex; @@ -63,18 +63,18 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia private int instanceCount = 1; /** - * Additional binding properties (see {@link BinderProperties}) per binding name (e.g., 'input`). - * - * For example; This sets the content-type for the 'input' binding of a Sink application: + * Additional binding properties (see {@link BinderProperties}) per binding name (e.g., 'input`). + * + * For example; This sets the content-type for the 'input' binding of a Sink application: * 'spring.cloud.stream.bindings.input.contentType=text/plain' */ private Map bindings = new TreeMap<>( String.CASE_INSENSITIVE_ORDER); /** - * Additional per-binder properties (see {@link BinderProperties}) if more then one binder of the same type is used - * (i.e., connect to multiple instances of RabbitMq). Here you can specify multiple - * binder configurations, each with different environment settings. For example; + * Additional per-binder properties (see {@link BinderProperties}) if more then one binder of the same type is used + * (i.e., connect to multiple instances of RabbitMq). Here you can specify multiple + * binder configurations, each with different environment settings. For example; * spring.cloud.stream.binders.rabbit1.environment. . . , spring.cloud.stream.binders.rabbit2.environment. . . */ private Map binders = new HashMap<>(); @@ -93,9 +93,9 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia * Retry interval (in seconds) used to schedule binding attempts. Default: 30 sec. */ private int bindingRetryInterval = DEFAULT_BINDING_RETRY_INTERVAL; - + private ConfigurableApplicationContext applicationContext; - + private ConversionService conversionService; public Map getBindings() { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java index f4db36c93..fa2de2ea3 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java @@ -30,6 +30,8 @@ import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.stream.binder.test.TestChannelBinder; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.cloud.stream.binding.MessageConverterConfigurer; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.ApplicationContext; import org.springframework.core.ParameterizedTypeReference; @@ -56,9 +58,9 @@ import static org.assertj.core.api.Assertions.fail; public class PollableConsumerTests { private ApplicationContext context; - - private SmartMessageConverter messageConverter; - + + private SmartMessageConverter messageConverter; + @Before public void before() { this.messageConverter = new CompositeMessageConverterFactory().getMessageConverterForAllRegistered(); @@ -67,8 +69,8 @@ public class PollableConsumerTests { @Test public void testSimple() { TestChannelBinder binder = createBinder(); - MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); - + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); pollableSource.addInterceptor(new ChannelInterceptorAdapter() { @@ -99,12 +101,12 @@ public class PollableConsumerTests { @Test public void testConvertSimple() { TestChannelBinder binder = createBinder(); - MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); - + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"foo\":\"bar\"}".getBytes())); DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); - + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); properties.setMaxAttempts(1); properties.setBackOffInitialInterval(0); @@ -124,22 +126,53 @@ public class PollableConsumerTests { } @Test - public void testConvertList() { + public void testConvertSimpler() { TestChannelBinder binder = createBinder(); - MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); - - binder.setMessageSourceDelegate(() -> new GenericMessage<>("[{\"foo\":\"bar\"},{\"foo\":\"baz\"}]".getBytes())); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + BindingServiceProperties bsps = this.context.getBean(BindingServiceProperties.class); + BindingProperties props = new BindingProperties(); + props.setContentType("text/plain"); + bsps.setBindings(Collections.singletonMap("foo", props)); + + binder.setMessageSourceDelegate(() -> new GenericMessage<>("foo".getBytes())); DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); - + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); properties.setMaxAttempts(1); properties.setBackOffInitialInterval(0); - - - binder.bindPollableConsumer("foo", "bar", pollableSource, properties); - + final AtomicReference payload = new AtomicReference<>(); + assertThat(pollableSource.poll(received -> { + payload.set(received.getPayload()); + }, new ParameterizedTypeReference() {})).isTrue(); + assertThat(payload.get()).isInstanceOf(String.class); + assertThat(payload.get()).isEqualTo("foo"); + // test the cache for coverage + assertThat(pollableSource.poll(received -> { + payload.set(received.getPayload()); + }, new ParameterizedTypeReference() {})).isTrue(); + assertThat(payload.get()).isInstanceOf(String.class); + assertThat(payload.get()).isEqualTo("foo"); + } + + @Test + public void testConvertList() { + TestChannelBinder binder = createBinder(); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + binder.setMessageSourceDelegate(() -> new GenericMessage<>("[{\"foo\":\"bar\"},{\"foo\":\"baz\"}]".getBytes())); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); + + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); + properties.setMaxAttempts(1); + properties.setBackOffInitialInterval(0); + + + + binder.bindPollableConsumer("foo", "bar", pollableSource, properties); + final AtomicReference payload = new AtomicReference<>(); assertThat(pollableSource.poll(received -> { payload.set(received.getPayload()); @@ -154,12 +187,12 @@ public class PollableConsumerTests { @Test public void testConvertMap() { TestChannelBinder binder = createBinder(); - MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); - + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"qux\":{\"foo\":\"bar\"}}".getBytes())); DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); - + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); properties.setMaxAttempts(1); properties.setBackOffInitialInterval(0); @@ -177,8 +210,8 @@ public class PollableConsumerTests { @Test public void testEmbedded() { TestChannelBinder binder = createBinder(); - MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); - + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + binder.setMessageSourceDelegate(() -> { MessageValues original = new MessageValues("foo".getBytes(), Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/octet-stream")); @@ -215,8 +248,8 @@ public class PollableConsumerTests { @Test public void testErrors() { TestChannelBinder binder = createBinder(); - MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); - + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); pollableSource.addInterceptor(new ChannelInterceptorAdapter() { @@ -251,8 +284,8 @@ public class PollableConsumerTests { @Test public void testErrorsNoRetry() { TestChannelBinder binder = createBinder(); - MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); - + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); pollableSource.addInterceptor(new ChannelInterceptorAdapter() { @@ -280,9 +313,9 @@ public class PollableConsumerTests { assertThat(count.get()).isEqualTo(1); } - private TestChannelBinder createBinder() { + private TestChannelBinder createBinder(String... args) { this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration.getCompleteConfiguration()) - .web(WebApplicationType.NONE).run(); + .web(WebApplicationType.NONE).run(args); TestChannelBinder binder = context.getBean(TestChannelBinder.class); return binder; }