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

RouterEndpoint now delegates directly to a single ChannelResolver strategy. This removes the extra level of indirection that was provided by the Router interface. Also, instead of providing multiple ChannelResolver strategy interfaces, the name-resolving and single-channel implementations are now available as abstract base classes.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-04 01:55:32 +00:00
parent b3a155de94
commit 40fc9c207d
30 changed files with 617 additions and 794 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
org.springframework.integration/src/main/java/org/springframework/integration/config/RouterParser.java
View File

@@ -17,7 +17,7 @@
package org.springframework.integration.config;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.integration.router.MethodInvokingChannelResolver;
import org.springframework.integration.router.RouterEndpoint;
/**
@@ -34,7 +34,7 @@ public class RouterParser extends AbstractEndpointParser {
@Override
protected Class<?> getMethodInvokingAdapterClass() {
return MethodInvokingRouter.class;
return MethodInvokingChannelResolver.class;
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java
View File

@@ -24,7 +24,7 @@ import org.springframework.integration.annotation.Router;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.integration.router.MethodInvokingChannelResolver;
import org.springframework.integration.router.RouterEndpoint;
import org.springframework.util.StringUtils;
@@ -42,13 +42,13 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP
@Override
protected Object createMethodInvokingAdapter(Object bean, Method method, Router annotation) {
return new MethodInvokingRouter(bean, method);
return new MethodInvokingChannelResolver(bean, method);
}
@Override
protected AbstractEndpoint createEndpoint(Object adapter) {
if (adapter instanceof MethodInvokingRouter) {
return new RouterEndpoint((MethodInvokingRouter) adapter);
if (adapter instanceof MethodInvokingChannelResolver) {
return new RouterEndpoint((MethodInvokingChannelResolver) adapter);
}
return null;
}

56
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractChannelResolver.java Normal file
View File

@@ -0,0 +1,56 @@
/*
* Copyright 2002-2008 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.router;
import java.util.Collection;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
/**
* @author Mark Fisher
*/
public abstract class AbstractChannelResolver implements ChannelResolver, ChannelRegistryAware {
private volatile ChannelRegistry channelRegistry;
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
protected MessageChannel lookupChannel(String channelName, boolean required) {
MessageChannel channel = null;
if (channelName != null) {
if (this.channelRegistry == null) {
throw new ConfigurationException("unable to resolve channels, no ChannelRegistry available");
}
channel = this.channelRegistry.lookupChannel(channelName);
}
if (channel == null && required) {
throw new MessagingException("unable to resolve channel '" + channelName + "'");
}
return channel;
}
public abstract Collection<MessageChannel> resolveChannels(Message<?> message);
}

24
org.springframework.integration/src/main/java/org/springframework/integration/router/Router.java → org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMultiChannelNameResolver.java
View File

@@ -16,15 +16,31 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.Collection;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* Strategy interface for routing a Message to one or more channels.
*
* @author Mark Fisher
*/
public interface Router {
public abstract class AbstractMultiChannelNameResolver extends AbstractChannelResolver {
boolean route(Message<?> message);
@Override
public Collection<MessageChannel> resolveChannels(Message<?> message) {
Collection<MessageChannel> channels = new ArrayList<MessageChannel>();
String[] channelNames = this.resolveChannelNames(message);
if (channelNames == null) {
return null;
}
for (String channelName : channelNames) {
MessageChannel channel = this.lookupChannel(channelName, true);
channels.add(channel);
}
return channels;
}
protected abstract String[] resolveChannelNames(Message<?> message);
}

92
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractRouter.java
View File

@@ -1,92 +0,0 @@
/*
* Copyright 2002-2008 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.router;
import java.util.Collection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.MessagingException;
/**
* Base class for message router implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractRouter implements Router, ChannelRegistryAware {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile ChannelRegistry channelRegistry;
private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
protected ChannelRegistry getChannelRegistry() {
return this.channelRegistry;
}
public final boolean route(Message<?> message) {
Collection<?> results = this.resolveChannels(message);
if (results == null || results.isEmpty()) {
return false;
}
boolean sent = false;
for (Object channelOrName : results) {
MessageTarget target = null;
if (channelOrName == null) {
continue;
}
if (channelOrName instanceof MessageTarget) {
target = (MessageTarget) channelOrName;
}
else if (channelOrName instanceof String) {
if (this.channelRegistry == null) {
throw new MessagingException(message, "router has no ChannelRegistry");
}
target = this.channelRegistry.lookupChannel((String) channelOrName);
}
else {
throw new MessagingException(message, "unsupported return type for router [" + channelOrName.getClass() + "]");
}
if (target == null) {
throw new MessageDeliveryException(message, "unable to resolve channel for '" + channelOrName + "'");
}
this.messageExchangeTemplate.send(message, target);
sent = true;
}
return sent;
}
/**
* Subclasses must implement this method to return 0 or more MessageChannel
* instances or channel names to which the given Message should be routed.
*/
protected abstract Collection<?> resolveChannels(Message<?> message);
}

14
org.springframework.integration/src/main/java/org/springframework/integration/router/MultiChannelResolver.java → org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractSingleChannelNameResolver.java
View File

