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

INT-912 Channels with "datatype" now delegate to the "conversionService" bean if available.

Browse Source
This commit is contained in:
Mark Fisher
2009-12-10 22:00:50 +00:00
parent 2e72122373
commit 34802e77fd
3 changed files with 179 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -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.
* <p>
* 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();
}

29
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java
View File

@@ -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();

86
org.springframework.integration/src/test/java/org/springframework/integration/channel/DatatypeChannelTests.java Normal file
View File

@@ -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<Integer>(2)));
Exception exception = null;
try {
channel.send(new GenericMessage<Date>(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;
}
}