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

Added Router strategy interface, and refactored Message-routing support in general to avoid MessageHandler.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-28 18:46:36 +00:00
parent 62c7439960
commit ba9c086aea
19 changed files with 900 additions and 665 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java Normal file
View File

@@ -0,0 +1,112 @@
/*
* 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.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Base class parser for elements that create Message Endpoints.
*
* @author Mark Fisher
*/
public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitionParser {
protected static final String REF_ATTRIBUTE = "ref";
protected static final String METHOD_ATTRIBUTE = "method";
protected static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
protected static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String POLLER_ELEMENT = "poller";
private static final String ERROR_HANDLER_ATTRIBUTE = "error-handler";
@Override
protected Class<?> getBeanClass(Element element) {
return this.getEndpointClass();
}
@Override
protected boolean shouldGenerateId() {
return false;
}
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String ref = element.getAttribute(REF_ATTRIBUTE);
if (!StringUtils.hasText(ref)) {
throw new ConfigurationException("The '" + REF_ATTRIBUTE + "' attribute is required.");
}
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
String method = element.getAttribute(METHOD_ATTRIBUTE);
String adapterBeanName = this.parseAdapter(ref, method, element, parserContext);
builder.addConstructorArgReference(adapterBeanName);
}
else {
builder.addConstructorArgReference(ref);
}
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
if (!StringUtils.hasText(inputChannel)) {
throw new ConfigurationException("the '" + INPUT_CHANNEL_ATTRIBUTE + "' attribute is required");
}
Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
if (pollerElement != null) {
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(inputChannel, pollerElement, parserContext);
builder.addPropertyReference("source", pollerBeanName);
}
else {
builder.addPropertyValue("inputChannelName", inputChannel);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(
builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, ERROR_HANDLER_ATTRIBUTE);
}
private String parseAdapter(String ref, String method, Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(this.getMethodInvokingAdapterClass());
builder.addConstructorArgReference(ref);
builder.addConstructorArgValue(method);
String adapterBeanName = BeanDefinitionReaderUtils.generateBeanName(builder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionHolder holder = new BeanDefinitionHolder(builder.getBeanDefinition(), adapterBeanName);
parserContext.registerBeanComponent(new BeanComponentDefinition(holder));
return adapterBeanName;
}
protected abstract Class<? extends MessageEndpoint> getEndpointClass();
protected abstract Class<?> getMethodInvokingAdapterClass();
}

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

