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

Implemented multi-channel routing with support for resolution by channel instance or channel name.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-02 22:22:01 +00:00
parent a4d542a179
commit 57bb42270a
8 changed files with 556 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
spring-integration-core/src/main/java/org/springframework/integration/router/AbstractRoutingMessageHandler.java Normal file
View File

@@ -0,0 +1,113 @@
/*
* Copyright 2002-2007 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.MessageHandlingException;
import org.springframework.integration.MessagingConfigurationException;
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;
/**
* 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 errorMessage = "failed to resolve any channel for message";
if (this.resolutionRequired) {
throw new MessageHandlingException(errorMessage);
}
if (logger.isWarnEnabled()) {
logger.warn(errorMessage);
}
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("failed to send message to channel '" + channel.getName() + "'");
}
}
protected abstract void validate() throws MessagingConfigurationException;
protected abstract List<MessageChannel> resolveChannels(Message<?> message);
}

30
spring-integration-core/src/main/java/org/springframework/integration/router/MultiChannelNameResolver.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* Copyright 2002-2007 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);
}

33
spring-integration-core/src/main/java/org/springframework/integration/router/MultiChannelResolver.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* Copyright 2002-2007 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;
/**
* Strategy interface for content-based routing to multiple channels.
*
* @author Mark Fisher
*/
public interface MultiChannelResolver {
List<MessageChannel> resolve(Message<?> message);
}

86
spring-integration-core/src/main/java/org/springframework/integration/router/MultiChannelRouter.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* Copyright 2002-2007 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.List;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* 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}.
*
* @author Mark Fisher
*/
public class MultiChannelRouter extends AbstractRoutingMessageHandler {
private MultiChannelResolver channelResolver;
private MultiChannelNameResolver channelNameResolver;
public void setChannelResolver(MultiChannelResolver channelResolver) {
this.channelResolver = channelResolver;
}
public void setChannelNameResolver(MultiChannelNameResolver channelNameResolver) {
this.channelNameResolver = channelNameResolver;
}
@Override
public void validate() {
if (!(this.channelResolver != null ^ this.channelNameResolver != null)) {
throw new MessagingConfigurationException(
"exactly one of 'channelResolver' or 'channelNameResolver' must be provided");
}
if (this.channelNameResolver != null && this.getChannelRegistry() == null) {
throw new MessagingConfigurationException("'channelRegistry' is required when resolving by channel name");
}
}
@Override
public List<MessageChannel> resolveChannels(Message<?> message) {
if (this.channelResolver != null) {
return this.channelResolver.resolve(message);
}
if (this.channelNameResolver == null || this.getChannelRegistry() == null) {
throw new MessagingConfigurationException("router configuration requires either "
+ "a 'channelResolver' or both 'channelNameResolver' and 'channelRegistry'");
}
String[] channelNames = this.channelNameResolver.resolve(message);
if (channelNames == null) {
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;
}
}

102
spring-integration-core/src/main/java/org/springframework/integration/router/RoutingMessageHandler.java
View File

@@ -1,102 +0,0 @@
/*
* Copyright 2002-2007 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.beans.factory.InitializingBean;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingConfigurationException;
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;
/**
* @author Mark Fisher
*/
public class RoutingMessageHandler implements MessageHandler, ChannelRegistryAware, InitializingBean {
private ChannelResolver channelResolver;
private ChannelNameResolver channelNameResolver;
private ChannelRegistry channelRegistry;
private long timeout = -1;
/**
* Set the timeout for sending a message to the resolved channel.
* Default is no timeout, meaning the send will block indefinitely.
*/
public void setTimeout(long timeout) {
this.timeout = timeout;
}
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
}
public void setChannelNameResolver(ChannelNameResolver channelNameResolver) {
this.channelNameResolver = channelNameResolver;
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
public void afterPropertiesSet() {
if(!(this.channelResolver != null ^ this.channelNameResolver != null)) {
throw new MessagingConfigurationException("exactly one of 'channelResolver' or 'channelNameResolver' must be provided");
}
if (this.channelNameResolver != null && this.channelRegistry == null) {
throw new MessagingConfigurationException("'channelRegistry' is required when resolving by channel name");
}
}
public Message<?> handle(Message<?> message) {
MessageChannel channel = this.resolveChannel(message);
if (channel == null) {
throw new MessageHandlingException("failed to resolve channel");
}
boolean sent = false;
if (timeout < 0) {
sent = channel.send(message);
}
else {
sent = channel.send(message, timeout);
}
if (!sent) {
throw new MessageHandlingException(
"failed to send message to channel '" + channel.getName() + "'");
}
return null;
}
private MessageChannel resolveChannel(Message<?> message) {
if (this.channelResolver != null) {
return this.channelResolver.resolve(message);
}
else if (this.channelNameResolver != null && this.channelRegistry != null) {
String channelName = this.channelNameResolver.resolve(message);
return this.channelRegistry.lookupChannel(channelName);
}
throw new MessagingConfigurationException("router configuration requires either " +
"a 'channelResolver' or both 'channelNameResolver' and 'channelRegistry'");
}
}

84
spring-integration-core/src/main/java/org/springframework/integration/router/SingleChannelRouter.java Normal file
View File

@@ -0,0 +1,84 @@
/*
* Copyright 2002-2007 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.List;
import org.springframework.integration.MessagingConfigurationException;
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. 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}.
*
* @author Mark Fisher
*/
public class SingleChannelRouter extends AbstractRoutingMessageHandler {
private ChannelResolver channelResolver;
private ChannelNameResolver channelNameResolver;
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
}
public void setChannelNameResolver(ChannelNameResolver channelNameResolver) {
this.channelNameResolver = channelNameResolver;
}
@Override
public void validate() {
if (!(this.channelResolver != null ^ this.channelNameResolver != null)) {
throw new MessagingConfigurationException(
"exactly one of 'channelResolver' or 'channelNameResolver' must be provided");
}
if (this.channelNameResolver != null && this.getChannelRegistry() == null) {
throw new MessagingConfigurationException("'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);
}
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 MessagingConfigurationException("router configuration requires either "
+ "a 'channelResolver' or both 'channelNameResolver' and 'channelRegistry'");
}
String channelName = this.channelNameResolver.resolve(message);
return this.getChannelRegistry().lookupChannel(channelName);
}
}