From 5b6a96c34fcd98d57e8814261c39950e9c19b0af Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 13 Oct 2010 15:14:33 -0400 Subject: [PATCH] INT-1377, polishing, added test for expression based router dynamics --- .../router/AbstractMessageRouter.java | 257 +++++++++--------- .../ErrorMessageExceptionTypeRouter.java | 4 +- .../integration/router/PayloadTypeRouter.java | 14 +- .../router/RecipientListRouter.java | 44 +-- .../config/RouterWithMappingTests-context.xml | 2 +- .../router/config/RouterWithMappingTests.java | 15 + 6 files changed, 174 insertions(+), 162 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index 5183f017ad..b91370bdc0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java @@ -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 channelIdentifierMap; + protected volatile Map channelIdentifierMap = new ConcurrentHashMap(); /** * 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 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 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 determineTargetChannels(Message message) { - this.afterPropertiesSet(); - Collection channels = new ArrayList(); - Collection 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 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 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(); - } + 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 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 determineTargetChannels(Message message) { + this.afterPropertiesSet(); + Collection channels = new ArrayList(); + Collection 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 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 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 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() + "]"); + } + } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java index 50548a1ec2..1dcb2d051a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/ErrorMessageExceptionTypeRouter.java @@ -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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java index 3293db5845..6951895fe7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java @@ -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 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java index 00812ab592..ee0ea2cdf1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java @@ -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 determineTargetChannels(Message message) { -// List channels = new ArrayList(); -// List recipientList = this.recipients; -// for (Recipient recipient : recipientList) { -// if (recipient.accept(message)) { -// channels.add(recipient.getChannel()); -// } -// } -// return channels; -// } - - + + @Override + protected List getChannelIndicatorList(Message message) { + List channels = new ArrayList(); + List 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 getChannelIndicatorList(Message message) { - List channels = new ArrayList(); - List recipientList = this.recipients; - for (Recipient recipient : recipientList) { - if (recipient.accept(message)) { - channels.add(recipient.getChannel()); - } - } - return channels; - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests-context.xml index fcce0a6a52..ebb93cb347 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests-context.xml @@ -19,7 +19,7 @@ - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests.java index e7af29a277..0d6fb5bd6e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/config/RouterWithMappingTests.java @@ -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