@@ -16,19 +16,25 @@
package org.springframework.integration.config;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.router.RouterMessageHandler;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.integration.router.RouterEndpoint;
/**
* Parser for the &lt;router/&gt; element.
*
* @author Mark Fisher
*/
public class RouterParser extends AbstractMessageEndpointParser {
public class RouterParser extends AbstractEndpointParser {
@Override
protected Class<? extends MessageHandler> getHandlerAdapterClass() {
return RouterMessageHandler.class;
protected Class<? extends MessageEndpoint> getEndpointClass() {
return RouterEndpoint.class;
}
@Override
protected Class<?> getMethodInvokingAdapterClass() {
return MethodInvokingRouter.class;
}
}

79
org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterParser.java
View File

@@ -16,94 +16,25 @@
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.splitter.MethodInvokingSplitter;
import org.springframework.integration.splitter.SplitterEndpoint;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Parser for the &lt;splitter/&gt; element.
*
* @author Mark Fisher
*/
public class SplitterParser extends AbstractSingleBeanDefinitionParser {
protected static final String REF_ATTRIBUTE = "ref";
protected static final String METHOD_ATTRIBUTE = "method";
protected static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
protected static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String POLLER_ELEMENT = "poller";
private static final String ERROR_HANDLER_ATTRIBUTE = "error-handler";
public class SplitterParser extends AbstractEndpointParser {
@Override
protected Class<?> getBeanClass(Element element) {
protected Class<? extends MessageEndpoint> getEndpointClass() {
return SplitterEndpoint.class;
}
@Override
protected boolean shouldGenerateId() {
return false;
}
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String ref = element.getAttribute(REF_ATTRIBUTE);
if (!StringUtils.hasText(ref)) {
throw new ConfigurationException("The '" + REF_ATTRIBUTE + "' attribute is required.");
}
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
String method = element.getAttribute(METHOD_ATTRIBUTE);
String adapterBeanName = this.parseAdapter(ref, method, element, parserContext);
builder.addConstructorArgReference(adapterBeanName);
}
else {
builder.addConstructorArgReference(ref);
}
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
if (!StringUtils.hasText(inputChannel)) {
throw new ConfigurationException("the '" + INPUT_CHANNEL_ATTRIBUTE + "' attribute is required");
}
Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
if (pollerElement != null) {
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(inputChannel, pollerElement, parserContext);
builder.addPropertyReference("source", pollerBeanName);
}
else {
builder.addPropertyValue("inputChannelName", inputChannel);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(
builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, ERROR_HANDLER_ATTRIBUTE);
}
private String parseAdapter(String ref, String method, Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingSplitter.class);
builder.addConstructorArgReference(ref);
builder.addConstructorArgValue(method);
String adapterBeanName = BeanDefinitionReaderUtils.generateBeanName(builder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionHolder holder = new BeanDefinitionHolder(builder.getBeanDefinition(), adapterBeanName);
parserContext.registerBeanComponent(new BeanComponentDefinition(holder));
return adapterBeanName;
protected Class<?> getMethodInvokingAdapterClass() {
return MethodInvokingSplitter.class;
}
}

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

@@ -0,0 +1,92 @@
/*
* 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);
}

114
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractRoutingMessageHandler.java
View File

@@ -1,114 +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.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
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.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
/**
* Base class for message router implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractRoutingMessageHandler implements MessageHandler, ChannelRegistryAware, InitializingBean {
protected Log logger = LogFactory.getLog(this.getClass());
private ChannelRegistry channelRegistry;
private boolean resolutionRequired = false;
private long timeout = -1;
/**
* Set whether this router should always be required to resolve at least one
* channel. The default is 'false'. To trigger an exception whenever the
* resolver returns null or an empty channel list, set this value to 'true'.
*/
public void setResolutionRequired(boolean resolutionRequired) {
this.resolutionRequired = resolutionRequired;
}
/**
* Set the timeout for sending a message to the resolved channel. By
* default, there is no timeout, meaning the send will block indefinitely.
*/
public void setTimeout(long timeout) {
this.timeout = timeout;
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
protected ChannelRegistry getChannelRegistry() {
return this.channelRegistry;
}
public final void afterPropertiesSet() {
this.validate();
}
public final Message<?> handle(Message<?> message) {
List<MessageChannel> channels = this.resolveChannels(message);
if (channels == null || channels.size() == 0) {
String description = "failed to resolve any channel for message";
if (this.resolutionRequired) {
throw new MessageHandlingException(message, description);
}
if (logger.isWarnEnabled()) {
logger.warn(description);
}
return null;
}
for (MessageChannel channel : channels) {
this.sendMesage(message, channel);
}
return null;
}
private void sendMesage(Message<?> message, MessageChannel channel) {
boolean sent = false;
if (timeout < 0) {
sent = channel.send(message);
}
else {
sent = channel.send(message, timeout);
}
if (!sent) {
throw new MessageHandlingException(message,
"failed to send message to channel '" + channel.getName() + "'");
}
}
protected abstract void validate() throws ConfigurationException;
protected abstract List<MessageChannel> resolveChannels(Message<?> message);
}

86
org.springframework.integration/src/main/java/org/springframework/integration/router/MethodInvokingRouter.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* 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.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
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;
import org.springframework.integration.message.MessageMappingMethodInvoker;
/**
* 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.
*
* @author Mark Fisher
*/
public class MethodInvokingRouter extends AbstractRouter implements InitializingBean {
private final MessageMappingMethodInvoker invoker;
public MethodInvokingRouter(Object object, Method method) {
this.invoker = new MessageMappingMethodInvoker(object, method);
}
public MethodInvokingRouter(Object object, String methodName) {
this.invoker = new MessageMappingMethodInvoker(object, methodName);
}
public void afterPropertiesSet() throws Exception {
this.invoker.afterPropertiesSet();
}
@Override
protected Collection<?> resolveChannels(Message<?> message) {
Object result = this.invoker.invokeMethod(message);
if (result == null) {
return null;
}
List<Object> channels = new ArrayList<Object>();
if (result instanceof Collection) {
channels.addAll((Collection<?>) result);
}
else if (result instanceof MessageChannel[]) {
channels.addAll(Arrays.asList((MessageChannel[]) result));
}
else if (result instanceof String[]) {
channels.addAll(Arrays.asList((String[]) result));
}
else if (result instanceof MessageChannel) {
channels.add((MessageChannel) result);
}
else if (result instanceof String) {
channels.add(result);
}
else {
throw new ConfigurationException(
"router method must return type 'MessageChannel' or 'String'");
}
return channels;
}
}

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

@@ -16,29 +16,26 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.List;
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. In the case of the latter, the
* {@link org.springframework.integration.channel.ChannelRegistry} reference
* must also be provided. For convenience, the superclass does implement
* {@link org.springframework.integration.channel.ChannelRegistryAware}.
* 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 AbstractRoutingMessageHandler {
public class MultiChannelRouter extends AbstractRouter implements InitializingBean {
private MultiChannelResolver channelResolver;
private volatile MultiChannelResolver channelResolver;
private MultiChannelNameResolver channelNameResolver;
private volatile MultiChannelNameResolver channelNameResolver;
public void setChannelResolver(MultiChannelResolver channelResolver) {
@@ -49,38 +46,23 @@ public class MultiChannelRouter extends AbstractRoutingMessageHandler {
this.channelNameResolver = channelNameResolver;
}
@Override
public void validate() {
public void afterPropertiesSet() {
if (!(this.channelResolver != null ^ this.channelNameResolver != null)) {
throw new ConfigurationException(
"exactly one of 'channelResolver' or 'channelNameResolver' must be provided");
}
if (this.channelNameResolver != null && this.getChannelRegistry() == null) {
throw new ConfigurationException("'channelRegistry' is required when resolving by channel name");
}
}
@Override
public List<MessageChannel> resolveChannels(Message<?> message) {
public Collection<?> resolveChannels(Message<?> message) {
if (this.channelResolver != null) {
return this.channelResolver.resolve(message);
}
if (this.channelNameResolver == null || this.getChannelRegistry() == null) {
throw new ConfigurationException("router configuration requires either "
+ "a 'channelResolver' or both 'channelNameResolver' and 'channelRegistry'");
}
String[] channelNames = this.channelNameResolver.resolve(message);
if (channelNames == null) {
if (ObjectUtils.isEmpty(channelNames)) {
return null;
}
List<MessageChannel> channels = new ArrayList<MessageChannel>(channelNames.length);
for (String channelName : channelNames) {
MessageChannel channel = this.getChannelRegistry().lookupChannel(channelName);
if (channel != null) {
channels.add(channel);
}
}
return channels;
return Arrays.asList(channelNames);
}
}

30
org.springframework.integration/src/main/java/org/springframework/integration/router/Router.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* 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 routing a Message to one or more channels.
*
* @author Mark Fisher
*/
public interface Router {
boolean route(Message<?> message);
}

107
org.springframework.integration/src/main/java/org/springframework/integration/router/RouterEndpoint.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* 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.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.util.Assert;
/**
* @author Mark Fisher
*/
public class RouterEndpoint extends AbstractEndpoint {
private final Router router;
private volatile MessageChannel defaultOutputChannel;
private volatile boolean resolutionRequired;
public RouterEndpoint(Router router) {
Assert.notNull(router, "router must not be null");
this.router = router;
}
@Override
public void setChannelRegistry(ChannelRegistry channelRegistry) {
super.setChannelRegistry(channelRegistry);
if (this.router instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.router).setChannelRegistry(channelRegistry);
}
}
public void setDefaultOutputChannel(MessageChannel defaultOutputChannel) {
this.defaultOutputChannel = defaultOutputChannel;
}
/**
* Set the timeout for sending a message to the resolved channel. By
* default, there is no timeout, meaning the send will block indefinitely.
*/
public void setTimeout(long timeout) {
this.getMessageExchangeTemplate().setSendTimeout(timeout);
}
/**
* Set whether this router should always be required to resolve at least one
* channel. The default is 'false'. To trigger an exception whenever the
* resolver returns null or an empty channel list, and this endpoint has
* no 'defaultOutputChannel' configured, set this value to 'true'.
*/
public void setResolutionRequired(boolean resolutionRequired) {
this.resolutionRequired = resolutionRequired;
}
@Override
protected boolean sendInternal(Message<?> message) {
boolean sent = this.router.route(message);
if (!sent) {
if (this.defaultOutputChannel != null) {
sent = this.getMessageExchangeTemplate().send(message, this.defaultOutputChannel);
}
else if (this.resolutionRequired) {
throw new MessageDeliveryException(message,
"no target resolved by router and no default output channel defined");
}
}
return sent;
}
// TODO: remove these methods after refactoring
private volatile String inputChannelName;
public String getInputChannelName() {
return this.inputChannelName;
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public String getOutputChannelName() {
return null;
}
}

81
org.springframework.integration/src/main/java/org/springframework/integration/router/RouterMessageHandler.java
View File

@@ -17,95 +17,40 @@
package org.springframework.integration.router;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.CompositeMessage;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageTarget;
/**
* MessageHandler adapter for methods annotated with {@link Router @Router}.
*
* @author Mark Fisher
*/
public class RouterMessageHandler extends AbstractMessageHandler {
public class RouterMessageHandler implements MessageHandler, ChannelRegistryAware {
private volatile MessageChannel defaultChannel;
private final Router router;
public RouterMessageHandler(Object object, Method method) {
super(object, method);
this.router = new MethodInvokingRouter(object, method);
}
public RouterMessageHandler(Object object, String methodName) {
super(object, methodName);
}
public RouterMessageHandler() {
this.router = new MethodInvokingRouter(object, methodName);
}
public void setDefaultChannel(MessageChannel defaultChannel) {
this.defaultChannel = defaultChannel;
public Message<?> handle(Message<?> message) {
this.router.route(message);
return null;
}
@Override
protected Message<?> createReplyMessage(Object result, Message<?> requestMessage) {
final List<Object> channels = new ArrayList<Object>();
if (result != null) {
if (result instanceof Collection) {
channels.addAll((Collection<?>) result);
}
else if (result instanceof MessageChannel[]) {
channels.addAll(Arrays.asList((MessageChannel[]) result));
}
else if (result instanceof String[]) {
channels.addAll(Arrays.asList((String[]) result));
}
else if (result instanceof MessageChannel) {
channels.add((MessageChannel) result);
}
else if (result instanceof String) {
channels.add(result);
}
else {
throw new ConfigurationException(
"router method must return type 'MessageChannel' or 'String'");
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
if (this.router instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.router).setChannelRegistry(channelRegistry);
}
if (channels.size() == 0) {
if (this.defaultChannel != null) {
return MessageBuilder.fromMessage(requestMessage).setNextTarget(this.defaultChannel).build();
}
return null;
}
List<Message<?>> replies = new ArrayList<Message<?>>();
for (Object channel : channels) {
MessageBuilder<?> builder = MessageBuilder.fromMessage(requestMessage);
if (channel instanceof MessageTarget) {
builder.setNextTarget((MessageTarget) channel);
}
else if (channel instanceof String) {
builder.setNextTarget((String) channel);
}
replies.add(builder.build());
}
return new CompositeMessage(replies);
}
@Override
protected Message<?> postProcessReplyMessage(Message<?> replyMessage, Message<?> requestMessage) {
throw new MessageHandlingException(requestMessage,
"router method must return type 'MessageChannel' or 'String', but a Message was returned: " + replyMessage);
}
}

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

@@ -16,9 +16,10 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.List;
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;
@@ -26,14 +27,11 @@ 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. In the case of the latter, the
* {@link org.springframework.integration.channel.ChannelRegistry} reference
* must also be provided. For convenience, the superclass does implement
* {@link org.springframework.integration.channel.ChannelRegistryAware}.
* strategy instance.
*
* @author Mark Fisher
*/
public class SingleChannelRouter extends AbstractRoutingMessageHandler {
public class SingleChannelRouter extends AbstractRouter implements InitializingBean {
private ChannelResolver channelResolver;
@@ -48,37 +46,22 @@ public class SingleChannelRouter extends AbstractRoutingMessageHandler {
this.channelNameResolver = channelNameResolver;
}
@Override
public void validate() {
public void afterPropertiesSet() {
if (!(this.channelResolver != null ^ this.channelNameResolver != null)) {
throw new ConfigurationException(
"exactly one of 'channelResolver' or 'channelNameResolver' must be provided");
}
if (this.channelNameResolver != null && this.getChannelRegistry() == null) {
throw new ConfigurationException("'channelRegistry' is required when resolving by channel name");
}
}
@Override
public List<MessageChannel> resolveChannels(Message<?> message) {
List<MessageChannel> channels = new ArrayList<MessageChannel>();
MessageChannel channel = this.resolveChannel(message);
if (channel != null) {
channels.add(channel);
protected Collection<?> resolveChannels(Message<?> message) {
Object result = (this.channelResolver != null)
? this.channelResolver.resolve(message)
: this.channelNameResolver.resolve(message);
if (result == null) {
return null;
}
return channels;
}
private MessageChannel resolveChannel(Message<?> message) {
if (this.channelResolver != null) {
return this.channelResolver.resolve(message);
}
if (this.channelNameResolver == null || this.getChannelRegistry() == null) {
throw new ConfigurationException("router configuration requires either "
+ "a 'channelResolver' or both 'channelNameResolver' and 'channelRegistry'");
}
String channelName = this.channelNameResolver.resolve(message);
return this.getChannelRegistry().lookupChannel(channelName);
return Collections.singletonList(result);
}
}