diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index fe69a7ae91..e528ae0677 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -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. + *

+ * Note: 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(); + } } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationRegistrar.java index 4a7af30840..473bab1aae 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationRegistrar.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationRegistrar.java @@ -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()); + } + + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageHistoryRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageHistoryRegistrar.java index 5c47be3ed4..063f32ff03 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageHistoryRegistrar.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageHistoryRegistrar.java @@ -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 componentNamePatternsSet = new ManagedSet(); 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java index a653d1d67f..8455297bde 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java @@ -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"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java index 030fa49fbd..b7de144c9e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java @@ -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. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java new file mode 100644 index 0000000000..32fe96afac --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java @@ -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"); + } + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd index e23973e8fd..643511ae5e 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd @@ -498,6 +498,29 @@ + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java index 96e93ded2a..2d1cbfa98e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java @@ -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("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.TRUE))); } @@ -80,11 +87,14 @@ public class DatatypeChannelTests { QueueChannel channel = createChannel(Integer.class); GenericConversionService conversionService = new DefaultConversionService(); conversionService.addConverter(new Converter() { + @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.TRUE))); assertEquals(1, channel.receive().getPayload()); } @@ -93,6 +103,7 @@ public class DatatypeChannelTests { public void conversionServiceBeanUsedByDefault() { GenericApplicationContext context = new GenericApplicationContext(); Converter converter = new Converter() { + @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.TRUE))); assertEquals(1, channel.receive().getPayload()); + context.close(); } @Test public void conversionServiceReferenceOverridesDefault() { GenericApplicationContext context = new GenericApplicationContext(); Converter defaultConverter = new Converter() { + @Override public Integer convert(Boolean source) { return source ? 1 : 0; } }; GenericConversionService customConversionService = new DefaultConversionService(); customConversionService.addConverter(new Converter() { + @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.TRUE))); assertEquals(99, channel.receive().getPayload()); + context.close(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index b946c0cf91..2b7ff02f5d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java @@ -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("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(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("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(123))); assertTrue(channel.send(new GenericMessage(123.45))); + assertTrue(channel.send(new GenericMessage(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(123))); assertTrue(channel.send(new GenericMessage("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(true)); + channel.send(new GenericMessage(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("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 lowPriorityMessage = MessageBuilder.withPayload("low").setPriority(-14).build(); Message 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("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(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 { + + @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; + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml index a752b936e1..0f68088a85 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml @@ -6,6 +6,10 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> + + + + @@ -27,15 +31,18 @@ - + + + - + diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index 94cbb49994..229003127f 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -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); } } diff --git a/src/reference/docbook/channel.xml b/src/reference/docbook/channel.xml index 633b1e1beb..efbfc35da9 100644 --- a/src/reference/docbook/channel.xml +++ b/src/reference/docbook/channel.xml @@ -509,6 +509,18 @@ payload to an Integer. For more information regarding Payload Type Conversion, please read . + + Beginning with version 4.0, the integrationConversionService is invoked + by the DefaultDatatypeChannelMessageConverter, which looks up the conversion + service in the application context. To use a different conversion technique, you can specify the + message-converter attribute on the channel. This must be a reference to a + MessageConverter implementation. + Only the fromMessage 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 content-type). The method can return just the converted + payload, or a full Message object. If the latter, the converter must + be careful to copy all the headers from the inbound message. +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index dea71d9fd0..f3ae43f607 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -89,5 +89,13 @@ considered for outbound messages. For more information see .
+
+ Datatype Channels + + You can now specify a MessageConverter to be used when converting + (if necessary) payloads to one of the accepted datatypes in a Datatype channel. + For more information see . + +