SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1247: Polled Consumer Simple Type Conversion

Browse Source 
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<String>()` with contentType `text/plain` to work.
This commit is contained in:
Gary Russell
2018-02-26 15:47:11 -05:00
committed by Oleg Zhurakousky
parent a381f818dd
commit 05a8eef31c
4 changed files with 98 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java
View File

@@ -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<ChannelInterceptor> 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<AttributeAccessor, Message<?>> 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();

38
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -89,7 +89,7 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
private final Map<String, PartitionKeyExtractorStrategy> partitionKeyExtractors;
private final Map<String, PartitionSelectorStrategy> 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<String, Object> headersMap = (Map<String, Object>) 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<String, Object> headersMap = (Map<String, Object>) ReflectionUtils.getField(MessageConverterConfigurer.this.headersField, message.getHeaders());
headersMap.put(MessageHeaders.CONTENT_TYPE, this.mimeType);
}
@SuppressWarnings("unchecked")
Message<byte[]> outboundMessage = message.getPayload() instanceof byte[]
Message<byte[]> outboundMessage = message.getPayload() instanceof byte[]
? (Message<byte[]>)message : (Message<byte[]>) 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<String, Object> headersMap = (Map<String, Object>) 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 {

18
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java
View File

@@ -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<String, BindingProperties> 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<String, BinderProperties> 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<String, BindingProperties> getBindings() {

89
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java
View File

@@ -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<Object> 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<Object> properties = new ExtendedConsumerProperties<>(null);
properties.setMaxAttempts(1);
properties.setBackOffInitialInterval(0);
binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
final AtomicReference<Object> payload = new AtomicReference<>();
assertThat(pollableSource.poll(received -> {
payload.set(received.getPayload());
}, new ParameterizedTypeReference<String>() {})).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<String>() {})).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<Object> properties = new ExtendedConsumerProperties<>(null);
properties.setMaxAttempts(1);
properties.setBackOffInitialInterval(0);
binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
final AtomicReference<Object> 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<Object> 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;
}