diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index f6bdc1cb16..e7c182dc8d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -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, AnnotationHandlerCreator> handlerCreators = - new ConcurrentHashMap, AnnotationHandlerCreator>(); + private Log logger = LogFactory.getLog(this.getClass()); + + private Map, MessageHandlerCreator> handlerCreators = + new ConcurrentHashMap, MessageHandlerCreator>(); private MessageBus messageBus; @@ -70,16 +77,17 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor } public void setCustomHandlerCreators( - Map, AnnotationHandlerCreator> customHandlerCreators) { - for (Map.Entry, AnnotationHandlerCreator> entry : customHandlerCreators.entrySet()) { + Map, MessageHandlerCreator> customHandlerCreators) { + for (Map.Entry, 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 handlers = new ArrayList(); ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - for (Class 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 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()); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Handler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Handler.java index f0367f51bd..850ff40031 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Handler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Handler.java @@ -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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Router.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Router.java index 3167eee982..3136d3cdcd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Router.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Router.java @@ -33,6 +33,7 @@ import java.lang.annotation.Target; @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented +@Handler public @interface Router { /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Splitter.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Splitter.java new file mode 100644 index 0000000000..f0f4412cb9 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/Splitter.java @@ -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(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AbstractAnnotationHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/AbstractMessageHandlerCreator.java similarity index 80% rename from spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AbstractAnnotationHandlerCreator.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/config/AbstractMessageHandlerCreator.java index d477197d54..56aa3c3ce3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AbstractAnnotationHandlerCreator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/AbstractMessageHandlerCreator.java @@ -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 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 attributes); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/DefaultAnnotationHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/DefaultMessageHandlerCreator.java similarity index 76% rename from spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/DefaultAnnotationHandlerCreator.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/config/DefaultMessageHandlerCreator.java index b4b3551135..f767594b25 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/DefaultAnnotationHandlerCreator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/DefaultMessageHandlerCreator.java @@ -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 attributes) { return new DefaultMessageHandlerAdapter(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AnnotationHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/MessageHandlerCreator.java similarity index 68% rename from spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AnnotationHandlerCreator.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/config/MessageHandlerCreator.java index b85029646f..f03b1e3604 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/AnnotationHandlerCreator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/MessageHandlerCreator.java @@ -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 attributes); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/RouterAnnotationHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/RouterMessageHandlerCreator.java similarity index 58% rename from spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/RouterAnnotationHandlerCreator.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/config/RouterMessageHandlerCreator.java index 174266302b..1813a08c70 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/annotation/RouterAnnotationHandlerCreator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/RouterMessageHandlerCreator.java @@ -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 attributes) { + return new RouterMessageHandlerAdapter(object, method, attributes); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/config/SplitterMessageHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/SplitterMessageHandlerCreator.java new file mode 100644 index 0000000000..e5c34f873a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/config/SplitterMessageHandlerCreator.java @@ -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 attributes) { + return new SplitterMessageHandlerAdapter(object, method, attributes); + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java index 38a8a6845b..21e475bd4a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RouterMessageHandlerAdapter.java @@ -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 attributes; + private ChannelRegistry channelRegistry; - public RouterMessageHandlerAdapter(Object object, Method method, Router routerAnnotation) { + public RouterMessageHandlerAdapter(Object object, Method method, Map 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)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java new file mode 100644 index 0000000000..177efd631b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java @@ -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 attributes; + + private Method method; + + private ChannelRegistry channelRegistry; + + + public SplitterMessageHandlerAdapter(Object object, Method method, Map 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); + } + +}