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

INT-1377, polishing, added test for expression based router dynamics

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-10-13 15:14:33 -04:00
parent 713baedf43
commit 5b6a96c34f
6 changed files with 174 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

257
spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
View File

@@ -37,10 +37,11 @@ import org.springframework.integration.support.channel.BeanFactoryChannelResolve
import org.springframework.integration.support.channel.ChannelResolutionException;
import org.springframework.integration.support.channel.ChannelResolver;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
/**
* Base class for Message Routers.
* Base class for all Message Routers.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
@@ -65,7 +66,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
private volatile boolean ignoreChannelNameResolutionFailures;
protected volatile Map<String, String> channelIdentifierMap;
protected volatile Map<String, String> channelIdentifierMap = new ConcurrentHashMap<String, String>();
/**
* Specify the {@link ChannelResolver} strategy to use.
@@ -97,139 +98,26 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
public void setIgnoreChannelNameResolutionFailures(boolean ignoreChannelNameResolutionFailures) {
this.ignoreChannelNameResolutionFailures = ignoreChannelNameResolutionFailures;
}
@Override
public void onInit() {
BeanFactory beanFactory = this.getBeanFactory();
if (this.channelResolver == null && beanFactory != null) {
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
}
private 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 (!this.ignoreChannelNameResolutionFailures) {
throw new MessagingException(message,
"failed to resolve channel name '" + channelName + "'", e);
}
}
if (channel == null && !this.ignoreChannelNameResolutionFailures) {
throw new MessagingException(message,
"failed to resolve channel name '" + channelName + "'");
}
return channel;
}
private void addChannelFromString(Collection<MessageChannel> channels, String channelIdentifier, Message<?> message) {
if (channelIdentifier.indexOf(',') != -1) {
for (String name : StringUtils.commaDelimitedListToStringArray(channelIdentifier)) {
addChannelFromString(channels, name, message);
}
return;
}
if (this.prefix != null) {
channelIdentifier = this.prefix + channelIdentifier;
}
if (this.suffix != null) {
channelIdentifier = channelIdentifier + suffix;
}
/*
* Some routers due to their complex nature will already resolve 'channelIdentifier'
* to 'channelName' (e.g., PTR, EMETR)
*/
String channelName = channelIdentifier;
if (channelIdentifierMap != null && channelIdentifierMap.containsKey(channelIdentifier)){
channelName = channelIdentifierMap.get(channelIdentifier);
}
if (this.channelResolver != null){
MessageChannel channel = resolveChannelForName(channelName, message);
if (channel != null) {
channels.add(channel);
}
}
}
private 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 MessageChannel) {
channels.add((MessageChannel) channelIndicator);
}
else if (channelIndicator instanceof MessageChannel[]) {
channels.addAll(Arrays.asList((MessageChannel[]) channelIndicator));
}
else if (channelIndicator instanceof String) {
addChannelFromString(channels, (String) channelIndicator, message);
}
else if (channelIndicator instanceof String[]) {
for (String indicatorName : (String[]) channelIndicator) {
addChannelFromString(channels, indicatorName, message);
}
}
else if (channelIndicator instanceof Collection) {
addToCollection(channels, (Collection<?>) channelIndicator, message);
}
else if (this.getRequiredConversionService().canConvert(channelIndicator.getClass(), String.class)) {
addChannelFromString(channels,
this.getConversionService().convert(channelIndicator, String.class), message);
}
else {
throw new MessagingException(
"unsupported return type for router [" + channelIndicator.getClass() + "]");
}
}
}
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
this.afterPropertiesSet();
Collection<MessageChannel> channels = new ArrayList<MessageChannel>();
Collection<Object> channelsReturned = this.getChannelIndicatorList(message);
addToCollection(channels, channelsReturned, message);
return channels;
}
protected ConversionService getRequiredConversionService() {
if (this.getConversionService() == null) {
this.setConversionService(ConversionServiceFactory.createDefaultConversionService());
}
return this.getConversionService();
}
public Map<String, String> getChannelIdentifierMap() {
return channelIdentifierMap;
}
/**
* Allows you to set the map which will map channel identifiers to channel names.
* Channel names will be resolve via {@link ChannelResolver}
* @param channelIdentifierMap
*/
public void setChannelIdentifierMap(Map<String, String> channelIdentifierMap) {
this.channelIdentifierMap = channelIdentifierMap;
this.channelIdentifierMap.clear();
this.channelIdentifierMap.putAll(channelIdentifierMap);
}
public synchronized void setChannelMapping(String channelIdentifier, String channelName){
if (channelIdentifierMap == null){
channelIdentifierMap = new ConcurrentHashMap<String, String>();
}
public void setChannelMapping(String channelIdentifier, String channelName){
this.channelIdentifierMap.put(channelIdentifier, channelName);
}
public synchronized void removeChannelMapping(String channelIdentifier){
/**
* Removes channel mapping for a give channel identifier
* @param channelIdentifier
*/
public void removeChannelMapping(String channelIdentifier){
this.channelIdentifierMap.remove(channelIdentifier);
}
/**
* Subclasses must implement this method to return the channel indicators.
*/
protected abstract List<Object> getChannelIndicatorList(Message<?> message);
/**
* Set the default channel where Messages should be sent if channel resolution fails to return any channels. If no
@@ -287,6 +175,33 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
protected MessagingTemplate getMessagingTemplate() {
return this.messagingTemplate;
}
@Override
public void onInit() {
BeanFactory beanFactory = this.getBeanFactory();
if (this.channelResolver == null && beanFactory != null) {
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
}
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
this.afterPropertiesSet();
Collection<MessageChannel> channels = new ArrayList<MessageChannel>();
Collection<Object> channelsReturned = this.getChannelIndicatorList(message);
addToCollection(channels, channelsReturned, message);
return channels;
}
protected ConversionService getRequiredConversionService() {
if (this.getConversionService() == null) {
this.setConversionService(ConversionServiceFactory.createDefaultConversionService());
}
return this.getConversionService();
}
/**
* Subclasses must implement this method to return the channel indicators.
*/
protected abstract List<Object> getChannelIndicatorList(Message<?> message);
@Override
protected void handleMessageInternal(Message<?> message) {
@@ -324,4 +239,90 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
}
}
}
private 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 (!this.ignoreChannelNameResolutionFailures) {
throw new MessagingException(message,
"failed to resolve channel name '" + channelName + "'", e);
}
}
if (channel == null && !this.ignoreChannelNameResolutionFailures) {
throw new MessagingException(message,
"failed to resolve channel name '" + channelName + "'");
}
return channel;
}
private void addChannelFromString(Collection<MessageChannel> channels, String channelIdentifier, Message<?> message) {
if (channelIdentifier.indexOf(',') != -1) {
for (String name : StringUtils.commaDelimitedListToStringArray(channelIdentifier)) {
addChannelFromString(channels, name, message);
}
return;
}
if (this.prefix != null) {
channelIdentifier = this.prefix + channelIdentifier;
}
if (this.suffix != null) {
channelIdentifier = channelIdentifier + suffix;
}
/*
* Some routers due to their complex nature will already resolve 'channelIdentifier'
* to 'channelName' (e.g., PTR, EMETR)
*/
String channelName = channelIdentifier;
if (!CollectionUtils.isEmpty(channelIdentifierMap) && channelIdentifierMap.containsKey(channelIdentifier)){
channelName = channelIdentifierMap.get(channelIdentifier);
}
if (this.channelResolver != null){
MessageChannel channel = resolveChannelForName(channelName, message);
if (channel != null) {
channels.add(channel);
}
}
}
private 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 MessageChannel) {
channels.add((MessageChannel) channelIndicator);
}
else if (channelIndicator instanceof MessageChannel[]) {
channels.addAll(Arrays.asList((MessageChannel[]) channelIndicator));
}
else if (channelIndicator instanceof String) {
addChannelFromString(channels, (String) channelIndicator, message);
}
else if (channelIndicator instanceof String[]) {
for (String indicatorName : (String[]) channelIndicator) {
addChannelFromString(channels, indicatorName, message);
}
}
else if (channelIndicator instanceof Collection) {
addToCollection(channels, (Collection<?>) channelIndicator, message);
}
else if (this.getRequiredConversionService().canConvert(channelIndicator.getClass(), String.class)) {
addChannelFromString(channels,
this.getConversionService().convert(channelIndicator, String.class), message);
}
else {
throw new MessagingException(
"unsupported return type for router [" + channelIndicator.getClass() + "]");
}
}
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2010 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.
@@ -29,7 +29,7 @@ import org.springframework.integration.MessageChannel;
*
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
*/
public class ErrorMessageExceptionTypeRouter extends AbstractMessageRouter {
@Override

14
spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java
View File

@@ -24,6 +24,7 @@ import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
/**
@@ -34,13 +35,22 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
*/
public class PayloadTypeRouter extends AbstractMessageRouter {
/**
* Will select the most appropriate channel name matching channel identifiers
* which are fully qualifies class name to type available while traversing payload type.
* To resolve ties and conflicts (e.g., Serializable and String) it will match:
* 1. Type name to channel identifier else...
* 2. Name of the subclass of the type to channel identifier elc...
* 3. Name of the Interface of the type to channel identifier while also
* preferring direct interface over in-direct subclass
*
*/
@Override
protected List<Object> getChannelIndicatorList(Message<?> message) {
Class<?> firstInterfaceMatch = null;
Class<?> type = message.getPayload().getClass();
while (type != null && channelIdentifierMap != null) {
while (type != null && !CollectionUtils.isEmpty(channelIdentifierMap)) {
Class<?>[] interfaces = type.getInterfaces();
// first try to find a match amongst the interfaces and also check if there is more then one
for (Class<?> interfase : interfaces) {

44
spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -17,8 +17,8 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
@@ -51,6 +51,7 @@ import org.springframework.util.Assert;
* solution.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public class RecipientListRouter extends AbstractMessageRouter implements InitializingBean {
@@ -88,20 +89,19 @@ public class RecipientListRouter extends AbstractMessageRouter implements Initia
public final void onInit() {
Assert.notEmpty(this.recipients, "a non-empty recipient list is required");
}
// @Override
// protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
// List<MessageChannel> channels = new ArrayList<MessageChannel>();
// List<Recipient> recipientList = this.recipients;
// for (Recipient recipient : recipientList) {
// if (recipient.accept(message)) {
// channels.add(recipient.getChannel());
// }
// }
// return channels;
// }
@Override
protected List<Object> getChannelIndicatorList(Message<?> message) {
List<Object> channels = new ArrayList<Object>();
List<Recipient> recipientList = this.recipients;
for (Recipient recipient : recipientList) {
if (recipient.accept(message)) {
channels.add(recipient.getChannel());
}
}
return channels;
}
public static class Recipient {
private final MessageChannel channel;
@@ -125,18 +125,4 @@ public class RecipientListRouter extends AbstractMessageRouter implements Initia
return this.channel;
}
}
@Override
protected List<Object> getChannelIndicatorList(Message<?> message) {
List<Object> channels = new ArrayList<Object>();
List<Recipient> recipientList = this.recipients;
for (Recipient recipient : recipientList) {
if (recipient.accept(message)) {
channels.add(recipient.getChannel());
}
}
return channels;
}
}

2
spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests-context.xml
View File

@@ -19,7 +19,7 @@
<queue capacity="1" />
</channel>
<router input-channel="expressionRouter" expression="payload.name"
<router id="spelRouter" input-channel="expressionRouter" expression="payload.name"
default-output-channel="defaultChannelForExpression"
ignore-channel-name-resolution-failures="true">
<mapping value="foo" channel="fooChannelForExpression"/>

15
spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests.java
View File

@@ -23,10 +23,14 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.router.AbstractMessageRouter;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -39,6 +43,10 @@ public class RouterWithMappingTests {
@Autowired
private MessageChannel expressionRouter;
@Autowired
@Qualifier("spelRouter")
private ConsumerEndpointFactoryBean spelRouter;
@Autowired
private MessageChannel pojoRouter;
@@ -79,6 +87,13 @@ public class RouterWithMappingTests {
assertNotNull(defaultChannelForExpression.receive(0));
assertNull(fooChannelForExpression.receive(0));
assertNull(barChannelForExpression.receive(0));
// validate dynamics
AbstractMessageRouter router = (AbstractMessageRouter) TestUtils.getPropertyValue(spelRouter, "handler");
router.setChannelMapping("baz", "fooChannelForExpression");
expressionRouter.send(message3);
assertNull(defaultChannelForExpression.receive(10));
assertNotNull(fooChannelForExpression.receive(10));
assertNull(barChannelForExpression.receive(0));
}
@Test