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

Refactored RouterMessageHandlerAdapter (now RouterMessageHandler). Router endpoints are now SimpleEndpoint instances.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-12 03:44:23 +00:00
parent f91ec6f48f
commit 178c438159
5 changed files with 343 additions and 426 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,8 +16,10 @@
package org.springframework.integration.config;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.router.RouterMessageHandlerAdapter;
import org.springframework.integration.router.RouterMessageHandler;
/**
* Parser for the <router/> element.
@@ -26,9 +28,14 @@ import org.springframework.integration.router.RouterMessageHandlerAdapter;
*/
public class RouterParser extends AbstractHandlerEndpointParser {
@Override
protected Class<? extends MessageEndpoint> getEndpointClass() {
return SimpleEndpoint.class;
}
@Override
protected Class<? extends MessageHandler> getHandlerAdapterClass() {
return RouterMessageHandlerAdapter.class;
return RouterMessageHandler.class;
}
}

104
org.springframework.integration/src/main/java/org/springframework/integration/router/RouterMessageHandler.java Normal file
View File

@@ -0,0 +1,104 @@
/*
* 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.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.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageTarget;
/**
* MessageHandler adapter for methods annotated with {@link Router @Router}.
*
* @author Mark Fisher
*/
public class RouterMessageHandler extends AbstractMessageHandler {
private volatile MessageChannel defaultChannel;
public RouterMessageHandler(Object object, Method method) {
super(object, method);
}
public RouterMessageHandler(Object object, String methodName) {
super(object, methodName);
}
public RouterMessageHandler() {
}
public void setDefaultChannel(MessageChannel defaultChannel) {
this.defaultChannel = defaultChannel;
}
@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'");
}
}
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);
}
}

137
org.springframework.integration/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java
View File

@@ -1,137 +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.lang.reflect.Method;
import java.util.Collection;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.annotation.AnnotationMethodMessageMapper;
import org.springframework.integration.message.Message;
/**
* MessageHandler adapter for methods annotated with {@link Router @Router}.
*
* @author Mark Fisher
*/
public class RouterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware {
private volatile ChannelRegistry channelRegistry;
public RouterMessageHandlerAdapter(Object object, Method method) {
this.setObject(object);
this.setMethod(method);
if (method.getParameterTypes().length < 1) {
throw new ConfigurationException("The router method must accept at least one argument.");
}
if (method.getParameterTypes()[0].equals(Message.class)) {
this.setMethodExpectsMessage(true);
}
}
public RouterMessageHandlerAdapter(Object object, String methodName) {
this.setObject(object);
this.setMethodName(methodName);
}
public RouterMessageHandlerAdapter() {
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
@Override
protected void initialize() {
Object target = this.getObject();
if (target != null && this.channelRegistry != null && (target instanceof ChannelRegistryAware)) {
((ChannelRegistryAware) target).setChannelRegistry(this.channelRegistry);
}
Method method = this.getMethod();
if (method != null) {
this.setMessageMapper(new AnnotationMethodMessageMapper(method));
}
}
@Override
protected Message<?> handleReturnValue(Object returnValue, Message<?> originalMessage) {
if (returnValue != null) {
if (returnValue instanceof Collection) {
Collection<?> channels = (Collection<?>) returnValue;
for (Object channel : channels) {
if (channel instanceof MessageChannel) {
this.sendMessage(originalMessage, (MessageChannel) channel);
}
else if (channel instanceof String) {
this.sendMessage(originalMessage, (String) channel);
}
else {
throw new ConfigurationException(
"router method must return type 'MessageChannel' or 'String'");
}
}
}
else if (returnValue instanceof MessageChannel[]) {
for (MessageChannel channel : (MessageChannel[]) returnValue) {
this.sendMessage(originalMessage, channel);
}
}
else if (returnValue instanceof String[]) {
for (String channelName : (String[]) returnValue) {
this.sendMessage(originalMessage, channelName);
}
}
else if (returnValue instanceof MessageChannel) {
this.sendMessage(originalMessage, (MessageChannel) returnValue);
}
else if (returnValue instanceof String) {
this.sendMessage(originalMessage, (String) returnValue);
}
else {
throw new ConfigurationException(
"router method must return type 'MessageChannel' or 'String'");
}
}
return null;
}
private boolean sendMessage(Message<?> message, String channelName) {
MessageChannel channel = this.channelRegistry.lookupChannel(channelName);
if (channel == null) {
if (logger.isWarnEnabled()) {
logger.warn("unable to resolve channel for name '" + channelName + "'");
}
return false;
}
return this.sendMessage(message, channel);
}
private boolean sendMessage(Message<?> message, MessageChannel channel) {
if (logger.isDebugEnabled()) {
logger.debug("sending message to channel '" + channel + "'");
}
return channel.send(message);
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/router/RouterMessageHandlerCreator.java
View File

@@ -30,7 +30,7 @@ import org.springframework.integration.handler.config.AbstractMessageHandlerCrea
public class RouterMessageHandlerCreator extends AbstractMessageHandlerCreator {
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
return new RouterMessageHandlerAdapter(object, method);
return new RouterMessageHandler(object, method);
}
}