Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3295 Datatype Channel : Use MessageConverter

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3295

Instead of invoking the conversion service directly, do it
via a MessageConverter (DefaultDatatypeChannelMessageConverter).

That way, users can override the MessageConverter; for example,
XD wants to look at the content-type header during the conversion
process (and update the content-type).

INT-3295 Polishing

PR Comments.
This commit is contained in:
Gary Russell
2014-02-12 20:20:43 +02:00
committed by Artem Bilan
parent 814caea5d5
commit 25489adaa3
13 changed files with 301 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -20,19 +20,19 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.OrderComparator;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -51,8 +51,6 @@ import org.springframework.util.StringUtils;
public abstract class AbstractMessageChannel extends IntegrationObjectSupport
implements MessageChannel, TrackableComponent, ChannelInterceptorAware {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile boolean shouldTrack = false;
private volatile Class<?>[] datatypes = new Class<?>[] { Object.class };
@@ -61,6 +59,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
private volatile String fullChannelName;
private volatile MessageConverter messageConverter;
@Override
public String getComponentType() {
@@ -83,7 +82,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
*
* @param datatypes The supported data types.
*
* @see #setConversionService(ConversionService)
* @see #setMessageConverter(MessageConverter)
*/
public void setDatatypes(Class<?>... datatypes) {
this.datatypes = (datatypes != null && datatypes.length > 0)
@@ -129,14 +128,36 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
* does not already match. If this property is not set explicitly but
* the channel is managed within a context, it will attempt to locate a
* bean named "integrationConversionService" defined within that context.
* Finally, if that bean is not available, it will fallback to the
* "conversionService" bean, if available.
*
* @param conversionService The conversion service.
* @deprecated No longer used; see {@link DefaultDatatypeChannelMessageConverter}.
*/
@Deprecated
@Override
public void setConversionService(ConversionService conversionService) {
super.setConversionService(conversionService);
if (logger.isWarnEnabled()) {
logger.warn("The conversion service is no longer used; see setMessageConverter()");
}
}
/**
* Specify the {@link MessageConverter} to use when trying to convert to
* one of this channel's supported datatypes (in order) for a Message whose payload
* does not already match.
* <p>
* <b>Note:</b> only the {@link MessageConverter#fromMessage(Message, Class)}
* method is used. If the returned object is not a {@link Message}, the inbound
* headers will be copied; if the returned object is a {@code Message}, it is
* expected that the converter will have fully populated the headers; no
* further action is performed by the channel. If {@code null} is returned,
* conversion to the next datatype (if any) will be attempted.
*
* Defaults to a {@link DefaultDatatypeChannelMessageConverter}.
*
* @param messageConverter The message converter.
*/
public void setMessageConverter(MessageConverter messageConverter) {
this.messageConverter = messageConverter;
}
/**
@@ -156,6 +177,21 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
return this.interceptors;
}
@Override
protected void onInit() throws Exception {
super.onInit();
if (this.messageConverter == null) {
if (this.getBeanFactory() != null) {
if (this.getBeanFactory().containsBean(
IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME)) {
this.messageConverter = this.getBeanFactory().getBean(
IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME,
MessageConverter.class);
}
}
}
}
/**
* Returns the fully qualified channel name including the application context
* id, if available.
@@ -235,13 +271,17 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
return message;
}
}
// second pass applies conversion if possible, attempting datatypes in order
ConversionService conversionService = this.getConversionService();
if (conversionService != null) {
if (this.messageConverter != null) {
// second pass applies conversion if possible, attempting datatypes in order
for (Class<?> datatype : this.datatypes) {
if (conversionService.canConvert(message.getPayload().getClass(), datatype)) {
Object convertedPayload = conversionService.convert(message.getPayload(), datatype);
return MessageBuilder.withPayload(convertedPayload).copyHeaders(message.getHeaders()).build();
Object converted = this.messageConverter.fromMessage(message, datatype);
if (converted != null) {
if (converted instanceof Message) {
return (Message<?>) converted;
}
else {
return MessageBuilder.withPayload(converted).copyHeaders(message.getHeaders()).build();
}
}
}
}

28
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationRegistrar.java
View File

@@ -48,6 +48,7 @@ import org.springframework.integration.config.annotation.MessagingAnnotationPost
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.expression.IntegrationEvaluationContextAwareBeanPostProcessor;
import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
@@ -55,6 +56,7 @@ import org.springframework.util.StringUtils;
* {@link ImportBeanDefinitionRegistrar} implementation that configures integration infrastructure.
*
* @author Artem Bilan
* @author Gary Russell
* @since 4.0
*/
public class IntegrationRegistrar implements ImportBeanDefinitionRegistrar, BeanClassLoaderAware {
@@ -84,6 +86,7 @@ public class IntegrationRegistrar implements ImportBeanDefinitionRegistrar, Bean
this.registerHeaderChannelRegistry(registry);
this.registerBuiltInBeans(registry);
this.registerDefaultConfiguringBeanFactoryPostProcessor(registry);
this.registerDefaultDatatypeChannelMessageConverter(registry);
if (importingClassMetadata != null) {
this.registerMessagingAnnotationPostProcessors(importingClassMetadata, registry);
}
@@ -325,4 +328,29 @@ public class IntegrationRegistrar implements ImportBeanDefinitionRegistrar, Bean
}
}
/**
* Register the default datatype channel MessageConverter.
*
* @param registry the registry.
*/
private void registerDefaultDatatypeChannelMessageConverter(BeanDefinitionRegistry registry) {
boolean alreadyRegistered = false;
if (registry instanceof ListableBeanFactory) {
alreadyRegistered = ((ListableBeanFactory) registry)
.containsBean(IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME);
}
else {
alreadyRegistered = registry
.isBeanNameInUse(IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME);
}
if (!alreadyRegistered) {
BeanDefinitionBuilder converterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DefaultDatatypeChannelMessageConverter.class);
registry.registerBeanDefinition(
IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME,
converterBuilder.getBeanDefinition());
}
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/MessageHistoryRegistrar.java
View File

@@ -54,7 +54,7 @@ public class MessageHistoryRegistrar implements ImportBeanDefinitionRegistrar {
componentNamePatterns = componentNamePatternsString.substring(0, componentNamePatternsString.length() - 1);
}
if (!registry.containsBeanDefinition(IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER)) {
if (!registry.containsBeanDefinition(IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER_BEAN_NAME)) {
Set<Object> componentNamePatternsSet = new ManagedSet<Object>();
componentNamePatternsSet.add(componentNamePatterns);
@@ -62,11 +62,11 @@ public class MessageHistoryRegistrar implements ImportBeanDefinitionRegistrar {
.addPropertyValue("componentNamePatternsSet", componentNamePatternsSet)
.getBeanDefinition();
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER, messageHistoryConfigurer);
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER_BEAN_NAME, messageHistoryConfigurer);
}
else {
BeanDefinition beanDefinition = registry.getBeanDefinition(IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER);
BeanDefinition beanDefinition = registry.getBeanDefinition(IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER_BEAN_NAME);
PropertyValue propertyValue = beanDefinition
.getPropertyValues().getPropertyValue("componentNamePatternsSet");
if (propertyValue != null) {

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java
View File

@@ -55,6 +55,10 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser
if (StringUtils.hasText(datatypeAttr)) {
builder.addPropertyValue("datatypes", datatypeAttr);
}
String messageConverter = element.getAttribute("message-converter");
if (StringUtils.hasText(messageConverter)) {
builder.addPropertyReference("messageConverter", messageConverter);
}
builder.addPropertyValue("interceptors", interceptors);
AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
String scopeAttr = element.getAttribute("scope");

4
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
View File

@@ -66,7 +66,9 @@ public abstract class IntegrationContextUtils {
public static final String INTEGRATION_CONFIGURATION_POST_PROCESSOR_BEAN_NAME = "IntegrationConfigurationBeanFactoryPostProcessor";
public static final String INTEGRATION_MESSAGE_HISTORY_CONFIGURER = "messageHistoryConfigurer";
public static final String INTEGRATION_MESSAGE_HISTORY_CONFIGURER_BEAN_NAME = "messageHistoryConfigurer";
public static final String INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME = "datatypeChannelMessageConverter";
/**
* @param beanFactory BeanFactory for lookup, must not be null.

67
spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* Copyright 2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.support.converter;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
/**
* Default message converter for datatype channels. Registered under bean name
* 'datatypeChannelMessageConverter'. Delegates to the 'integrationConversionService',
* if present.
*
* @author Gary Russell
* @since 4.0
*
*/
public class DefaultDatatypeChannelMessageConverter extends IntegrationObjectSupport implements MessageConverter {
/**
* Specify the {@link ConversionService} to use when trying to convert to
* requested type. If this property is not set explicitly but
* the converter is managed within a context, it will attempt to locate a
* bean named "integrationConversionService" defined within that context.
*
* @param conversionService The conversion service.
*/
@Override
public void setConversionService(ConversionService conversionService) {
super.setConversionService(conversionService);
}
/**
* @return the converted payload or null if conversion is not possible.
*/
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
ConversionService conversionService = this.getConversionService();
if (conversionService != null) {
if (conversionService.canConvert(message.getPayload().getClass(), targetClass)) {
return conversionService.convert(message.getPayload(), targetClass);
}
}
return null;
}
@Override
public Message<?> toMessage(Object payload, MessageHeaders header) {
throw new UnsupportedOperationException("This converter does not support this method");
}
}

23
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd
View File

@@ -498,6 +498,29 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Used with 'datatype' to convert the message payload, if necessary,
to one of the datatypes (in order).
Note: only the MessageConverter.fromMessage(Message, Class) method is used.
If the returned object is not a Message, the inbound headers will be copied;
if the returned object is a Message, it is expected that the converter wil
have fully populated the headers; no further action is performed by the channel.
If null is returned, conversion to the next datatype (if any) will be attempted.
Default is a 'DefaultDatatypeChannelMessageConverter'
which, in turn, delegates to the 'integrationConversionService'
(if present).
]]>
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.converter.MessageConverter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:element name="gateway">

36
spring-integration-core/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -25,6 +25,7 @@ import java.util.Date;
import org.junit.Test;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.context.support.ConversionServiceFactoryBean;
import org.springframework.context.support.GenericApplicationContext;
@@ -33,16 +34,18 @@ import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Mark Fisher
* @author Gunnar Hillert
* @author Artem Bilan
* @author Gary Russell
* @since 2.0
*/
public class DatatypeChannelTests {
@@ -63,7 +66,9 @@ public class DatatypeChannelTests {
public void unsupportedTypeButConversionServiceSupports() {
QueueChannel channel = createChannel(Integer.class);
ConversionService conversionService = new DefaultConversionService();
channel.setConversionService(conversionService);
DefaultDatatypeChannelMessageConverter converter = new DefaultDatatypeChannelMessageConverter();
converter.setConversionService(conversionService);
channel.setMessageConverter(converter);
assertTrue(channel.send(new GenericMessage<String>("123")));
}
@@ -71,7 +76,9 @@ public class DatatypeChannelTests {
public void unsupportedTypeAndConversionServiceDoesNotSupport() {
QueueChannel channel = createChannel(Integer.class);
ConversionService conversionService = new DefaultConversionService();
channel.setConversionService(conversionService);
DefaultDatatypeChannelMessageConverter converter = new DefaultDatatypeChannelMessageConverter();
converter.setConversionService(conversionService);
channel.setMessageConverter(converter);
assertTrue(channel.send(new GenericMessage<Boolean>(Boolean.TRUE)));
}
@@ -80,11 +87,14 @@ public class DatatypeChannelTests {
QueueChannel channel = createChannel(Integer.class);
GenericConversionService conversionService = new DefaultConversionService();
conversionService.addConverter(new Converter<Boolean, Integer>() {
@Override
public Integer convert(Boolean source) {
return source ? 1 : 0;
}
});
channel.setConversionService(conversionService);
DefaultDatatypeChannelMessageConverter converter = new DefaultDatatypeChannelMessageConverter();
converter.setConversionService(conversionService);
channel.setMessageConverter(converter);
assertTrue(channel.send(new GenericMessage<Boolean>(Boolean.TRUE)));
assertEquals(1, channel.receive().getPayload());
}
@@ -93,6 +103,7 @@ public class DatatypeChannelTests {
public void conversionServiceBeanUsedByDefault() {
GenericApplicationContext context = new GenericApplicationContext();
Converter<Boolean, Integer> converter = new Converter<Boolean, Integer>() {
@Override
public Integer convert(Boolean source) {
return source ? 1 : 0;
}
@@ -102,6 +113,10 @@ public class DatatypeChannelTests {
conversionServiceBuilder.addPropertyValue("converters", Collections.singleton(converter));
context.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME,
conversionServiceBuilder.getBeanDefinition());
BeanDefinition messageConverter = BeanDefinitionBuilder.genericBeanDefinition(
DefaultDatatypeChannelMessageConverter.class).getBeanDefinition();
context.registerBeanDefinition(
IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME, messageConverter);
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(QueueChannel.class);
channelBuilder.addPropertyValue("datatypes", "java.lang.Integer, java.util.Date");
context.registerBeanDefinition("testChannel", channelBuilder.getBeanDefinition());
@@ -110,18 +125,21 @@ public class DatatypeChannelTests {
QueueChannel channel = context.getBean("testChannel", QueueChannel.class);
assertTrue(channel.send(new GenericMessage<Boolean>(Boolean.TRUE)));
assertEquals(1, channel.receive().getPayload());
context.close();
}
@Test
public void conversionServiceReferenceOverridesDefault() {
GenericApplicationContext context = new GenericApplicationContext();
Converter<Boolean, Integer> defaultConverter = new Converter<Boolean, Integer>() {
@Override
public Integer convert(Boolean source) {
return source ? 1 : 0;
}
};
GenericConversionService customConversionService = new DefaultConversionService();
customConversionService.addConverter(new Converter<Boolean, Integer>() {
@Override
public Integer convert(Boolean source) {
return source ? 99 : -99;
}
@@ -130,15 +148,21 @@ public class DatatypeChannelTests {
BeanDefinitionBuilder.genericBeanDefinition(ConversionServiceFactoryBean.class);
conversionServiceBuilder.addPropertyValue("converters", Collections.singleton(defaultConverter));
context.registerBeanDefinition("conversionService", conversionServiceBuilder.getBeanDefinition());
BeanDefinitionBuilder messageConverterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
DefaultDatatypeChannelMessageConverter.class);
messageConverterBuilder.addPropertyValue("conversionService", customConversionService);
context.registerBeanDefinition(
IntegrationContextUtils.INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME,
messageConverterBuilder.getBeanDefinition());
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(QueueChannel.class);
channelBuilder.addPropertyValue("datatypes", "java.lang.Integer, java.util.Date");
channelBuilder.addPropertyValue("conversionService", customConversionService);
context.registerBeanDefinition("testChannel", channelBuilder.getBeanDefinition());
context.refresh();
QueueChannel channel = context.getBean("testChannel", QueueChannel.class);
assertTrue(channel.send(new GenericMessage<Boolean>(Boolean.TRUE)));
assertEquals(99, channel.receive().getPayload());
context.close();
}
@Test

66
spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -32,8 +32,9 @@ import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.FatalBeanException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.convert.converter.Converter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -41,11 +42,15 @@ import org.springframework.integration.config.TestChannelInterceptor;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.GenericMessage;
@@ -53,6 +58,7 @@ import org.springframework.messaging.support.GenericMessage;
* @author Mark Fisher
* @author Iwein Fuld
* @author Gunnar Hillert
* @author Gary Russell
*
* @see ChannelWithCustomQueueParserTests
*/
@@ -60,7 +66,7 @@ public class ChannelParserTests {
@Test(expected = FatalBeanException.class)
public void testChannelWithoutId() {
new ClassPathXmlApplicationContext("channelWithoutId.xml", this.getClass());
new ClassPathXmlApplicationContext("channelWithoutId.xml", this.getClass()).close();
}
@Test
@@ -73,6 +79,7 @@ public class ChannelParserTests {
assertTrue(result);
}
assertFalse(channel.send(new GenericMessage<String>("test"), 3));
context.close();
}
@Test
@@ -86,6 +93,7 @@ public class ChannelParserTests {
assertThat(dispatcher, is(instanceOf(UnicastingDispatcher.class)));
assertThat(new DirectFieldAccessor(dispatcher).getPropertyValue("loadBalancingStrategy"),
is(instanceOf(RoundRobinLoadBalancingStrategy.class)));
context.close();
}
@Test
@@ -98,6 +106,7 @@ public class ChannelParserTests {
Object dispatcher = accessor.getPropertyValue("dispatcher");
assertThat(dispatcher, is(instanceOf(UnicastingDispatcher.class)));
assertNull(new DirectFieldAccessor(dispatcher).getPropertyValue("loadBalancingStrategy"));
context.close();
}
@Test
@@ -123,6 +132,7 @@ public class ChannelParserTests {
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
Object executorBean = context.getBean("taskExecutor");
assertEquals(executorBean, innerExecutor);
context.close();
}
@Test
@@ -134,6 +144,7 @@ public class ChannelParserTests {
assertEquals(QueueChannel.class, channelWithCustomQueue.getClass());
Object actualQueue = new DirectFieldAccessor(channelWithCustomQueue).getPropertyValue("queue");
assertSame(customQueue, actualQueue);
context.close();
}
@Test
@@ -142,6 +153,7 @@ public class ChannelParserTests {
.getClass());
MessageChannel channel = (MessageChannel) context.getBean("integerChannel");
assertTrue(channel.send(new GenericMessage<Integer>(123)));
context.close();
}
@Test(expected = MessageDeliveryException.class)
@@ -150,6 +162,8 @@ public class ChannelParserTests {
.getClass());
MessageChannel channel = (MessageChannel) context.getBean("integerChannel");
channel.send(new GenericMessage<String>("incorrect type"));
context.close();
assertTrue(TestUtils.getPropertyValue(channel, "messageConverter") instanceof UselessMessageConverter);
}
@Test
@@ -159,6 +173,10 @@ public class ChannelParserTests {
MessageChannel channel = (MessageChannel) context.getBean("numberChannel");
assertTrue(channel.send(new GenericMessage<Integer>(123)));
assertTrue(channel.send(new GenericMessage<Double>(123.45)));
assertTrue(channel.send(new GenericMessage<Boolean>(Boolean.TRUE)));
assertTrue(TestUtils.getPropertyValue(channel, "messageConverter") instanceof DefaultDatatypeChannelMessageConverter);
assertNotNull(TestUtils.getPropertyValue(channel, "messageConverter.conversionService"));
context.close();
}
@Test
@@ -168,6 +186,7 @@ public class ChannelParserTests {
MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel");
assertTrue(channel.send(new GenericMessage<Integer>(123)));
assertTrue(channel.send(new GenericMessage<String>("accepted type")));
context.close();
}
@Test(expected = MessageDeliveryException.class)
@@ -175,12 +194,13 @@ public class ChannelParserTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this
.getClass());
MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel");
channel.send(new GenericMessage<Boolean>(true));
channel.send(new GenericMessage<Boolean>(Boolean.TRUE));
context.close();
}
@Test
public void testChannelInteceptorRef() {
ApplicationContext context = new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", this
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", this
.getClass());
PollableChannel channel = (PollableChannel) context.getBean("channelWithInterceptorRef");
TestChannelInterceptor interceptor = (TestChannelInterceptor) context.getBean("interceptor");
@@ -190,21 +210,23 @@ public class ChannelParserTests {
assertEquals(0, interceptor.getReceiveCount());
channel.receive();
assertEquals(1, interceptor.getReceiveCount());
context.close();
}
@Test
public void testChannelInteceptorInnerBean() {
ApplicationContext context = new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", this
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", this
.getClass());
PollableChannel channel = (PollableChannel) context.getBean("channelWithInterceptorInnerBean");
channel.send(new GenericMessage<String>("test"));
Message<?> transformed = channel.receive(1000);
assertEquals("TEST", transformed.getPayload());
context.close();
}
@Test
public void testPriorityChannelWithDefaultComparator() {
ApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this.getClass());
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this.getClass());
PollableChannel channel = (PollableChannel) context.getBean("priorityChannelWithDefaultComparator");
Message<String> lowPriorityMessage = MessageBuilder.withPayload("low").setPriority(-14).build();
Message<String> midPriorityMessage = MessageBuilder.withPayload("mid").setPriority(0).build();
@@ -218,11 +240,12 @@ public class ChannelParserTests {
assertEquals("high", reply1.getPayload());
assertEquals("mid", reply2.getPayload());
assertEquals("low", reply3.getPayload());
context.close();
}
@Test
public void testPriorityChannelWithCustomComparator() {
ApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this
.getClass());
PollableChannel channel = (PollableChannel) context.getBean("priorityChannelWithCustomComparator");
channel.send(new GenericMessage<String>("C"));
@@ -237,11 +260,12 @@ public class ChannelParserTests {
assertEquals("B", reply2.getPayload());
assertEquals("C", reply3.getPayload());
assertEquals("D", reply4.getPayload());
context.close();
}
@Test
public void testPriorityChannelWithIntegerDatatypeEnforced() {
ApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this
.getClass());
PollableChannel channel = (PollableChannel) context.getBean("integerOnlyPriorityChannel");
channel.send(new GenericMessage<Integer>(3));
@@ -259,6 +283,7 @@ public class ChannelParserTests {
threwException = true;
}
assertTrue(threwException);
context.close();
}
public static class TestInterceptor extends ChannelInterceptorAdapter {
@@ -270,4 +295,27 @@ public class ChannelParserTests {
}
public static class TestConverter implements Converter<Boolean, Number> {
@Override
public Number convert(Boolean source) {
return source ? 1 : 0;
}
}
public static class UselessMessageConverter implements MessageConverter {
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
return null;
}
@Override
public Message<?> toMessage(Object payload, MessageHeaders header) {
return null;
}
}
}

11
spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml
View File

@@ -6,6 +6,10 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<converter>
<beans:bean class="org.springframework.integration.channel.config.ChannelParserTests$TestConverter" />
</converter>
<channel id="capacityChannel">
<queue capacity="10" />
</channel>
@@ -27,15 +31,18 @@
<publish-subscribe-channel id="publishSubscribeChannelWithTaskExecutorRef"
task-executor="taskExecutor" />
<channel id="integerChannel" datatype="java.lang.Integer">
<channel id="integerChannel" datatype="java.lang.Integer" message-converter="uselessConverter">
<queue capacity="10" />
</channel>
<beans:bean id="uselessConverter" class="org.springframework.integration.channel.config.ChannelParserTests$UselessMessageConverter" />
<channel id="numberChannel" datatype="java.lang.Number">
<queue capacity="10" />
</channel>
<channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number">
<channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number"
message-converter="uselessConverter">
<queue capacity="10" />
</channel>

4
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java
View File

@@ -255,7 +255,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
}
if (IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER.equals(beanName)
if (IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER_BEAN_NAME.equals(beanName)
&& bean instanceof MessageHistoryConfigurer) {
this.messageHistoryConfigurer = (MessageHistoryConfigurer) bean;
return bean;
@@ -462,7 +462,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
registerEndpoints();
if (this.messageHistoryConfigurer != null) {
this.registerBeanInstance(this.messageHistoryConfigurer,
IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER);
IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER_BEAN_NAME);
}
}

12
src/reference/docbook/channel.xml
View File

@@ -509,6 +509,18 @@ payload to an Integer.
For more information regarding Payload Type Conversion, please read <xref linkend="payload-type-conversion"/>.
</para>
</note>
<para>
Beginning with <emphasis>version 4.0</emphasis>, the <code>integrationConversionService</code> is invoked
by the <classname>DefaultDatatypeChannelMessageConverter</classname>, which looks up the conversion
service in the application context. To use a different conversion technique, you can specify the
<code>message-converter</code> attribute on the channel. This must be a reference to a
<interfacename>MessageConverter</interfacename> implementation.
Only the <code>fromMessage</code> method is used, which provides the
converter with access to the message headers (for example if the conversion might need information
from the headers, such as <code>content-type</code>). The method can return just the converted
payload, or a full <classname>Message</classname> object. If the latter, the converter must
be careful to copy all the headers from the inbound message.
</para>
</section>
<section id="channel-configuration-queuechannel">

8
src/reference/docbook/whats-new.xml
View File

@@ -89,5 +89,13 @@
considered for outbound messages. For more information see <xref linkend="jms-header-mapping"/>.
</para>
</section>
<section id="4.0-datatype-channel">
<title>Datatype Channels</title>
<para>
You can now specify a <interfacename>MessageConverter</interfacename> to be used when converting
(if necessary) payloads to one of the accepted <code>datatype</code>s in a Datatype channel.
For more information see <xref linkend="channel-datatype-channel"/>.
</para>
</section>
</section>
</chapter>