@@ -16,18 +16,20 @@
package org.springframework.integration.router;
import java.util.List;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* Strategy interface for content-based routing to multiple channels.
*
* @author Mark Fisher
*/
public interface MultiChannelResolver {
public abstract class AbstractSingleChannelNameResolver extends AbstractSingleChannelResolver {
List<MessageChannel> resolve(Message<?> message);
@Override
protected MessageChannel resolveChannel(Message<?> message) {
String channelName = this.resolveChannelName(message);
return this.lookupChannel(channelName, true);
}
protected abstract String resolveChannelName(Message<?> message);
}

16
org.springframework.integration/src/main/java/org/springframework/integration/router/ChannelNameResolver.java → org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractSingleChannelResolver.java
View File

@@ -16,15 +16,23 @@
package org.springframework.integration.router;
import java.util.Collection;
import java.util.Collections;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* Strategy interface for content-based routing to a channel name.
*
* @author Mark Fisher
*/
public interface ChannelNameResolver {
public abstract class AbstractSingleChannelResolver extends AbstractChannelResolver {
String resolve(Message<?> message);
public Collection<MessageChannel> resolveChannels(Message<?> message) {
MessageChannel channel = this.resolveChannel(message);
return (channel != null) ?
Collections.singletonList(channel) : null;
}
protected abstract MessageChannel resolveChannel(Message<?> message);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/router/ChannelResolver.java
View File

@@ -16,16 +16,18 @@
package org.springframework.integration.router;
import java.util.Collection;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* Strategy interface for content-based routing to a channel instance.
* Strategy interface for content-based routing.
*
* @author Mark Fisher
*/
public interface ChannelResolver {
MessageChannel resolve(Message<?> message);
Collection<MessageChannel> resolveChannels(Message<?> message);
}

45
org.springframework.integration/src/main/java/org/springframework/integration/router/MethodInvokingRouter.java → org.springframework.integration/src/main/java/org/springframework/integration/router/MethodInvokingChannelResolver.java
View File

@@ -27,25 +27,26 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMappingMethodInvoker;
import org.springframework.integration.message.MessagingException;
/**
* A {@link Router} implementation that invokes the specified method
* on the given object. The method's return value may be a single
* MessageChannel instance, a single String to be interpreted as
* a channel name, or a Collection (or Array) of either type.
* A {@link ChannelResolver} implementation that invokes the specified method
* on the given object. The method's return value may be a single MessageChannel
* instance, a single String to be interpreted as a channel name, or a Collection
* (or Array) of either type.
*
* @author Mark Fisher
*/
public class MethodInvokingRouter extends AbstractRouter implements InitializingBean {
public class MethodInvokingChannelResolver extends AbstractChannelResolver implements InitializingBean {
private final MessageMappingMethodInvoker invoker;
public MethodInvokingRouter(Object object, Method method) {
public MethodInvokingChannelResolver(Object object, Method method) {
this.invoker = new MessageMappingMethodInvoker(object, method);
}
public MethodInvokingRouter(Object object, String methodName) {
public MethodInvokingChannelResolver(Object object, String methodName) {
this.invoker = new MessageMappingMethodInvoker(object, methodName);
}
@@ -55,26 +56,30 @@ public class MethodInvokingRouter extends AbstractRouter implements Initializing
}
@Override
protected Collection<?> resolveChannels(Message<?> message) {
public final Collection<MessageChannel> resolveChannels(Message<?> message) {
Object result = this.invoker.invokeMethod(message);
if (result == null) {
return null;
}
List<Object> channels = new ArrayList<Object>();
List<MessageChannel> channels = new ArrayList<MessageChannel>();
if (result instanceof Collection) {
channels.addAll((Collection<?>) result);
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[]) {
channels.addAll(Arrays.asList((String[]) result));
for (String channelName : (String[]) result) {
this.addChannel(channelName, channels);
}
}
else if (result instanceof MessageChannel) {
channels.add((MessageChannel) result);
}
else if (result instanceof String) {
channels.add(result);
this.addChannel((String) result, channels);
}
else {
throw new ConfigurationException(
@@ -83,4 +88,20 @@ public class MethodInvokingRouter extends AbstractRouter implements Initializing
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) {
MessageChannel channel = this.lookupChannel((String) channelOrName, true);
channels.add(channel);
}
else {
throw new MessagingException("unsupported return type for router [" + channelOrName.getClass() + "]");
}
}
}

30
org.springframework.integration/src/main/java/org/springframework/integration/router/MultiChannelNameResolver.java
View File

@@ -1,30 +0,0 @@
/*
* Copyright 2002-2008 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.router;
import org.springframework.integration.message.Message;
/**
* Strategy interface for content-based routing to multiple channel names.
*
* @author Mark Fisher
*/
public interface MultiChannelNameResolver {
String[] resolve(Message<?> message);
}

68
org.springframework.integration/src/main/java/org/springframework/integration/router/MultiChannelRouter.java
View File

@@ -1,68 +0,0 @@
/*
* Copyright 2002-2008 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.router;
import java.util.Arrays;
import java.util.Collection;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.util.ObjectUtils;
/**
* A router implementation for sending to potentially multiple {@link MessageChannel MessageChannels}.
* Requires either a {@link MultiChannelResolver} or {@link MultiChannelNameResolver} strategy instance.
*
* @author Mark Fisher
*/
public class MultiChannelRouter extends AbstractRouter implements InitializingBean {
private volatile MultiChannelResolver channelResolver;
private volatile MultiChannelNameResolver channelNameResolver;
public void setChannelResolver(MultiChannelResolver channelResolver) {
this.channelResolver = channelResolver;
}
public void setChannelNameResolver(MultiChannelNameResolver channelNameResolver) {
this.channelNameResolver = channelNameResolver;
}
public void afterPropertiesSet() {
if (!(this.channelResolver != null ^ this.channelNameResolver != null)) {
throw new ConfigurationException(
"exactly one of 'channelResolver' or 'channelNameResolver' must be provided");
}
}
@Override
public Collection<?> resolveChannels(Message<?> message) {
if (this.channelResolver != null) {
return this.channelResolver.resolve(message);
}
String[] channelNames = this.channelNameResolver.resolve(message);
if (ObjectUtils.isEmpty(channelNames)) {
return null;
}
return Arrays.asList(channelNames);
}
}

27
org.springframework.integration/src/main/java/org/springframework/integration/router/PayloadTypeRouter.java → org.springframework.integration/src/main/java/org/springframework/integration/router/PayloadTypeChannelResolver.java
View File

@@ -24,39 +24,24 @@ import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* A router implementation that resolves the {@link MessageChannel} based on the
* {@link Message Message's} payload type.
* A ChannelResolver implementation that resolves the {@link MessageChannel} based
* on the {@link Message Message's} payload type.
*
* @author Mark Fisher
*/
public class PayloadTypeRouter extends SingleChannelRouter {
public class PayloadTypeChannelResolver extends AbstractSingleChannelResolver {
private Map<Class<?>, MessageChannel> channelMappings = new ConcurrentHashMap<Class<?>, MessageChannel>();
private MessageChannel defaultChannel;
public PayloadTypeRouter() {
this.setChannelResolver(new PayloadTypeChannelResolver());
}
public void setChannelMappings(Map<Class<?>, MessageChannel> channelMappings) {
Assert.notNull(channelMappings, "'channelMappings' must not be null");
this.channelMappings = channelMappings;
}
public void setDefaultChannel(MessageChannel defaultChannel) {
this.defaultChannel = defaultChannel;
}
private class PayloadTypeChannelResolver implements ChannelResolver {
public MessageChannel resolve(Message<?> message) {
MessageChannel channel = channelMappings.get(message.getPayload().getClass());
return channel != null ? channel : defaultChannel;
}
@Override
protected MessageChannel resolveChannel(Message<?> message) {
return this.channelMappings.get(message.getPayload().getClass());
}
}

71
org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListChannelResolver.java Normal file
View File

@@ -0,0 +1,71 @@
/*
* Copyright 2002-2008 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.router;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* A {@link ChannelResolver} implementation that routes to a statically
* configured list of recipients. The recipients are provided either as a list
* of {@link MessageChannel} instances or as a String array of channel names.
* For dynamic recipient lists, either implement {@link ChannelResolver} or
* extend the {@link AbstractMultiChannelNameResolver} base class.
*
* @author Mark Fisher
*/
public class RecipientListChannelResolver extends AbstractChannelResolver implements InitializingBean {
private volatile List<MessageChannel> channels;
private volatile String[] channelNames;
public void setChannels(List<MessageChannel> channels) {
this.channels = channels;
}
public void setChannelNames(String[] channelNames) {
this.channelNames = channelNames;
}
public void afterPropertiesSet() {
if ((this.channels != null && this.channelNames != null)
|| (this.channels == null && this.channelNames == null)) {
throw new ConfigurationException("either 'channels' or 'channelNames' should be provided, but not both");
}
}
@Override
public Collection<MessageChannel> resolveChannels(Message<?> message) {
if (this.channels == null && this.channelNames != null) {
List<MessageChannel> resolved = new ArrayList<MessageChannel>();
for (String channelName : channelNames) {
resolved.add(this.lookupChannel(channelName, true));
}
return resolved;
}
return this.channels;
}
}

72
org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -1,72 +0,0 @@
/*
* Copyright 2002-2008 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.router;
import java.util.List;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* A simple extension of {@link MultiChannelRouter} that routes to a statically
* configured list of recipients. The recipients are provided either as a list
* of {@link MessageChannel} instances or as a String array of channel names.
* For dynamic recipient lists, implement either {@link MultiChannelResolver} or
* {@link MultiChannelNameResolver} and then explicitly configure an instance of
* {@link MultiChannelRouter}.
*
* @author Mark Fisher
*/
public class RecipientListRouter extends MultiChannelRouter {
public void setChannelNames(String[] channelNames) {
this.setChannelNameResolver(new RecipientListChannelNameResolver(channelNames));
}
public void setChannels(List<MessageChannel> channels) {
this.setChannelResolver(new RecipientListChannelResolver(channels));
}
private static class RecipientListChannelResolver implements MultiChannelResolver {
private List<MessageChannel> channels;
RecipientListChannelResolver(List<MessageChannel> channels) {
this.channels = channels;
}
public List<MessageChannel> resolve(Message<?> message) {
return this.channels;
}
}
private static class RecipientListChannelNameResolver implements MultiChannelNameResolver {
private String[] channelNames;
RecipientListChannelNameResolver(String[] channelNames) {
this.channelNames = channelNames;
}
public String[] resolve(Message<?> message) {
return this.channelNames;
}
}
}

45
org.springframework.integration/src/main/java/org/springframework/integration/router/RootCauseErrorMessageRouter.java → org.springframework.integration/src/main/java/org/springframework/integration/router/RootCauseErrorMessageChannelResolver.java
View File

@@ -24,52 +24,39 @@ import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* A router implementation that resolves the {@link MessageChannel} for messages
* whose payload is an Exception. The channel resolution is based upon the most
* specific cause of the error for which a channel-mapping exists.
* A ChannelResolver implementation that resolves the {@link MessageChannel} for
* messages whose payload is an Exception. The channel resolution is based upon the
* most specific cause of the error for which a channel-mapping exists.
*
* @author Mark Fisher
*/
public class RootCauseErrorMessageRouter extends SingleChannelRouter {
public class RootCauseErrorMessageChannelResolver extends AbstractSingleChannelResolver {
private Map<Class<? extends Throwable>, MessageChannel> channelMappings =
new ConcurrentHashMap<Class<? extends Throwable>, MessageChannel>();
private MessageChannel defaultChannel;
public RootCauseErrorMessageRouter() {
this.setChannelResolver(new RootCauseResolver());
}
public void setChannelMappings(Map<Class<? extends Throwable>, MessageChannel> channelMappings) {
Assert.notNull(channelMappings, "'channelMappings' must not be null");
this.channelMappings = channelMappings;
}
public void setDefaultChannel(MessageChannel defaultChannel) {
this.defaultChannel = defaultChannel;
}
private class RootCauseResolver implements ChannelResolver {
public MessageChannel resolve(Message<?> message) {
MessageChannel channel = null;
Object payload = message.getPayload();
if (payload != null && (payload instanceof Throwable)) {
Throwable mostSpecificCause = (Throwable) payload;
while (mostSpecificCause != null) {
MessageChannel mappedChannel = channelMappings.get(mostSpecificCause.getClass());
if (mappedChannel != null) {
channel = mappedChannel;
}
mostSpecificCause = mostSpecificCause.getCause();
@Override
protected MessageChannel resolveChannel(Message<?> message) {
MessageChannel channel = null;
Object payload = message.getPayload();
if (payload != null && (payload instanceof Throwable)) {
Throwable mostSpecificCause = (Throwable) payload;
while (mostSpecificCause != null) {
MessageChannel mappedChannel = this.channelMappings.get(mostSpecificCause.getClass());
if (mappedChannel != null) {
channel = mappedChannel;
}
mostSpecificCause = mostSpecificCause.getCause();
}
return channel != null ? channel : defaultChannel;
}
return channel;
}
}

29
org.springframework.integration/src/main/java/org/springframework/integration/router/RouterEndpoint.java
View File

@@ -16,12 +16,15 @@
package org.springframework.integration.router;
import java.util.Collection;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.util.Assert;
/**
@@ -29,24 +32,26 @@ import org.springframework.util.Assert;
*/
public class RouterEndpoint extends AbstractEndpoint {
private final Router router;
private final ChannelResolver channelResolver;
private volatile MessageChannel defaultOutputChannel;
private volatile boolean resolutionRequired;
private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
public RouterEndpoint(Router router) {
Assert.notNull(router, "router must not be null");
this.router = router;
public RouterEndpoint(ChannelResolver channelResolver) {
Assert.notNull(channelResolver, "ChannelResolver must not be null");
this.channelResolver = channelResolver;
}
@Override
public void setChannelRegistry(ChannelRegistry channelRegistry) {
super.setChannelRegistry(channelRegistry);
if (this.router instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.router).setChannelRegistry(channelRegistry);
if (this.channelResolver instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.channelResolver).setChannelRegistry(channelRegistry);
}
}
@@ -74,7 +79,17 @@ public class RouterEndpoint extends AbstractEndpoint {
@Override
protected boolean sendInternal(Message<?> message) {
boolean sent = this.router.route(message);
boolean sent = false;
Collection<MessageChannel> results = this.channelResolver.resolveChannels(message);
if (results != null) {
for (MessageChannel channel : results) {
if (channel != null) {
if (this.messageExchangeTemplate.send(message, channel)) {
sent = true;
}
}
}
}
if (!sent) {
if (this.defaultOutputChannel != null) {
sent = this.getMessageExchangeTemplate().send(message, this.defaultOutputChannel);

67
org.springframework.integration/src/main/java/org/springframework/integration/router/SingleChannelRouter.java
View File

@@ -1,67 +0,0 @@
/*
* Copyright 2002-2008 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.router;
import java.util.Collection;
import java.util.Collections;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* A router implementation for sending to at most one {@link MessageChannel}.
* Requires either a {@link ChannelResolver} or {@link ChannelNameResolver}
* strategy instance.
*
* @author Mark Fisher
*/
public class SingleChannelRouter extends AbstractRouter implements InitializingBean {
private ChannelResolver channelResolver;
private ChannelNameResolver channelNameResolver;
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
}
public void setChannelNameResolver(ChannelNameResolver channelNameResolver) {
this.channelNameResolver = channelNameResolver;
}
public void afterPropertiesSet() {
if (!(this.channelResolver != null ^ this.channelNameResolver != null)) {
throw new ConfigurationException(
"exactly one of 'channelResolver' or 'channelNameResolver' must be provided");
}
}
@Override
protected Collection<?> resolveChannels(Message<?> message) {
Object result = (this.channelResolver != null)
? this.channelResolver.resolve(message)
: this.channelNameResolver.resolve(message);
if (result == null) {
return null;
}
return Collections.singletonList(result);
}
}

206
org.springframework.integration/src/test/java/org/springframework/integration/router/MethodInvokingRouterTests.java
View File

@@ -53,10 +53,11 @@ public class MethodInvokingRouterTests {
channelRegistry.registerChannel(barChannel);
SingleChannelNameRoutingTestBean testBean = new SingleChannelNameRoutingTestBean();
Method routingMethod = testBean.getClass().getMethod("routePayload", String.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
router.setChannelRegistry(channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new GenericMessage<String>("bar");
assertTrue(router.route(message));
assertTrue(endpoint.send(message));
Message<?> replyMessage = barChannel.receive();
assertNotNull(replyMessage);
assertEquals(message, replyMessage);
@@ -69,10 +70,11 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(barChannel);
SingleChannelNameRoutingTestBean testBean = new SingleChannelNameRoutingTestBean();
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routePayload");
router.setChannelRegistry(channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routePayload");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new GenericMessage<String>("bar");
assertTrue(router.route(message));
assertTrue(endpoint.send(message));
Message<?> replyMessage = barChannel.receive();
assertNotNull(replyMessage);
assertEquals(message, replyMessage);
@@ -89,11 +91,12 @@ public class MethodInvokingRouterTests {
channelRegistry.registerChannel(barChannel);
SingleChannelNameRoutingTestBean testBean = new SingleChannelNameRoutingTestBean();
Method routingMethod = testBean.getClass().getMethod("routeByHeader", String.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
router.setChannelRegistry(channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = MessageBuilder.fromPayload("bar")
.setHeader("targetChannel", "foo").build();
assertTrue(router.route(message));
assertTrue(endpoint.send(message));
Message<?> fooReply = fooChannel.receive(0);
Message<?> barReply = barChannel.receive(0);
assertNotNull(fooReply);
@@ -105,26 +108,29 @@ public class MethodInvokingRouterTests {
public void failsWhenRequiredHeaderIsNotProvided() throws Exception {
SingleChannelNameRoutingTestBean testBean = new SingleChannelNameRoutingTestBean();
Method routingMethod = testBean.getClass().getMethod("routeByHeader", String.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
router.route(new GenericMessage<String>("testing"));
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.send(new GenericMessage<String>("testing"));
}
@Test
public void channelNameResolutionByMessageConfiguredByMethodReference() throws Exception {
SingleChannelNameRoutingTestBean testBean = new SingleChannelNameRoutingTestBean();
Method routingMethod = testBean.getClass().getMethod("routeMessage", Message.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestChannelNameResolutionByMessage(router);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestChannelNameResolutionByMessage(endpoint);
}
@Test
public void channelNameResolutionByMessageConfiguredByMethodName() {
SingleChannelNameRoutingTestBean testBean = new SingleChannelNameRoutingTestBean();
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routeMessage");
this.doTestChannelNameResolutionByMessage(router);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routeMessage");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestChannelNameResolutionByMessage(endpoint);
}
private void doTestChannelNameResolutionByMessage(MethodInvokingRouter router) {
private void doTestChannelNameResolutionByMessage(RouterEndpoint endpoint) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
@@ -132,19 +138,19 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1 = fooChannel.receive(0);
assertNotNull(result1);
assertEquals("foo", result1.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2 = barChannel.receive(0);
assertNotNull(result2);
assertEquals("bar", result2.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -152,19 +158,21 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
SingleChannelInstanceRoutingTestBean testBean = new SingleChannelInstanceRoutingTestBean(channelRegistry);
Method routingMethod = testBean.getClass().getMethod("routePayload", String.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestChannelInstanceResolutionByPayload(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestChannelInstanceResolutionByPayload(endpoint, channelRegistry);
}
@Test
public void channelInstanceResolutionByPayloadConfiguredByMethodName() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
SingleChannelInstanceRoutingTestBean testBean = new SingleChannelInstanceRoutingTestBean(channelRegistry);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routePayload");
this.doTestChannelInstanceResolutionByPayload(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routePayload");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestChannelInstanceResolutionByPayload(endpoint, channelRegistry);
}
private void doTestChannelInstanceResolutionByPayload(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestChannelInstanceResolutionByPayload(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
@@ -174,16 +182,16 @@ public class MethodInvokingRouterTests {
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
assertTrue(router.route(fooMessage));
endpoint.setChannelRegistry(channelRegistry);
assertTrue(endpoint.send(fooMessage));
Message<?> result1 = fooChannel.receive(0);
assertNotNull(result1);
assertEquals("foo", result1.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2 = barChannel.receive(0);
assertNotNull(result2);
assertEquals("bar", result2.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -191,38 +199,40 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
SingleChannelInstanceRoutingTestBean testBean = new SingleChannelInstanceRoutingTestBean(channelRegistry);
Method routingMethod = testBean.getClass().getMethod("routeMessage", Message.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestChannelInstanceResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestChannelInstanceResolutionByMessage(endpoint, channelRegistry);
}
@Test
public void channelInstanceResolutionByMessageConfiguredByMethodName() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
SingleChannelInstanceRoutingTestBean testBean = new SingleChannelInstanceRoutingTestBean(channelRegistry);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routeMessage");
this.doTestChannelInstanceResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routeMessage");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestChannelInstanceResolutionByMessage(endpoint, channelRegistry);
}
private void doTestChannelInstanceResolutionByMessage(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestChannelInstanceResolutionByMessage(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1 = fooChannel.receive(0);
assertNotNull(result1);
assertEquals("foo", result1.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2 = barChannel.receive(0);
assertNotNull(result2);
assertEquals("bar", result2.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -230,44 +240,46 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelNameRoutingTestBean testBean = new MultiChannelNameRoutingTestBean();
Method routingMethod = testBean.getClass().getMethod("routePayload", String.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestMultiChannelNameResolutionByPayload(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelNameResolutionByPayload(endpoint, channelRegistry);
}
@Test
public void multiChannelNameResolutionByPayloadConfiguredByMethodName() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelNameRoutingTestBean testBean = new MultiChannelNameRoutingTestBean();
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routePayload");
this.doTestMultiChannelNameResolutionByPayload(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routePayload");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelNameResolutionByPayload(endpoint, channelRegistry);
}
private void doTestMultiChannelNameResolutionByPayload(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestMultiChannelNameResolutionByPayload(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1a = fooChannel.receive(0);
Message<?> result1b = barChannel.receive(0);
assertNotNull(result1a);
assertEquals("foo", result1a.getPayload());
assertNotNull(result1b);
assertEquals("foo", result1b.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2a = fooChannel.receive(0);
Message<?> result2b = barChannel.receive(0);
assertNotNull(result2a);
assertEquals("bar", result2a.getPayload());
assertNotNull(result2b);
assertEquals("bar", result2b.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -275,44 +287,46 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelNameRoutingTestBean testBean = new MultiChannelNameRoutingTestBean();
Method routingMethod = testBean.getClass().getMethod("routeMessage", Message.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestMultiChannelNameResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelNameResolutionByMessage(endpoint, channelRegistry);
}
@Test
public void multiChannelNameResolutionByMessageConfiguredByMethodName() throws Exception {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelNameRoutingTestBean testBean = new MultiChannelNameRoutingTestBean();
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routeMessage");
this.doTestMultiChannelNameResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routeMessage");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelNameResolutionByMessage(endpoint, channelRegistry);
}
private void doTestMultiChannelNameResolutionByMessage(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestMultiChannelNameResolutionByMessage(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1a = fooChannel.receive(0);
assertNotNull(result1a);
assertEquals("foo", result1a.getPayload());
Message<?> result1b = barChannel.receive(0);
assertNotNull(result1b);
assertEquals("foo", result1b.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2a = fooChannel.receive(0);
assertNotNull(result2a);
assertEquals("bar", result2a.getPayload());
Message<?> result2b = barChannel.receive(0);
assertNotNull(result2b);
assertEquals("bar", result2b.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -320,44 +334,46 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelNameRoutingTestBean testBean = new MultiChannelNameRoutingTestBean();
Method routingMethod = testBean.getClass().getMethod("routeMessageToArray", Message.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestMultiChannelNameArrayResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelNameArrayResolutionByMessage(endpoint, channelRegistry);
}
@Test
public void multiChannelNameArrayResolutionByMessageConfiguredByMethodName() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelNameRoutingTestBean testBean = new MultiChannelNameRoutingTestBean();
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routeMessageToArray");
this.doTestMultiChannelNameArrayResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routeMessageToArray");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelNameArrayResolutionByMessage(endpoint, channelRegistry);
}
private void doTestMultiChannelNameArrayResolutionByMessage(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestMultiChannelNameArrayResolutionByMessage(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1a = fooChannel.receive(0);
assertNotNull(result1a);
assertEquals("foo", result1a.getPayload());
Message<?> result1b = barChannel.receive(0);
assertNotNull(result1b);
assertEquals("foo", result1b.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2a = fooChannel.receive(0);
assertNotNull(result2a);
assertEquals("bar", result2a.getPayload());
Message<?> result2b = barChannel.receive(0);
assertNotNull(result2b);
assertEquals("bar", result2b.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -365,44 +381,46 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelInstanceRoutingTestBean testBean = new MultiChannelInstanceRoutingTestBean(channelRegistry);
Method routingMethod = testBean.getClass().getMethod("routePayload", String.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestMultiChannelListResolutionByPayload(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelListResolutionByPayload(endpoint, channelRegistry);
}
@Test
public void multiChannelListResolutionByPayloadConfiguredByMethodName() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelInstanceRoutingTestBean testBean = new MultiChannelInstanceRoutingTestBean(channelRegistry);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routePayload");
this.doTestMultiChannelListResolutionByPayload(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routePayload");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelListResolutionByPayload(endpoint, channelRegistry);
}
private void doTestMultiChannelListResolutionByPayload(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestMultiChannelListResolutionByPayload(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1a = fooChannel.receive(0);
Message<?> result1b = barChannel.receive(0);
assertNotNull(result1a);
assertEquals("foo", result1a.getPayload());
assertNotNull(result1b);
assertEquals("foo", result1b.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2a = fooChannel.receive(0);
Message<?> result2b = barChannel.receive(0);
assertNotNull(result2a);
assertEquals("bar", result2a.getPayload());
assertNotNull(result2b);
assertEquals("bar", result2b.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -410,44 +428,46 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelInstanceRoutingTestBean testBean = new MultiChannelInstanceRoutingTestBean(channelRegistry);
Method routingMethod = testBean.getClass().getMethod("routeMessage", Message.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestMultiChannelListResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelListResolutionByMessage(endpoint, channelRegistry);
}
@Test
public void multiChannelListResolutionByMessageConfiguredByMethodName() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelInstanceRoutingTestBean testBean = new MultiChannelInstanceRoutingTestBean(channelRegistry);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routeMessage");
this.doTestMultiChannelListResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routeMessage");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelListResolutionByMessage(endpoint, channelRegistry);
}
private void doTestMultiChannelListResolutionByMessage(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestMultiChannelListResolutionByMessage(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1a = fooChannel.receive(0);
Message<?> result1b = barChannel.receive(0);
assertNotNull(result1a);
assertEquals("foo", result1a.getPayload());
assertNotNull(result1b);
assertEquals("foo", result1b.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2a = fooChannel.receive(0);
Message<?> result2b = barChannel.receive(0);
assertNotNull(result2a);
assertEquals("bar", result2a.getPayload());
assertNotNull(result2b);
assertEquals("bar", result2b.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}
@Test
@@ -455,44 +475,46 @@ public class MethodInvokingRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelInstanceRoutingTestBean testBean = new MultiChannelInstanceRoutingTestBean(channelRegistry);
Method routingMethod = testBean.getClass().getMethod("routeMessageToArray", Message.class);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, routingMethod);
this.doTestMultiChannelArrayResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, routingMethod);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelArrayResolutionByMessage(endpoint, channelRegistry);
}
@Test
public void multiChannelArrayResolutionByMessageConfiguredByMethodName() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelInstanceRoutingTestBean testBean = new MultiChannelInstanceRoutingTestBean(channelRegistry);
MethodInvokingRouter router = new MethodInvokingRouter(testBean, "routeMessageToArray");
this.doTestMultiChannelArrayResolutionByMessage(router, channelRegistry);
MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(testBean, "routeMessageToArray");
RouterEndpoint endpoint = new RouterEndpoint(resolver);
this.doTestMultiChannelArrayResolutionByMessage(endpoint, channelRegistry);
}
private void doTestMultiChannelArrayResolutionByMessage(MethodInvokingRouter router, ChannelRegistry channelRegistry) {
private void doTestMultiChannelArrayResolutionByMessage(RouterEndpoint endpoint, ChannelRegistry channelRegistry) {
QueueChannel fooChannel = new QueueChannel();
QueueChannel barChannel = new QueueChannel();
fooChannel.setBeanName("foo-channel");
barChannel.setBeanName("bar-channel");
channelRegistry.registerChannel(fooChannel);
channelRegistry.registerChannel(barChannel);
router.setChannelRegistry(channelRegistry);
endpoint.setChannelRegistry(channelRegistry);
Message<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> badMessage = new StringMessage("bad");
assertTrue(router.route(fooMessage));
assertTrue(endpoint.send(fooMessage));
Message<?> result1a = fooChannel.receive(0);
Message<?> result1b = barChannel.receive(0);
assertNotNull(result1a);
assertEquals("foo", result1a.getPayload());
assertNotNull(result1b);
assertEquals("foo", result1b.getPayload());
assertTrue(router.route(barMessage));
assertTrue(endpoint.send(barMessage));
Message<?> result2a = fooChannel.receive(0);
Message<?> result2b = barChannel.receive(0);
assertNotNull(result2a);
assertEquals("bar", result2a.getPayload());
assertNotNull(result2b);
assertEquals("bar", result2b.getPayload());
assertFalse(router.route(badMessage));
assertFalse(endpoint.send(badMessage));
}

73
org.springframework.integration/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java
View File

@@ -24,13 +24,12 @@ import java.util.List;
import org.junit.Test;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.DefaultChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.StringMessage;
/**
@@ -42,19 +41,17 @@ public class MultiChannelRouterTests {
public void routeWithChannelResolver() {
final QueueChannel channel1 = new QueueChannel();
final QueueChannel channel2 = new QueueChannel();
MultiChannelResolver channelResolver = new MultiChannelResolver() {
public List<MessageChannel> resolve(Message<?> message) {
AbstractChannelResolver channelResolver = new AbstractChannelResolver() {
public List<MessageChannel> resolveChannels(Message<?> message) {
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channel1);
channels.add(channel2);
return channels;
}
};
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelResolver(channelResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(channelResolver);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
Message<?> result1 = channel1.receive(25);
assertNotNull(result1);
assertEquals("test", result1.getPayload());
@@ -65,8 +62,8 @@ public class MultiChannelRouterTests {
@Test
public void routeWithChannelNameResolver() {
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return new String[] {"channel1", "channel2"};
}
};
@@ -77,12 +74,10 @@ public class MultiChannelRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(channel1);
channelRegistry.registerChannel(channel2);
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
Message<?> result1 = channel1.receive(25);
assertNotNull(result1);
assertEquals("test", result1.getPayload());
@@ -91,46 +86,30 @@ public class MultiChannelRouterTests {
assertEquals("test", result2.getPayload());
}
@Test(expected = ConfigurationException.class)
public void configuringBothChannelResolverAndChannelNameResolverIsNotAllowed() {
MultiChannelResolver channelResolver = new MultiChannelResolver() {
public List<MessageChannel> resolve(Message<?> message) {
return null;
}
};
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
return null;
}
};
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelResolver(channelResolver);
router.setChannelNameResolver(channelNameResolver);
router.afterPropertiesSet();
}
@Test(expected = MessageDeliveryException.class)
@Test(expected = MessagingException.class)
public void channelNameLookupFailure() {
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return new String[] {"noSuchChannel"};
}
};
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
}
@Test(expected = ConfigurationException.class)
public void channelResolverIsRequired() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
@Test(expected = MessagingException.class)
public void channelRegistryNotAvailable() {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return new String[] {"noSuchChannel"};
}
};
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
Message<String> message = new StringMessage("test");
endpoint.send(message);
}
}

67
org.springframework.integration/src/test/java/org/springframework/integration/router/PayloadTypeRouterTests.java
View File

@@ -24,6 +24,7 @@ import java.util.concurrent.ConcurrentHashMap;
import org.junit.Test;
import org.springframework.integration.channel.DefaultChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.message.GenericMessage;
@@ -36,41 +37,67 @@ import org.springframework.integration.message.StringMessage;
public class PayloadTypeRouterTests {
@Test
public void testRoutingByPayloadType() {
public void resolveByPayloadType() {
QueueChannel stringChannel = new QueueChannel();
QueueChannel integerChannel = new QueueChannel();
Map<Class<?>, MessageChannel> channelMappings = new ConcurrentHashMap<Class<?>, MessageChannel>();
channelMappings.put(String.class, stringChannel);
channelMappings.put(Integer.class, integerChannel);
PayloadTypeRouter router = new PayloadTypeRouter();
router.setChannelMappings(channelMappings);
router.afterPropertiesSet();
PayloadTypeChannelResolver resolver = new PayloadTypeChannelResolver();
resolver.setChannelMappings(channelMappings);
Message<String> message1 = new StringMessage("test");
Message<Integer> message2 = new GenericMessage<Integer>(123);
router.route(message1);
router.route(message2);
Message<?> result1 = stringChannel.receive(25);
assertNotNull(result1);
assertEquals("test", result1.getPayload());
Message<?> result2 = integerChannel.receive(25);
assertNotNull(result2);
assertEquals(123, result2.getPayload());
MessageChannel result1 = resolver.resolveChannel(message1);
MessageChannel result2 = resolver.resolveChannel(message2);
assertEquals(stringChannel, result1);
assertEquals(integerChannel, result2);
}
@Test
public void testRoutingToDefaultChannelWhenNoTypeMatches() {
public void resolveByPayloadTypeWithRouterEndpoint() {
QueueChannel stringChannel = new QueueChannel();
QueueChannel defaultChannel = new QueueChannel();
QueueChannel integerChannel = new QueueChannel();
stringChannel.setBeanName("stringChannel");
integerChannel.setBeanName("integerChannel");
Map<Class<?>, MessageChannel> channelMappings = new ConcurrentHashMap<Class<?>, MessageChannel>();
channelMappings.put(String.class, stringChannel);
PayloadTypeRouter router = new PayloadTypeRouter();
router.setChannelMappings(channelMappings);
router.setDefaultChannel(defaultChannel);
router.afterPropertiesSet();
channelMappings.put(Integer.class, integerChannel);
DefaultChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(stringChannel);
channelRegistry.registerChannel(integerChannel);
PayloadTypeChannelResolver resolver = new PayloadTypeChannelResolver();
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message1 = new StringMessage("test");
Message<Integer> message2 = new GenericMessage<Integer>(123);
router.route(message1);
router.route(message2);
endpoint.send(message1);
endpoint.send(message2);
Message<?> reply1 = stringChannel.receive(0);
Message<?> reply2 = integerChannel.receive(0);
assertEquals("test", reply1.getPayload());
assertEquals(123, reply2.getPayload());
}
@Test
public void routingToDefaultChannelWhenNoTypeMatches() {
QueueChannel stringChannel = new QueueChannel();
stringChannel.setBeanName("stringChannel");
QueueChannel defaultChannel = new QueueChannel();
defaultChannel.setBeanName("defaultChannel");
DefaultChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(stringChannel);
channelRegistry.registerChannel(defaultChannel);
Map<Class<?>, MessageChannel> channelMappings = new ConcurrentHashMap<Class<?>, MessageChannel>();
channelMappings.put(String.class, stringChannel);
PayloadTypeChannelResolver resolver = new PayloadTypeChannelResolver();
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setDefaultOutputChannel(defaultChannel);
Message<String> message1 = new StringMessage("test");
Message<Integer> message2 = new GenericMessage<Integer>(123);
endpoint.send(message1);
endpoint.send(message2);
Message<?> result1 = stringChannel.receive(25);
assertNotNull(result1);
assertEquals("test", result1.getPayload());

60
org.springframework.integration/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java
View File

@@ -19,8 +19,10 @@ package org.springframework.integration.router;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.junit.Test;
@@ -38,6 +40,23 @@ import org.springframework.integration.message.StringMessage;
*/
public class RecipientListRouterTests {
@Test
public void resolveWithChannelList() {
QueueChannel channel1 = new QueueChannel();
QueueChannel channel2 = new QueueChannel();
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channel1);
channels.add(channel2);
RecipientListChannelResolver resolver = new RecipientListChannelResolver();
resolver.setChannels(channels);
resolver.afterPropertiesSet();
Message<String> message = new StringMessage("test");
Collection<MessageChannel> resolved = resolver.resolveChannels(message);
assertEquals(2, resolved.size());
assertTrue(resolved.contains(channel1));
assertTrue(resolved.contains(channel2));
}
@Test
public void routeWithChannelList() {
QueueChannel channel1 = new QueueChannel();
@@ -45,11 +64,12 @@ public class RecipientListRouterTests {
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channel1);
channels.add(channel2);
RecipientListRouter router = new RecipientListRouter();
router.setChannels(channels);
router.afterPropertiesSet();
RecipientListChannelResolver resolver = new RecipientListChannelResolver();
resolver.setChannels(channels);
resolver.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(resolver);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
Message<?> result1 = channel1.receive(25);
assertNotNull(result1);
assertEquals("test", result1.getPayload());
@@ -67,12 +87,13 @@ public class RecipientListRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(channel1);
channelRegistry.registerChannel(channel2);
RecipientListRouter router = new RecipientListRouter();
router.setChannelNames(new String[] {"channel1", "channel2"});
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
RecipientListChannelResolver resolver = new RecipientListChannelResolver();
resolver.setChannelNames(new String[] {"channel1", "channel2"});
resolver.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
Message<?> result1 = channel1.receive(25);
assertNotNull(result1);
assertEquals("test", result1.getPayload());
@@ -90,12 +111,13 @@ public class RecipientListRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(channel1);
channelRegistry.registerChannel(channel2);
RecipientListRouter router = new RecipientListRouter();
router.setChannelNames(new String[] {"channel1"});
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
RecipientListChannelResolver resolver = new RecipientListChannelResolver();
resolver.setChannelNames(new String[] {"channel1"});
resolver.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
Message<?> result1 = channel1.receive(25);
assertNotNull(result1);
assertEquals("test", result1.getPayload());
@@ -115,11 +137,11 @@ public class RecipientListRouterTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(channel1);
channelRegistry.registerChannel(channel2);
RecipientListRouter router = new RecipientListRouter();
router.setChannels(channels);
router.setChannelNames(new String[] {"channel1"});
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
RecipientListChannelResolver resolver = new RecipientListChannelResolver();
resolver.setChannels(channels);
resolver.setChannelNames(new String[] {"channel1"});
resolver.setChannelRegistry(channelRegistry);
resolver.afterPropertiesSet();
}
}

65
org.springframework.integration/src/test/java/org/springframework/integration/router/RootCauseErrorMessageRouterTests.java
View File

@@ -56,16 +56,16 @@ public class RootCauseErrorMessageRouterTests {
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
ErrorMessage message = new ErrorMessage(error);
RootCauseErrorMessageRouter router = new RootCauseErrorMessageRouter();
RootCauseErrorMessageChannelResolver resolver = new RootCauseErrorMessageChannelResolver();
Map<Class<? extends Throwable>, MessageChannel> channelMappings =
new HashMap<Class<? extends Throwable>, MessageChannel>();
channelMappings.put(IllegalArgumentException.class, illegalArgumentChannel);
channelMappings.put(RuntimeException.class, runtimeExceptionChannel);
channelMappings.put(MessageHandlingException.class, messageHandlingExceptionChannel);
router.setChannelMappings(channelMappings);
router.setDefaultChannel(defaultChannel);
router.afterPropertiesSet();
router.route(message);
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setDefaultOutputChannel(defaultChannel);
endpoint.send(message);
assertNotNull(illegalArgumentChannel.receive(1000));
assertNull(defaultChannel.receive(0));
assertNull(runtimeExceptionChannel.receive(0));
@@ -79,15 +79,15 @@ public class RootCauseErrorMessageRouterTests {
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
ErrorMessage message = new ErrorMessage(error);
RootCauseErrorMessageRouter router = new RootCauseErrorMessageRouter();
RootCauseErrorMessageChannelResolver resolver = new RootCauseErrorMessageChannelResolver();
Map<Class<? extends Throwable>, MessageChannel> channelMappings =
new HashMap<Class<? extends Throwable>, MessageChannel>();
channelMappings.put(RuntimeException.class, runtimeExceptionChannel);
channelMappings.put(MessageHandlingException.class, messageHandlingExceptionChannel);
router.setChannelMappings(channelMappings);
router.setDefaultChannel(defaultChannel);
router.afterPropertiesSet();
router.route(message);
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setDefaultOutputChannel(defaultChannel);
endpoint.send(message);
assertNotNull(runtimeExceptionChannel.receive(1000));
assertNull(illegalArgumentChannel.receive(0));
assertNull(defaultChannel.receive(0));
@@ -101,14 +101,14 @@ public class RootCauseErrorMessageRouterTests {
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
ErrorMessage message = new ErrorMessage(error);
RootCauseErrorMessageRouter router = new RootCauseErrorMessageRouter();
RootCauseErrorMessageChannelResolver resolver = new RootCauseErrorMessageChannelResolver();
Map<Class<? extends Throwable>, MessageChannel> channelMappings =
new HashMap<Class<? extends Throwable>, MessageChannel>();
channelMappings.put(MessageHandlingException.class, messageHandlingExceptionChannel);
router.setChannelMappings(channelMappings);
router.setDefaultChannel(defaultChannel);
router.afterPropertiesSet();
router.route(message);
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setDefaultOutputChannel(defaultChannel);
endpoint.send(message);
assertNotNull(messageHandlingExceptionChannel.receive(1000));
assertNull(runtimeExceptionChannel.receive(0));
assertNull(illegalArgumentChannel.receive(0));
@@ -122,10 +122,10 @@ public class RootCauseErrorMessageRouterTests {
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
ErrorMessage message = new ErrorMessage(error);
RootCauseErrorMessageRouter router = new RootCauseErrorMessageRouter();
router.setDefaultChannel(defaultChannel);
router.afterPropertiesSet();
router.route(message);
RootCauseErrorMessageChannelResolver resolver = new RootCauseErrorMessageChannelResolver();
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setDefaultOutputChannel(defaultChannel);
endpoint.send(message);
assertNotNull(defaultChannel.receive(1000));
assertNull(runtimeExceptionChannel.receive(0));
assertNull(illegalArgumentChannel.receive(0));
@@ -139,13 +139,12 @@ public class RootCauseErrorMessageRouterTests {
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
ErrorMessage message = new ErrorMessage(error);
RootCauseErrorMessageRouter router = new RootCauseErrorMessageRouter();
RootCauseErrorMessageChannelResolver resolver = new RootCauseErrorMessageChannelResolver();
Map<Class<? extends Throwable>, MessageChannel> channelMappings =
new HashMap<Class<? extends Throwable>, MessageChannel>();
channelMappings.put(MessageDeliveryException.class, messageDeliveryExceptionChannel);
router.setChannelMappings(channelMappings);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setResolutionRequired(true);
endpoint.send(message);
}
@@ -157,16 +156,16 @@ public class RootCauseErrorMessageRouterTests {
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
Message<?> message = new GenericMessage<Exception>(error);
RootCauseErrorMessageRouter router = new RootCauseErrorMessageRouter();
RootCauseErrorMessageChannelResolver resolver = new RootCauseErrorMessageChannelResolver();
Map<Class<? extends Throwable>, MessageChannel> channelMappings =
new HashMap<Class<? extends Throwable>, MessageChannel>();
channelMappings.put(IllegalArgumentException.class, illegalArgumentChannel);
channelMappings.put(RuntimeException.class, runtimeExceptionChannel);
channelMappings.put(MessageHandlingException.class, messageHandlingExceptionChannel);
router.setChannelMappings(channelMappings);
router.setDefaultChannel(defaultChannel);
router.afterPropertiesSet();
router.route(message);
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setDefaultOutputChannel(defaultChannel);
endpoint.send(message);
assertNotNull(illegalArgumentChannel.receive(1000));
assertNull(defaultChannel.receive(0));
assertNull(runtimeExceptionChannel.receive(0));
@@ -180,15 +179,15 @@ public class RootCauseErrorMessageRouterTests {
RuntimeException middleCause = new RuntimeException(rootCause);
MessageHandlingException error = new MessageHandlingException(failedMessage, "failed", middleCause);
ErrorMessage message = new ErrorMessage(error);
RootCauseErrorMessageRouter router = new RootCauseErrorMessageRouter();
RootCauseErrorMessageChannelResolver resolver = new RootCauseErrorMessageChannelResolver();
Map<Class<? extends Throwable>, MessageChannel> channelMappings =
new HashMap<Class<? extends Throwable>, MessageChannel>();
channelMappings.put(IllegalArgumentException.class, illegalArgumentChannel);
channelMappings.put(MessageHandlingException.class, messageHandlingExceptionChannel);
router.setChannelMappings(channelMappings);
router.setDefaultChannel(defaultChannel);
router.afterPropertiesSet();
router.route(message);
resolver.setChannelMappings(channelMappings);
RouterEndpoint endpoint = new RouterEndpoint(resolver);
endpoint.setDefaultOutputChannel(defaultChannel);
endpoint.send(message);
assertNotNull(illegalArgumentChannel.receive(1000));
assertNull(defaultChannel.receive(0));
assertNull(runtimeExceptionChannel.receive(0));

94
org.springframework.integration/src/test/java/org/springframework/integration/router/RouterEndpointTests.java
View File

@@ -38,30 +38,24 @@ public class RouterEndpointTests {
@Test
public void nullChannelIgnoredByDefault() {
MultiChannelResolver channelResolver = new MultiChannelResolver() {
public List<MessageChannel> resolve(Message<?> message) {
ChannelResolver channelResolver = new ChannelResolver() {
public List<MessageChannel> resolveChannels(Message<?> message) {
return null;
}
};
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelResolver(channelResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelResolver);
Message<String> message = new StringMessage("test");
assertFalse(endpoint.send(message));
}
@Test(expected = MessageDeliveryException.class)
public void nullChannelThrowsExceptionWhenResolutionRequired() {
MultiChannelResolver channelResolver = new MultiChannelResolver() {
public List<MessageChannel> resolve(Message<?> message) {
ChannelResolver channelResolver = new ChannelResolver() {
public List<MessageChannel> resolveChannels(Message<?> message) {
return null;
}
};
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelResolver(channelResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelResolver);
endpoint.setResolutionRequired(true);
Message<String> message = new StringMessage("test");
endpoint.send(message);
@@ -69,30 +63,24 @@ public class RouterEndpointTests {
@Test
public void emptyChannelListIgnoredByDefault() {
MultiChannelResolver channelResolver = new MultiChannelResolver() {
public List<MessageChannel> resolve(Message<?> message) {
ChannelResolver channelResolver = new ChannelResolver() {
public List<MessageChannel> resolveChannels(Message<?> message) {
return Collections.emptyList();
}
};
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelResolver(channelResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelResolver);
Message<String> message = new StringMessage("test");
assertFalse(endpoint.send(message));
}
@Test(expected = MessageDeliveryException.class)
public void emptyChannelListThrowsExceptionWhenResolutionRequired() {
MultiChannelResolver channelResolver = new MultiChannelResolver() {
public List<MessageChannel> resolve(Message<?> message) {
ChannelResolver channelResolver = new ChannelResolver() {
public List<MessageChannel> resolveChannels(Message<?> message) {
return Collections.emptyList();
}
};
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelResolver(channelResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelResolver);
endpoint.setResolutionRequired(true);
Message<String> message = new StringMessage("test");
endpoint.send(message);
@@ -100,16 +88,13 @@ public class RouterEndpointTests {
@Test
public void nullChannelNameArrayIgnoredByDefault() {
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return null;
}
};
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
assertFalse(endpoint.send(message));
@@ -117,16 +102,13 @@ public class RouterEndpointTests {
@Test(expected = MessageDeliveryException.class)
public void nullChannelNameArrayThrowsExceptionWhenResolutionRequired() {
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return null;
}
};
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setResolutionRequired(true);
Message<String> message = new StringMessage("test");
@@ -136,16 +118,13 @@ public class RouterEndpointTests {
@Test
public void emptyChannelNameArrayIgnoredByDefault() {
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return new String[] {};
}
};
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
assertFalse(endpoint.send(message));
@@ -153,16 +132,13 @@ public class RouterEndpointTests {
@Test(expected = MessageDeliveryException.class)
public void emptyChannelNameArrayThrowsExceptionWhenResolutionRequired() {
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return new String[] {};
}
};
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setResolutionRequired(true);
Message<String> message = new StringMessage("test");
@@ -171,28 +147,30 @@ public class RouterEndpointTests {
@Test(expected = MessagingException.class)
public void testChannelRegistryIsRequiredWhenUsingChannelNameResolverWithSingleChannelRouter() {
ChannelNameResolver channelNameResolver = new ChannelNameResolver() {
public String resolve(Message<?> message) {
AbstractSingleChannelNameResolver channelNameResolver = new AbstractSingleChannelNameResolver() {
public String resolveChannelName(Message<?> message) {
return "notImportant";
}
};
SingleChannelRouter router = new SingleChannelRouter();
router.setChannelNameResolver(channelNameResolver);
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.send(new StringMessage("this should fail"));
}
@Test(expected = MessagingException.class)
public void testChannelRegistryIsRequiredWhenUsingChannelNameResolverWithMultiChannelRouter() {
MultiChannelNameResolver channelNameResolver = new MultiChannelNameResolver() {
public String[] resolve(Message<?> message) {
AbstractMultiChannelNameResolver channelNameResolver = new AbstractMultiChannelNameResolver() {
public String[] resolveChannelNames(Message<?> message) {
return new String[] { "notImportant" };
}
};
MultiChannelRouter router = new MultiChannelRouter();
router.setChannelNameResolver(channelNameResolver);
RouterEndpoint endpoint = new RouterEndpoint(router);
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.send(new StringMessage("this should fail"));
}
@Test(expected = IllegalArgumentException.class)
public void testChannelResolverMustNotBeNull() {
AbstractSingleChannelNameResolver channelNameResolver = null;
new RouterEndpoint(channelNameResolver);
}
}

75
org.springframework.integration/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java
View File

@@ -22,13 +22,12 @@ import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.DefaultChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.StringMessage;
/**
@@ -39,16 +38,14 @@ public class SingleChannelRouterTests {
@Test
public void routeWithChannelResolver() {
final QueueChannel channel = new QueueChannel();
ChannelResolver channelResolver = new ChannelResolver() {
public MessageChannel resolve(Message<?> message) {
AbstractSingleChannelResolver channelResolver = new AbstractSingleChannelResolver() {
public MessageChannel resolveChannel(Message<?> message) {
return channel;
}
};
SingleChannelRouter router = new SingleChannelRouter();
router.setChannelResolver(channelResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(channelResolver);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
Message<?> result = channel.receive(25);
assertNotNull(result);
assertEquals("test", result.getPayload());
@@ -56,8 +53,8 @@ public class SingleChannelRouterTests {
@Test
public void routeWithChannelNameResolver() {
ChannelNameResolver channelNameResolver = new ChannelNameResolver() {
public String resolve(Message<?> message) {
AbstractSingleChannelNameResolver channelNameResolver = new AbstractSingleChannelNameResolver() {
public String resolveChannelName(Message<?> message) {
return "testChannel";
}
};
@@ -65,71 +62,39 @@ public class SingleChannelRouterTests {
channel.setBeanName("testChannel");
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel(channel);
SingleChannelRouter router = new SingleChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
router.route(message);
endpoint.send(message);
Message<?> result = channel.receive(25);
assertNotNull(result);
assertEquals("test", result.getPayload());
}
@Test(expected = ConfigurationException.class)
public void configuringBothChannelResolverAndChannelNameResolverIsNotAllowed() {
ChannelResolver channelResolver = new ChannelResolver() {
public MessageChannel resolve(Message<?> message) {
return new QueueChannel();
}
};
ChannelNameResolver channelNameResolver = new ChannelNameResolver() {
public String resolve(Message<?> message) {
return "";
}
};
SingleChannelRouter router = new SingleChannelRouter();
router.setChannelResolver(channelResolver);
router.setChannelNameResolver(channelNameResolver);
router.afterPropertiesSet();
}
@Test
public void nullChannelResult() {
ChannelResolver channelResolver = new ChannelResolver() {
public MessageChannel resolve(Message<?> message) {
AbstractSingleChannelResolver channelResolver = new AbstractSingleChannelResolver() {
public MessageChannel resolveChannel(Message<?> message) {
return null;
}
};
SingleChannelRouter router = new SingleChannelRouter();
router.setChannelResolver(channelResolver);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(channelResolver);
Message<String> message = new StringMessage("test");
assertFalse(router.route(message));
assertFalse(endpoint.send(message));
}
@Test(expected = MessageDeliveryException.class)
@Test(expected = MessagingException.class)
public void channelNameResolutionFailure() {
ChannelNameResolver channelNameResolver = new ChannelNameResolver() {
public String resolve(Message<?> message) {
AbstractSingleChannelNameResolver channelNameResolver = new AbstractSingleChannelNameResolver() {
public String resolveChannelName(Message<?> message) {
return "noSuchChannel";
}
};
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
SingleChannelRouter router = new SingleChannelRouter();
router.setChannelNameResolver(channelNameResolver);
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
RouterEndpoint endpoint = new RouterEndpoint(channelNameResolver);
endpoint.setChannelRegistry(channelRegistry);
Message<String> message = new StringMessage("test");
router.route(message);
}
@Test(expected = ConfigurationException.class)
public void testChannelResolverIsRequired() {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
SingleChannelRouter router = new SingleChannelRouter();
router.setChannelRegistry(channelRegistry);
router.afterPropertiesSet();
endpoint.send(message);
}
}