diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java index c2cb6837d6..cbf333ed96 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -24,8 +24,7 @@ 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. + * payload to produce multiple messages or payloads. * * @author Mark Fisher */ @@ -35,6 +34,4 @@ import java.lang.annotation.Target; @Handler public @interface Splitter { - String channel(); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java index 1a318e040c..a88b420b68 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java @@ -16,6 +16,9 @@ package org.springframework.integration.config; +import org.w3c.dom.Element; +import org.w3c.dom.NodeList; + import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.BeanComponentDefinition; @@ -27,8 +30,6 @@ import org.springframework.integration.router.AggregatingMessageHandler; import org.springframework.integration.router.AggregatorAdapter; import org.springframework.integration.router.CompletionStrategyAdapter; import org.springframework.util.StringUtils; -import org.w3c.dom.Element; -import org.w3c.dom.NodeList; /** * Parser for the aggregator element of the integration namespace. @@ -80,6 +81,7 @@ public class AggregatorParser implements BeanDefinitionParser { public static final String COMPLETION_STRATEGY_ELEMENT = "completion-strategy"; + public BeanDefinition parse(Element element, ParserContext parserContext) { return parseAggregatorElement(element, parserContext, true); } 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 35062f1b95..8b18198c50 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 @@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentHashMap; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.InitializingBean; @@ -52,6 +53,7 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.handler.AbstractMessageHandlerAdapter; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.handler.config.DefaultMessageHandlerCreator; @@ -83,11 +85,13 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor private final MessageBus messageBus; + public MessageEndpointAnnotationPostProcessor(MessageBus messageBus) { Assert.notNull(messageBus, "'messageBus' must not be null"); this.messageBus = messageBus; } + public void setCustomHandlerCreators(Map, MessageHandlerCreator> customHandlerCreators) { for (Map.Entry, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) { this.handlerCreators.put(entry.getKey(), entry.getValue()); @@ -114,10 +118,16 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor if (bean instanceof ChannelRegistryAware) { ((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus); } - MessageHandlerChain handlerChain = this.createHandlerChain(bean); + String defaultOutputChannelName = endpointAnnotation.defaultOutput(); + MessageHandlerChain handlerChain = this.createHandlerChain(bean, defaultOutputChannelName); DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handlerChain); this.configureInput(bean, beanName, endpointAnnotation, endpoint); - this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint); + if (StringUtils.hasText(defaultOutputChannelName)) { + endpoint.setDefaultOutputChannelName(defaultOutputChannelName); + } + else { + this.configureDefaultOutput(bean, beanName, endpoint); + } Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(beanClass, Concurrency.class); if (concurrencyAnnotation != null) { ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(), @@ -173,16 +183,9 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor }); } - private void configureDefaultOutput(final Object bean, final String beanName, final MessageEndpoint annotation, - final DefaultMessageEndpoint endpoint) { - String channelName = annotation.defaultOutput(); - if (StringUtils.hasText(channelName)) { - endpoint.setDefaultOutputChannelName(channelName); - return; - } + private void configureDefaultOutput(final Object bean, final String beanName, final DefaultMessageEndpoint endpoint) { ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { boolean foundDefaultOutput = false; - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class); if (annotation != null) { @@ -246,14 +249,15 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor } @SuppressWarnings("unchecked") - private MessageHandlerChain createHandlerChain(final Object bean) { + private MessageHandlerChain createHandlerChain(final Object bean, final String defaultOutputChannelName) { final List handlers = new ArrayList(); ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Annotation[] annotations = AnnotationUtils.getAnnotations(method); for (Annotation annotation : annotations) { if (isHandlerAnnotation(annotation)) { - Map attributes = AnnotationUtils.getAnnotationAttributes(annotation); + Map attributes = AnnotationUtils.getAnnotationAttributes(annotation); + attributes.put(AbstractMessageHandlerAdapter.DEFAULT_OUTPUT_CHANNEL_NAME_KEY, defaultOutputChannelName); MessageHandlerCreator handlerCreator = handlerCreators.get(annotation.annotationType()); if (handlerCreator == null) { if (logger.isWarnEnabled()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 8e55b36a0e..3f9bb45240 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -240,7 +240,7 @@ - + @@ -252,12 +252,12 @@ - + - Defines a completion strategy. + Defines a completion strategy. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java index 9ffe276448..23c0f17957 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java @@ -39,6 +39,9 @@ import org.springframework.util.Assert; */ public abstract class AbstractMessageHandlerAdapter implements MessageHandler, Ordered, InitializingBean { + public static final String DEFAULT_OUTPUT_CHANNEL_NAME_KEY = "defaultOutputChannelName"; + + protected final Log logger = LogFactory.getLog(this.getClass()); private volatile T object; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index 428473c4bc..d6460a2d92 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java @@ -33,6 +33,7 @@ public class MessageHandlerChain implements MessageHandler { private final List handlers = new CopyOnWriteArrayList(); + /** * Add a handler to the end of the chain. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java index de89f84aa1..1cdcc6c4a7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java @@ -38,7 +38,7 @@ import org.springframework.util.ReflectionUtils; * @author Marius Bogoevici * @author Mark Fisher */ -public class AggregatorAdapter extends MessageListMethodAdapter implements Aggregator { +public class AggregatorAdapter extends MessageListMethodAdapter implements Aggregator { public AggregatorAdapter(Object object, Method method) { super(object, method); @@ -48,6 +48,7 @@ public class AggregatorAdapter extends MessageListMethodAdapter implements Aggr super(object, methodName); } + public Message aggregate(List> messages) { Object returnedValue = this.executeMethod(messages); if (returnedValue == null) { @@ -59,6 +60,4 @@ public class AggregatorAdapter extends MessageListMethodAdapter implements Aggr return new GenericMessage(returnedValue); } - - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java index 9c58209180..f38a59c3d4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java @@ -23,31 +23,30 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.message.Message; /** - * Aggregator adapter for methods annotated with - * {@link org.springframework.integration.annotation.CompletionStrategy @CompletionStrategy} - * and for 'aggregator' elements that include a 'method' - * attribute (e.g. <aggregator ref="beanReference" method="methodName"/>). + * Adapter for methods annotated with {@link org.springframework.integration.annotation.CompletionStrategy @CompletionStrategy} + * and for 'completion-strategy' elements that include a 'method' + * attribute (e.g. <completion-strategy ref="beanReference" method="methodName"/>). * * @author Marius Bogoevici */ - public class CompletionStrategyAdapter extends MessageListMethodAdapter implements CompletionStrategy { public CompletionStrategyAdapter(Object object, Method method) { super(object, method); - assertMethodReturnsBoolean(); + this.assertMethodReturnsBoolean(); } public CompletionStrategyAdapter(Object object, String methodName) { super(object, methodName); - assertMethodReturnsBoolean(); + this.assertMethodReturnsBoolean(); } + private void assertMethodReturnsBoolean() { if (!Boolean.class.equals(this.getMethod().getReturnType()) && !boolean.class.equals(this.getMethod().getReturnType())) { - throw new ConfigurationException("Method " + getMethod().getName() - + " does not return a boolean value"); + throw new ConfigurationException("Method '" + getMethod().getName() + + "' does not return a boolean value"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java index 0a95bee409..f1509ac52a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java @@ -24,7 +24,6 @@ import java.util.List; import org.springframework.integration.ConfigurationException; import org.springframework.integration.handler.HandlerMethodInvoker; -import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; @@ -39,7 +38,8 @@ public abstract class MessageListMethodAdapter { private final HandlerMethodInvoker invoker; - protected Method method; + protected volatile Method method; + public MessageListMethodAdapter(Object object, String methodName) { Assert.notNull(object, "'object' must not be null"); @@ -57,26 +57,23 @@ public abstract class MessageListMethodAdapter { Assert.notNull(method, "'method' must not be null"); if (method.getParameterTypes().length != 1 || !method.getParameterTypes()[0].equals(List.class)) { throw new ConfigurationException( - "Method must accept exactly one parameter, and it must be a Collection."); + "Method must accept exactly one parameter, and it must be a List."); } this.method = method; this.invoker = new HandlerMethodInvoker(object, this.method.getName()); } - private static boolean isActualTypeParametrizedMessage(Method method) { + private static boolean isActualTypeParameterizedMessage(Method method) { return getCollectionActualType(method) instanceof ParameterizedType - && Message.class.isAssignableFrom((Class) ((ParameterizedType) getCollectionActualType(method)) - .getRawType()); + && Message.class.isAssignableFrom((Class) ((ParameterizedType) getCollectionActualType(method)).getRawType()); } protected final Object executeMethod(List> messages) { - if (isMethodParameterParametrized(this.method) && isHavingActualTypeArguments(this.method) - && (isActualTypeRawMessage(this.method) || isActualTypeParametrizedMessage(this.method))) { + if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method) + && (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) { return this.invoker.invokeMethod(messages); } - else { - return this.invoker.invokeMethod(extractPayloadsFromMessages(messages)); - } + return this.invoker.invokeMethod(extractPayloadsFromMessages(messages)); } private List extractPayloadsFromMessages(List> messages) { @@ -99,7 +96,7 @@ public abstract class MessageListMethodAdapter { return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments().length == 1; } - private static boolean isMethodParameterParametrized(Method method) { + private static boolean isMethodParameterParameterized(Method method) { return method.getGenericParameterTypes().length == 1 && method.getGenericParameterTypes()[0] instanceof ParameterizedType; } @@ -112,4 +109,4 @@ public abstract class MessageListMethodAdapter { this.method = method; } -} \ No newline at end of file +} 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 index 615ca98234..38e00367b5 100644 --- 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 @@ -39,12 +39,9 @@ import org.springframework.util.Assert; */ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware { - public static final String CHANNEL_KEY = "channel"; - - private final Method method; - private final Map attributes; + private final String outputChannelName; private volatile ChannelRegistry channelRegistry; @@ -54,12 +51,12 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdap public SplitterMessageHandlerAdapter(T object, Method method, Map attributes) { Assert.notNull(object, "'object' must not be null"); Assert.notNull(method, "'method' must not be null"); - Assert.isTrue(attributes != null && attributes.get(CHANNEL_KEY) != null, - "the 'channel' attribute is required"); + Assert.isTrue(attributes != null && attributes.get(DEFAULT_OUTPUT_CHANNEL_NAME_KEY) != null, + "The '" + DEFAULT_OUTPUT_CHANNEL_NAME_KEY + "' attribute is required."); this.setObject(object); this.setMethodName(method.getName()); this.method = method; - this.attributes = attributes; + this.outputChannelName = (String) attributes.get(DEFAULT_OUTPUT_CHANNEL_NAME_KEY); } public void setChannelRegistry(ChannelRegistry channelRegistry) { @@ -77,7 +74,6 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdap throw new ConfigurationException( "Splitter 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)) { @@ -100,7 +96,7 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdap Message splitMessage = (item instanceof Message) ? (Message) item : this.createReplyMessage(item, originalMessageHeader); this.prepareMessage(splitMessage, message.getId(), ++sequenceNumber, sequenceSize); - this.sendMessage(splitMessage, channelName); + this.sendMessage(splitMessage, this.outputChannelName); } } else if (retval.getClass().isArray()) { @@ -111,7 +107,7 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdap Message splitMessage = (item instanceof Message) ? (Message) item : this.createReplyMessage(item, originalMessageHeader); this.prepareMessage(splitMessage, message.getId(), ++sequenceNumber, sequenceSize); - this.sendMessage(splitMessage, channelName); + this.sendMessage(splitMessage, this.outputChannelName); } } else { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java index a1c2edaaee..04203304b2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java @@ -1,15 +1,31 @@ +/* + * 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 java.util.List; import org.junit.Assert; import org.junit.Test; + import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.endpoint.ConcurrentHandler; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.router.AggregatingMessageHandler; @@ -34,8 +50,8 @@ public class CompletionStrategyAnnotationTests { @Test(expected=BeanCreationException.class) public void testInvalidAnnotation() { - ApplicationContext context = new ClassPathXmlApplicationContext( - new String[] { "classpath:/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml" }); + new ClassPathXmlApplicationContext(new String[] { + "classpath:/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml" }); } @SuppressWarnings("unchecked") diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java index bb547de7ff..c88e805771 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java @@ -127,7 +127,7 @@ public class CorrelationIdTests { ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("testChannel", testChannel); Map attributes = new HashMap(); - attributes.put("channel", "testChannel"); + attributes.put(AbstractMessageHandlerAdapter.DEFAULT_OUTPUT_CHANNEL_NAME_KEY, "testChannel"); SplitterMessageHandlerAdapter splitter = new SplitterMessageHandlerAdapter( new TestBean(), TestBean.class.getMethod("split", String.class), attributes); splitter.setChannelRegistry(channelRegistry); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java index d98968b90a..cd0166334d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java @@ -32,6 +32,7 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.handler.AbstractMessageHandlerAdapter; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -51,7 +52,7 @@ public class SplitterMessageHandlerAdapterTests { public SplitterMessageHandlerAdapterTests() { this.channelRegistry.registerChannel("testChannel", testChannel); - this.attribs.put(SplitterMessageHandlerAdapter.CHANNEL_KEY, "testChannel"); + this.attribs.put(AbstractMessageHandlerAdapter.DEFAULT_OUTPUT_CHANNEL_NAME_KEY, "testChannel"); } diff --git a/spring-integration-samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java b/spring-integration-samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java index ae7068d92b..cf968c14fa 100644 --- a/spring-integration-samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java +++ b/spring-integration-samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java @@ -24,10 +24,10 @@ import org.springframework.integration.annotation.Splitter; /** * @author Mark Fisher */ -@MessageEndpoint(input="orders") +@MessageEndpoint(input="orders", defaultOutput="drinks") public class OrderSplitter { - @Splitter(channel="drinks") + @Splitter public List split(DrinkOrder order) { return order.getDrinks(); }