resolves INT-582

Includes refactoring of router hierarchy to avoid duplication of channel resolution logic
Subclasses of AbstractChannelNameResolvingMessageRouter can return MessageChannel , MessageChannel[], collection ... in addition to String
MethodInvokingRouter now extends AbstractChannelNameResolvingMessageRouter and is simplified as a result
This commit is contained in:
Jonas Partner
2009-02-22 10:34:43 +00:00
parent 4cb87799b0
commit fca5ae5501
16 changed files with 221 additions and 140 deletions

View File

@@ -19,6 +19,7 @@ package org.springframework.integration.config;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.router.AbstractChannelNameResolvingMessageRouter;
import org.springframework.integration.router.AbstractMessageRouter;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.util.Assert;
@@ -28,13 +29,17 @@ import org.springframework.util.StringUtils;
* Factory bean for creating a Message Router.
*
* @author Mark Fisher
* @author Jonas Partner
*/
public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean {
private volatile ChannelResolver channelResolver;
private volatile MessageChannel defaultOutputChannel;
private volatile boolean resolutionRequired;
private volatile Boolean ignoreChannelNameResolutionFailures;
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
@@ -43,6 +48,14 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean {
public void setDefaultOutputChannel(MessageChannel defaultOutputChannel) {
this.defaultOutputChannel = defaultOutputChannel;
}
public void setResolutionRequired(boolean resolutionRequired){
this.resolutionRequired = resolutionRequired;
}
public void setIgnoreChannelNameResolutionFailures(boolean ignoreChannelNameResolutionFailures){
this.ignoreChannelNameResolutionFailures = ignoreChannelNameResolutionFailures;
}
@Override
protected MessageHandler createHandler(Object targetObject, String targetMethodName) {
@@ -51,6 +64,12 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean {
if (this.defaultOutputChannel != null) {
router.setDefaultOutputChannel(this.defaultOutputChannel);
}
if(ignoreChannelNameResolutionFailures != null ){
Assert.isTrue(router instanceof AbstractChannelNameResolvingMessageRouter,
"Ignore channel name resolution failures can only set on rotuers extending " + AbstractChannelNameResolvingMessageRouter.class.getName());
((AbstractChannelNameResolvingMessageRouter)router).setIgnoreChannelNameResolutionFailures(ignoreChannelNameResolutionFailures);
}
router.setResolutionRequired(resolutionRequired);
return router;
}

View File

@@ -49,6 +49,8 @@ public class RouterParser extends AbstractConsumerEndpointParser {
resolverBuilder.getBeanDefinition(), parserContext.getRegistry());
builder.addPropertyReference("channelResolver", resolverBeanName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "default-output-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "resolution-required");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-channel-name-resolution-failures");
return builder;
}

View File

@@ -788,6 +788,8 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="method" type="xsd:string" />
<xsd:attribute name="resolution-required" type="xsd:boolean" />
<xsd:attribute name="ignore-channel-name-resolution-failures" type="xsd:boolean" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

View File

@@ -17,12 +17,15 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
@@ -30,12 +33,14 @@ import org.springframework.integration.core.MessagingException;
import org.springframework.util.Assert;
/**
* A base class for router implementations that return only
* the channel name(s) rather than {@link MessageChannel} instances.
* A base class for router implementations that return only the channel name(s)
* rather than {@link MessageChannel} instances.
*
* @author Mark Fisher
* @author Jonas Partner
*/
public abstract class AbstractChannelNameResolvingMessageRouter extends AbstractMessageRouter implements BeanFactoryAware, InitializingBean {
public abstract class AbstractChannelNameResolvingMessageRouter extends
AbstractMessageRouter implements BeanFactoryAware, InitializingBean {
private volatile ChannelResolver channelResolver;
@@ -45,6 +50,7 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract
private volatile BeanFactory beanFactory;
private volatile boolean ignoreChannelNameResolutionFailures;
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
@@ -62,45 +68,97 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract
this.beanFactory = beanFactory;
}
public void setIgnoreChannelNameResolutionFailures(
boolean ignoreChannelNameResolutionFailures) {
this.ignoreChannelNameResolutionFailures = ignoreChannelNameResolutionFailures;
}
public void afterPropertiesSet() {
if (this.channelResolver == null) {
Assert.notNull(beanFactory, "either a ChannelResolver or BeanFactory is required");
this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory);
Assert.notNull(beanFactory,
"either a ChannelResolver or BeanFactory is required");
this.channelResolver = new BeanFactoryChannelResolver(
this.beanFactory);
}
}
protected MessageChannel resolveChannelForName(String channelName,
Message<?> message) {
Assert.state(this.channelResolver != null,
"unable to resolve channel names, no ChannelResolver available");
MessageChannel channel = null;
try {
channel = this.channelResolver.resolveChannelName(channelName);
} catch (ChannelResolutionException e) {
if (!ignoreChannelNameResolutionFailures)
throw new MessagingException(message,
"failed to resolve channel name '" + channelName + "'", e);
}
if(channel == null && !ignoreChannelNameResolutionFailures){
throw new MessagingException(message,
"failed to resolve channel name '" + channelName + "'");
}
return channel;
}
@Override
protected final Collection<MessageChannel> determineTargetChannels(Message<?> message) {
protected Collection<MessageChannel> determineTargetChannels(
Message<?> message) {
this.afterPropertiesSet();
Collection<MessageChannel> channels = new ArrayList<MessageChannel>();
String[] channelNames = this.determineTargetChannelNames(message);
if (channelNames == null) {
Collection<Object> channelsReturned = this.getChannelIndicatorList(message);
addToCollection(channels, channelsReturned, message);
if (channels == null) {
return null;
}
for (String channelName : channelNames) {
if (channelName != null) {
Assert.state(this.channelResolver != null,
"unable to resolve channel names, no ChannelResolver available");
if (this.prefix != null) {
channelName = this.prefix + channelName;
return channels;
}
protected void addToCollection(Collection<MessageChannel> channels, Collection<?> channelIndicators, Message<?> message){
if(channelIndicators == null){
return;
}
for (Object channelIndicator : channelIndicators) {
if (channelIndicator == null) {
continue;
} else if (channelIndicator instanceof String) {
addChannelFromString(channels, (String)channelIndicator, message);
} else if (channelIndicator instanceof MessageChannel){
channels.add((MessageChannel)channelIndicator);
} else if (channelIndicator instanceof Collection){
addToCollection(channels, (Collection<?>)channelIndicator, message);
} else if (channelIndicator instanceof MessageChannel[]) {
channels.addAll(Arrays.asList((MessageChannel[]) channelIndicator));
} else if (channelIndicator instanceof String[]) {
for (String indicatorName : (String[]) channelIndicator) {
addChannelFromString(channels, indicatorName, message);
}
if (this.suffix != null) {
channelName = channelName + suffix;
}
MessageChannel channel = this.channelResolver.resolveChannelName(channelName);
if (channel == null) {
throw new MessagingException(message,
"failed to resolve channel name '" + channelName + "'");
}
channels.add(channel);
}else {
throw new MessagingException("unsupported return type for router [" + channelIndicator.getClass() + "]");
}
}
return channels;
}
protected void addChannelFromString(Collection<MessageChannel> channels, String channelName, Message<?> message){
if (this.prefix != null) {
channelName = this.prefix + channelName;
}
if (this.suffix != null) {
channelName = channelName + suffix;
}
MessageChannel channel = resolveChannelForName(channelName,message);
if (channel != null) {
channels.add(channel);
}
}
/**
* Subclasses must implement this method to return the channel name(s).
* Subclasses must implement this method to return the channel indicators.
*/
protected abstract String[] determineTargetChannelNames(Message<?> message);
protected abstract List<Object> getChannelIndicatorList(Message<?> message);
}

View File

@@ -16,6 +16,10 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.core.Message;
/**
@@ -27,9 +31,13 @@ import org.springframework.integration.core.Message;
public abstract class AbstractSingleChannelNameRouter extends AbstractChannelNameResolvingMessageRouter {
@Override
protected final String[] determineTargetChannelNames(Message<?> message) {
String channelName = this.determineTargetChannelName(message);
return (channelName != null) ? new String[] { channelName } : null;
protected final List<Object> getChannelIndicatorList(Message<?> message) {
List<Object> channelList = new ArrayList<Object>();
String channelName = determineTargetChannelName(message);
if(channelName != null){
channelList.add(channelName);
}
return channelList;
}
/**

View File

@@ -18,17 +18,13 @@ package org.springframework.integration.router;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.handler.MessageMappingMethodInvoker;
import org.springframework.util.Assert;
/**
* A Message Router that invokes the specified method on the given object. The
@@ -39,12 +35,11 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class MethodInvokingRouter extends AbstractMessageRouter {
public class MethodInvokingRouter extends AbstractChannelNameResolvingMessageRouter {
private final MessageMappingMethodInvoker invoker;
private volatile ChannelResolver channelResolver;
public MethodInvokingRouter(Object object, Method method) {
this.invoker = new MessageMappingMethodInvoker(object, method);
@@ -58,68 +53,12 @@ public class MethodInvokingRouter extends AbstractMessageRouter {
this.invoker = new MessageMappingMethodInvoker(object, Router.class);
}
/**
* Provide the {@link ChannelResolver} strategy to use for methods that
* return a channel name rather than a {@link MessageChannel} instance.
*/
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
}
@Override
protected final Collection<MessageChannel> determineTargetChannels(Message<?> message) {
protected List<Object> getChannelIndicatorList(Message<?> message) {
Object result = this.invoker.invokeMethod(message);
if (result == null) {
return null;
}
List<MessageChannel> channels = new ArrayList<MessageChannel>();
if (result instanceof Collection) {
for (Object next : (Collection<?>) result) {
this.addChannel(next, channels);
}
}
else if (result instanceof MessageChannel[]) {
channels.addAll(Arrays.asList((MessageChannel[]) result));
}
else if (result instanceof String[]) {
for (String channelName : (String[]) result) {
this.addChannel(channelName, channels);
}
}
else if (result instanceof MessageChannel) {
channels.add((MessageChannel) result);
}
else if (result instanceof String) {
this.addChannel((String) result, channels);
}
else {
throw new IllegalStateException(
"router method must return type 'MessageChannel' or 'String'");
}
return channels;
}
private void addChannel(Object channelOrName, List<MessageChannel> channels) {
if (channelOrName == null) {
return;
}
if (channelOrName instanceof MessageChannel) {
channels.add((MessageChannel) channelOrName);
}
else if (channelOrName instanceof String) {
String channelName = (String) channelOrName;
Assert.state(this.channelResolver != null,
"unable to resolve channel names, no ChannelResolver available");
MessageChannel channel = this.channelResolver.resolveChannelName(channelName);
if (channel == null) {
throw new MessagingException("failed to resolve channel name '" + channelName + "'");
}
channels.add(channel);
}
else {
throw new MessagingException("unsupported return type for router [" + channelOrName.getClass() + "]");
}
List<Object> asList = new ArrayList<Object>();
asList.add(result);
return asList;
}
}

View File

@@ -25,11 +25,10 @@ import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.TestChannelResolver;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.TestChannelResolver;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;

View File

@@ -19,13 +19,15 @@ package org.springframework.integration.router;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.TestChannelResolver;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.StringMessage;
import org.springframework.util.CollectionUtils;
/**
* @author Mark Fisher
@@ -35,8 +37,9 @@ public class MultiChannelRouterTests {
@Test
public void routeWithChannelMapping() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
return new String[] {"channel1", "channel2"};
@SuppressWarnings("unchecked")
public List<Object> getChannelIndicatorList(Message<?> message) {
return CollectionUtils.arrayToList(new String[] {"channel1", "channel2"});
}
};
QueueChannel channel1 = new QueueChannel();
@@ -60,8 +63,9 @@ public class MultiChannelRouterTests {
@Test(expected = MessagingException.class)
public void channelNameLookupFailure() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
return new String[] {"noSuchChannel"};
@SuppressWarnings("unchecked")
public List<Object> getChannelIndicatorList(Message<?> message) {
return CollectionUtils.arrayToList(new String[] {"noSuchChannel"} );
}
};
TestChannelResolver channelResolver = new TestChannelResolver();
@@ -73,8 +77,9 @@ public class MultiChannelRouterTests {
@Test(expected = MessagingException.class)
public void channelMappingNotAvailable() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
return new String[] {"noSuchChannel"};
@SuppressWarnings("unchecked")
public List<Object> getChannelIndicatorList(Message<?> message) {
return CollectionUtils.arrayToList(new String[] {"noSuchChannel"});
}
};
Message<String> message = new StringMessage("test");

View File

@@ -18,11 +18,11 @@ package org.springframework.integration.router;
import static org.junit.Assert.assertEquals;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.junit.Test;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.TestChannelResolver;
@@ -31,6 +31,7 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.StringMessage;
import org.springframework.util.CollectionUtils;
/**
* @author Mark Fisher
@@ -86,7 +87,7 @@ public class RouterTests {
@Test
public void nullChannelNameArrayIgnoredByDefault() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
protected List<Object> getChannelIndicatorList(Message<?> message) {
return null;
}
};
@@ -99,7 +100,7 @@ public class RouterTests {
@Test(expected = MessageDeliveryException.class)
public void nullChannelNameArrayThrowsExceptionWhenResolutionRequired() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
protected List<Object> getChannelIndicatorList(Message<?> message) {
return null;
}
};
@@ -114,8 +115,8 @@ public class RouterTests {
@Test
public void emptyChannelNameArrayIgnoredByDefault() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
return new String[] {};
protected List<Object> getChannelIndicatorList(Message<?> message) {
return new ArrayList<Object>();
}
};
TestChannelResolver channelResolver = new TestChannelResolver();
@@ -127,8 +128,9 @@ public class RouterTests {
@Test(expected = MessageDeliveryException.class)
public void emptyChannelNameArrayThrowsExceptionWhenResolutionRequired() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
return new String[] {};
@SuppressWarnings("unchecked")
protected List<Object> getChannelIndicatorList(Message<?> message) {
return CollectionUtils.arrayToList(new String[] {});
}
};
TestChannelResolver channelResolver = new TestChannelResolver();
@@ -151,8 +153,9 @@ public class RouterTests {
@Test(expected = MessagingException.class)
public void channelMappingIsRequiredWhenResolvingChannelNamesWithMultiChannelRouter() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
return new String[] { "notImportant" };
@SuppressWarnings("unchecked")
protected List<Object> getChannelIndicatorList(Message<?> message){
return CollectionUtils.arrayToList(new String[] { "notImportant" });
}
};
router.handleMessage(new StringMessage("this should fail"));
@@ -177,8 +180,9 @@ public class RouterTests {
@Test
public void beanFactoryWithMultiChannelRouter() {
AbstractChannelNameResolvingMessageRouter router = new AbstractChannelNameResolvingMessageRouter() {
public String[] determineTargetChannelNames(Message<?> message) {
return new String[] { "testChannel" };
@SuppressWarnings("unchecked")
protected List<Object> getChannelIndicatorList(Message<?> message) {
return CollectionUtils.arrayToList(new String[] { "testChannel" });
}
};
QueueChannel testChannel = new QueueChannel();

View File

@@ -24,17 +24,19 @@ import java.util.Collection;
import java.util.Collections;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.router.AbstractMessageRouter;
/**
* @author Mark Fisher
* @author Jonas Partner
*/
public class RouterParserTests {
@@ -97,7 +99,25 @@ public class RouterParserTests {
assertNotNull(result);
assertEquals("test-annotation", result.getPayload());
}
@Test(expected=MessageDeliveryException.class)
public void testResolutionRequired() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"routerParserTests.xml", this.getClass());
context.start();
MessageChannel input = (MessageChannel) context.getBean("inputForRouterRequiringResolution");
input.send(new GenericMessage<Integer>(3));
}
@Test
public void testIgnoreChannelNameResolutionFailures() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"routerParserTests.xml", this.getClass());
context.start();
MessageChannel input = (MessageChannel) context.getBean("ignoreChannelNameResolutionFailuresInput");
input.send(new StringMessage("channelThatDoesNotExist"));
}
public static class TestRouterImplementation extends AbstractMessageRouter {
@@ -127,5 +147,16 @@ public class RouterParserTests {
return this.channel;
}
}
public static class ReturnStringPassedInAsChannelNameRouter {
@Router
public String route(Message<?> message) {
return (String)message.getPayload();
}
}
}

View File

@@ -54,5 +54,14 @@
<beans:bean id="annotated" class="org.springframework.integration.router.config.RouterParserTests$AnnotatedTestRouterBean">
<beans:constructor-arg ref="output4"/>
</beans:bean>
<channel id="inputForRouterRequiringResolution"/>
<router id="resolutionRequiredRouter" ref="pojo" method="route" input-channel="inputForRouterRequiringResolution" resolution-required="true" />
<channel id="ignoreChannelNameResolutionFailuresInput"/>
<beans:bean id="payloadAsChannelNameRouter" class="org.springframework.integration.router.config.RouterParserTests$ReturnStringPassedInAsChannelNameRouter" />
<router id="ignoreChannelNameResolutionFailuresRouter" ref="payloadAsChannelNameRouter" input-channel="ignoreChannelNameResolutionFailuresInput" ignore-channel-name-resolution-failures="true" />
</beans:beans>