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

Refactored handler creator strategies to be more generic (not directly tied to annotations), and added support for splitters.

Browse Source
This commit is contained in:
Mark Fisher
2007-12-19 20:08:24 +00:00
parent fc16b90f94
commit 2dbd2c169d
11 changed files with 301 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -24,6 +24,9 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
@@ -41,11 +44,13 @@ import org.springframework.integration.endpoint.annotation.MessageEndpoint;
import org.springframework.integration.endpoint.annotation.Polled;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.annotation.AnnotationHandlerCreator;
import org.springframework.integration.handler.annotation.DefaultAnnotationHandlerCreator;
import org.springframework.integration.handler.annotation.Handler;
import org.springframework.integration.handler.annotation.Router;
import org.springframework.integration.handler.annotation.RouterAnnotationHandlerCreator;
import org.springframework.integration.handler.annotation.Splitter;
import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.handler.config.RouterMessageHandlerCreator;
import org.springframework.integration.handler.config.SplitterMessageHandlerCreator;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -58,8 +63,10 @@ import org.springframework.util.StringUtils;
*/
public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean {
private Map<Class<? extends Annotation>, AnnotationHandlerCreator> handlerCreators =
new ConcurrentHashMap<Class<? extends Annotation>, AnnotationHandlerCreator>();
private Log logger = LogFactory.getLog(this.getClass());
private Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators =
new ConcurrentHashMap<Class<? extends Annotation>, MessageHandlerCreator>();
private MessageBus messageBus;
@@ -70,16 +77,17 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
}
public void setCustomHandlerCreators(
Map<Class<? extends Annotation>, AnnotationHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, AnnotationHandlerCreator> entry : customHandlerCreators.entrySet()) {
Map<Class<? extends Annotation>, MessageHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) {
this.handlerCreators.put(entry.getKey(), entry.getValue());
}
}
public void afterPropertiesSet() {
Assert.notNull(this.messageBus, "messageBus is required");
this.handlerCreators.put(Handler.class, new DefaultAnnotationHandlerCreator());
this.handlerCreators.put(Router.class, new RouterAnnotationHandlerCreator());
this.handlerCreators.put(Handler.class, new DefaultMessageHandlerCreator());
this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator());
this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator());
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
@@ -170,23 +178,33 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
final List<MessageHandler> handlers = new ArrayList<MessageHandler>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
for (Class<? extends Annotation> annotationType : handlerCreators.keySet()) {
Annotation annotation = AnnotationUtils.getAnnotation(method, annotationType);
if (annotation != null) {
MessageHandler handler = handlerCreators.get(annotationType).createHandler(bean, method, annotation);
if (handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) handler).setChannelRegistry(messageBus);
}
if (handler instanceof InitializingBean) {
try {
((InitializingBean) handler).afterPropertiesSet();
}
catch (Exception e) {
throw new MessagingConfigurationException("failed to create handler", e);
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
for (Annotation annotation : annotations) {
if (isHandlerAnnotation(annotation)) {
Map<String, ?> attributes = AnnotationUtils.getAnnotationAttributes(annotation);
MessageHandlerCreator handlerCreator = handlerCreators.get(annotation.annotationType());
if (handlerCreator == null) {
if (logger.isWarnEnabled()) {
logger.warn("No handler creator has been registered for handler annotation '" +
annotation.annotationType() + "'");
}
}
if (handler != null) {
handlers.add(handler);
else {
MessageHandler handler = handlerCreator.createHandler(bean, method, attributes);
if (handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) handler).setChannelRegistry(messageBus);
}
if (handler instanceof InitializingBean) {
try {
((InitializingBean) handler).afterPropertiesSet();
}
catch (Exception e) {
throw new MessagingConfigurationException("failed to create handler", e);
}
}
if (handler != null) {
handlers.add(handler);
}
}
}
}
@@ -203,4 +221,10 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
return null;
}
private boolean isHandlerAnnotation(Annotation annotation) {
return annotation.annotationType().equals(Handler.class)
|| annotation.annotationType().isAnnotationPresent(Handler.class)
|| this.handlerCreators.keySet().contains(annotation.annotationType());
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Handler.java
View File

@@ -32,7 +32,7 @@ import org.springframework.integration.endpoint.annotation.MessageEndpoint;
*
* @author Mark Fisher
*/
@Target(ElementType.METHOD)
@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented

1
spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Router.java
View File

@@ -33,6 +33,7 @@ import java.lang.annotation.Target;
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
@Handler
public @interface Router {
/**

40
spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Splitter.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* 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.handler.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Indicates that a method is capable of splitting a single message or message
* payload to produce multiple messages which are then sent to the channel whose
* name is provided with the 'channel' attribute.
*
* @author Mark Fisher
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Handler
public @interface Splitter {
String channel();
}

15
spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AbstractAnnotationHandlerCreator.java → spring-integration-core/src/main/java/org/springframework/integration/handler/config/AbstractMessageHandlerCreator.java
View File

@@ -14,10 +14,10 @@
* limitations under the License.
*/
package org.springframework.integration.handler.annotation;
package org.springframework.integration.handler.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.Order;
@@ -25,14 +25,15 @@ import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
/**
* Base class for handler creators for the provided object and method.
* Base class for handler creators that generate a {@link MessageHandler}
* adapter for the provided object and method.
*
* @author Mark Fisher
*/
public abstract class AbstractAnnotationHandlerCreator implements AnnotationHandlerCreator {
public abstract class AbstractMessageHandlerCreator implements MessageHandlerCreator {
public final MessageHandler createHandler(Object object, Method method, Annotation annotation) {
MessageHandler handler = this.doCreateHandler(object, method, annotation);
public final MessageHandler createHandler(Object object, Method method, Map<String, ?> attributes) {
MessageHandler handler = this.doCreateHandler(object, method, attributes);
if (handler instanceof AbstractMessageHandlerAdapter<?>) {
AbstractMessageHandlerAdapter adapter = ((AbstractMessageHandlerAdapter) handler);
adapter.setObject(object);
@@ -45,6 +46,6 @@ public abstract class AbstractAnnotationHandlerCreator implements AnnotationHand
return handler;
}
protected abstract MessageHandler doCreateHandler(Object object, Method method, Annotation annotation);
protected abstract MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes);
}

16
spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/DefaultAnnotationHandlerCreator.java → spring-integration-core/src/main/java/org/springframework/integration/handler/config/DefaultMessageHandlerCreator.java
View File

@@ -14,26 +14,26 @@
* limitations under the License.
*/
package org.springframework.integration.handler.annotation;
package org.springframework.integration.handler.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.core.annotation.Order;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
/**
* Default implementation of the handler creator strategy that creates a
* {@link DefaultMessageHandlerAdapter} for the provided object and method. This
* version does not even consider the annotation itself. It does however
* respect an {@link Order} annotation if present.
* {@link DefaultMessageHandlerAdapter} for the provided object and method.
* This version does not even consider the attributes. It does however respect
* an {@link Order} annotation if present.
*
* @author Mark Fisher
*/
public class DefaultAnnotationHandlerCreator extends AbstractAnnotationHandlerCreator {
public class DefaultMessageHandlerCreator extends AbstractMessageHandlerCreator {
public MessageHandler doCreateHandler(Object object, Method method, Annotation annotation) {
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
return new DefaultMessageHandlerAdapter<Object>();
}

11
spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AnnotationHandlerCreator.java → spring-integration-core/src/main/java/org/springframework/integration/handler/config/MessageHandlerCreator.java
View File

@@ -14,21 +14,20 @@
* limitations under the License.
*/
package org.springframework.integration.handler.annotation;
package org.springframework.integration.handler.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.integration.handler.MessageHandler;
/**
* A strategy for programmatically creating a {@link MessageHandler} based on
* metadata provided by an annotation.
* Strategy interface for creating {@link MessageHandler MessageHandlers}.
*
* @author Mark Fisher
*/
public interface AnnotationHandlerCreator {
public interface MessageHandlerCreator {
MessageHandler createHandler(Object object, Method method, Annotation annotation);
MessageHandler createHandler(Object object, Method method, Map<String, ?> attributes);
}

17
spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/RouterAnnotationHandlerCreator.java → spring-integration-core/src/main/java/org/springframework/integration/handler/config/RouterMessageHandlerCreator.java
View File

@@ -14,28 +14,23 @@
* limitations under the License.
*/
package org.springframework.integration.handler.annotation;
package org.springframework.integration.handler.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.router.RouterMessageHandlerAdapter;
/**
* A handler creator for methods annotation with {@link Router @Router}.
* Creates a {@link MessageHandler} adapter for router methods.
*
* @author Mark Fisher
*/
public class RouterAnnotationHandlerCreator extends AbstractAnnotationHandlerCreator {
public class RouterMessageHandlerCreator extends AbstractMessageHandlerCreator {
public MessageHandler doCreateHandler(Object object, Method method, Annotation annotation) {
if (!annotation.annotationType().equals(Router.class)) {
throw new MessagingConfigurationException(
"RouterAnnotationHandlerCreator only applies to the @Router annotation");
}
return new RouterMessageHandlerAdapter(object, method, (Router) annotation);
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
return new RouterMessageHandlerAdapter(object, method, attributes);
}
}

35
spring-integration-core/src/main/java/org/springframework/integration/handler/config/SplitterMessageHandlerCreator.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* 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.handler.config;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.router.SplitterMessageHandlerAdapter;
/**
* Creates a {@link MessageHandler} adapter for splitter methods.
*
* @author Mark Fisher
*/
public class SplitterMessageHandlerCreator extends AbstractMessageHandlerCreator {
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
return new SplitterMessageHandlerAdapter(object, method, attributes);
}
}

16
spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.router;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Map;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.channel.ChannelRegistry;
@@ -36,17 +37,22 @@ import org.springframework.util.StringUtils;
*/
public class RouterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware {
private Router routerAnnotation;
private static final String PROPERTY_KEY = "property";
private static final String ATTRIBUTE_KEY = "attribute";
private Method method;
private Map<String, ?> attributes;
private ChannelRegistry channelRegistry;
public RouterMessageHandlerAdapter(Object object, Method method, Router routerAnnotation) {
public RouterMessageHandlerAdapter(Object object, Method method, Map<String, ?> attributes) {
this.setObject(object);
this.method = method;
this.routerAnnotation = routerAnnotation;
this.attributes = attributes;
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
@@ -59,8 +65,8 @@ public class RouterMessageHandlerAdapter extends AbstractMessageHandlerAdapter i
throw new MessagingConfigurationException(
"method must accept exactly one parameter");
}
String propertyName = routerAnnotation.property();
String attributeName = routerAnnotation.attribute();
String propertyName = (String) attributes.get(PROPERTY_KEY);
String attributeName = (String) attributes.get(ATTRIBUTE_KEY);
Object retval = null;
if (StringUtils.hasText(propertyName)) {
if (StringUtils.hasText(attributeName)) {

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

@@ -0,0 +1,138 @@
/*
* 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.lang.reflect.Method;
import java.util.Collection;
import java.util.Map;
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.endpoint.SimpleMethodInvoker;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.annotation.Splitter;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
/**
* MessageHandler adapter for methods annotated with {@link Splitter @Splitter}.
*
* @author Mark Fisher
*/
public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware {
private static final String CHANNEL_KEY = "channel";
private Map<String, ?> attributes;
private Method method;
private ChannelRegistry channelRegistry;
public SplitterMessageHandlerAdapter(Object object, Method method, Map<String, ?> attributes) {
this.setObject(object);
this.method = method;
this.attributes = attributes;
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
@Override
protected Object doHandle(Message message, SimpleMethodInvoker invoker) {
if (method.getParameterTypes().length != 1) {
throw new MessagingConfigurationException(
"method must accept exactly one parameter");
}
String channelName = (String) attributes.get(CHANNEL_KEY);
Object retval = null;
Class<?> type = method.getParameterTypes()[0];
if (type.equals(Message.class)) {
retval = invoker.invokeMethod(message);
}
else if (type.equals(message.getPayload().getClass())) {
retval = invoker.invokeMethod(message.getPayload());
}
else {
if (logger.isWarnEnabled()) {
logger.warn("Message payload type did not match expected type of Splitter method '" + this.method.getName() + "'");
}
return message;
}
if (retval == null) {
if (logger.isWarnEnabled()) {
logger.warn("Splitter method '" + this.method.getName() + "' returned null");
}
return null;
}
if (retval instanceof Collection) {
Collection<?> items = (Collection<?>) retval;
int counter = 0;
for (Object item : items) {
if (item instanceof Message) {
Message splitMessage = (Message) item;
splitMessage.getHeader().setCorrelationId(message.getId());
this.sendMessage(splitMessage, channelName);
}
else {
Message splitMessage = new GenericMessage(message.getId() + "#" + (counter++), item);
splitMessage.getHeader().setCorrelationId(message.getId());
this.sendMessage(splitMessage, channelName);
}
}
}
else if (retval.getClass().isArray()) {
int counter = 0;
for (Object item : (Object[]) retval) {
if (item instanceof Message) {
Message splitMessage = (Message) item;
splitMessage.getHeader().setCorrelationId(message.getId());
this.sendMessage(splitMessage, channelName);
}
else {
Message splitMessage = new GenericMessage(message.getId() + "#" + (counter++), item);
splitMessage.getHeader().setCorrelationId(message.getId());
this.sendMessage(splitMessage, channelName);
}
}
}
else {
throw new MessagingConfigurationException(
"splitter method must return either a Collection or array");
}
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;
}
if (logger.isDebugEnabled()) {
logger.debug("sending message to channel '" + channelName + "'");
}
return channel.send(message);
}
}