diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 7039cf428e..307076cccd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -22,13 +22,18 @@ import java.util.concurrent.CopyOnWriteArrayList; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanNameAware; +import org.springframework.core.convert.ConversionService; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessagingException; import org.springframework.integration.core.MessageHistory.ComponentType; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * Base class for {@link MessageChannel} implementations providing common @@ -38,12 +43,18 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public abstract class AbstractMessageChannel implements MessageChannel, BeanNameAware { +public abstract class AbstractMessageChannel implements MessageChannel, BeanFactoryAware, BeanNameAware { private final Log logger = LogFactory.getLog(this.getClass()); private volatile String name; + private volatile Class[] datatypes = new Class[] { Object.class }; + + private volatile ConversionService conversionService; + + private volatile BeanFactory beanFactory; + private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); @@ -62,6 +73,21 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName return this.name; } + /** + * Specify the Message payload datatype(s) supported by this channel. If a + * payload type does not match directly, but the 'conversionService' is + * available, then type conversion will be attempted in the order of the + * elements provided in this array. + *

+ * If this property is not set explicitly, any Message payload type will be + * accepted. + * @see #setConversionService(ConversionService) + */ + public void setDatatypes(Class... datatypes) { + this.datatypes = (datatypes != null && datatypes.length > 0) + ? datatypes : new Class[] { Object.class }; + } + /** * Set the list of channel interceptors. This will clear any existing * interceptors. @@ -77,6 +103,35 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName this.interceptors.add(interceptor); } + /** + * Specify the {@link ConversionService} to use when trying to convert to + * one of this channel's supported datatypes for a Message whose payload + * does not already match. If this property is not set explicitly but + * the channel is managed within a context, it will fallback to a bean + * named "conversionService" defined within that context. + */ + public void setConversionService(ConversionService conversionService) { + this.conversionService = conversionService; + } + + public void setBeanFactory(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + } + + private ConversionService getConversionService() { + if (this.conversionService == null && this.beanFactory != null) { + if (this.beanFactory.containsBean("conversionService")) { + this.conversionService = this.beanFactory.getBean("conversionService", ConversionService.class); + } + else if (logger.isWarnEnabled()) { + logger.warn("Unable to attempt conversion of Message payload types. " + + "Datatype channel has no explicit ConversionService reference, " + + "and there is no 'conversionService' bean within the context."); + } + } + return this.conversionService; + } + /** * Exposes the interceptor list for subclasses. */ @@ -114,6 +169,8 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName */ public final boolean send(Message message, long timeout) { Assert.notNull(message, "message must not be null"); + Assert.notNull(message.getPayload(), "message payload must not be null"); + message = this.convertPayloadIfNecessary(message); message.getHeaders().getHistory().add(ComponentType.channel, this.getName()); message = this.interceptors.preSend(message, this); if (message == null) { @@ -133,6 +190,29 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName } } + private Message convertPayloadIfNecessary(Message message) { + // first pass checks if the payload type already matches any of the datatypes + for (Class datatype : this.datatypes) { + if (datatype.isAssignableFrom(message.getPayload().getClass())) { + return message; + } + } + // second pass applies conversion if possible, attempting datatypes in order + ConversionService conversionService = this.getConversionService(); + if (conversionService != null) { + 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(); + } + } + } + throw new MessageDeliveryException(message, "Channel '" + this.getName() + + "' expected one of the following datataypes [" + + StringUtils.arrayToCommaDelimitedString(this.datatypes) + + "], but received [" + message.getPayload().getClass() + "]"); + } + public String toString() { return (this.name != null) ? this.name : super.toString(); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java index c8a609318f..073893c780 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -18,14 +18,12 @@ package org.springframework.integration.config.xml; import org.w3c.dom.Element; -import org.springframework.beans.factory.config.RuntimeBeanReference; -import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.ManagedList; -import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; @@ -51,20 +49,15 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser } String datatypeAttr = element.getAttribute("datatype"); if (StringUtils.hasText(datatypeAttr)) { - String[] datatypes = StringUtils.commaDelimitedListToStringArray(datatypeAttr); - RootBeanDefinition selectorDef = new RootBeanDefinition(); - selectorDef.setBeanClassName(IntegrationNamespaceUtils.BASE_PACKAGE + ".selector.PayloadTypeSelector"); - selectorDef.getConstructorArgumentValues().addGenericArgumentValue(datatypes); - String selectorBeanName = parserContext.getReaderContext().generateBeanName(selectorDef); - BeanComponentDefinition selectorComponent = new BeanComponentDefinition(selectorDef, selectorBeanName); - parserContext.registerBeanComponent(selectorComponent); - RootBeanDefinition interceptorDef = new RootBeanDefinition(); - interceptorDef.setBeanClassName(IntegrationNamespaceUtils.BASE_PACKAGE + ".channel.interceptor.MessageSelectingInterceptor"); - interceptorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(selectorBeanName)); - String interceptorBeanName = parserContext.getReaderContext().generateBeanName(interceptorDef); - BeanComponentDefinition interceptorComponent = new BeanComponentDefinition(interceptorDef, interceptorBeanName); - parserContext.registerBeanComponent(interceptorComponent); - interceptors.add(new RuntimeBeanReference(interceptorBeanName)); + // TODO: remove this once the editor fallback is working (3.0 GA) + // it should be replaced with: builder.addPropertyValue("datatypes", datatypeAttr); + String[] classnames = StringUtils.commaDelimitedListToStringArray(datatypeAttr); + Class[] datatypes = new Class[classnames.length]; + int i = 0; + for (String classname : classnames) { + datatypes[i++] = ClassUtils.resolveClassName(classname.trim(), this.getClass().getClassLoader()); + } + builder.addPropertyValue("datatypes", datatypes); } builder.addPropertyValue("interceptors", interceptors); return builder.getBeanDefinition(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java new file mode 100644 index 0000000000..7ce7d3f8ec --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java @@ -0,0 +1,86 @@ +/* + * Copyright 2002-2009 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.channel; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.util.Date; + +import org.junit.Test; + +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessagingException; +import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class DatatypeChannelTests { + + @Test + public void supportedType() { + MessageChannel channel = createChannel(String.class); + assertTrue(channel.send(new StringMessage("test"))); + } + + @Test(expected = MessageDeliveryException.class) + public void unsupportedTypeAndNoConversionService() { + MessageChannel channel = createChannel(Integer.class); + channel.send(new StringMessage("test")); + } + + @Test + public void multipleTypes() { + MessageChannel channel = createChannel(String.class, Integer.class); + assertTrue(channel.send(new StringMessage("test1"))); + assertTrue(channel.send(new GenericMessage(2))); + Exception exception = null; + try { + channel.send(new GenericMessage(new Date())); + } + catch (MessageDeliveryException e) { + exception = e; + } + assertNotNull(exception); + } + + @Test + public void subclassOfAcceptedType() { + MessageChannel channel = createChannel(RuntimeException.class); + assertTrue(channel.send(new ErrorMessage(new MessagingException("test")))); + } + + @Test(expected = MessageDeliveryException.class) + public void superclassOfAcceptedTypeNotAccepted() { + MessageChannel channel = createChannel(RuntimeException.class); + channel.send(new ErrorMessage(new Exception("test"))); + } + + + private static QueueChannel createChannel(Class ... datatypes) { + QueueChannel channel = new QueueChannel(); + channel.setBeanName("testChannel"); + channel.setDatatypes(datatypes); + return channel; + } + +}