diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/Counter.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/Counter.java index 5861ae1a83..b4ec48f247 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/Counter.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/Counter.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. @@ -19,28 +19,21 @@ package org.springframework.integration.samples.oddeven; import java.util.concurrent.atomic.AtomicInteger; import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.MessageSource; import org.springframework.integration.annotation.Polled; -import org.springframework.integration.annotation.Router; /** * @author Mark Fisher */ -@MessageEndpoint +@MessageEndpoint(output="numbers") +@Polled(initialDelay=1000, period=3000) public class Counter { - private AtomicInteger count = new AtomicInteger(); + private final AtomicInteger count = new AtomicInteger(); - @Polled(initialDelay=1000, period=3000) - public int getNumber() { + @MessageSource + public int next() { return count.incrementAndGet(); } - @Router - public String resolveChannel(int i) { - if (i % 2 == 0) { - return "even"; - } - return "odd"; - } - } diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/NumberRouter.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/NumberRouter.java new file mode 100644 index 0000000000..793c2e1914 --- /dev/null +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/NumberRouter.java @@ -0,0 +1,36 @@ +/* + * 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.samples.oddeven; + +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Router; + +/** + * @author Mark Fisher + */ +@MessageEndpoint(input="numbers") +public class NumberRouter { + + @Router + public String resolveChannel(int i) { + if (i % 2 == 0) { + return "even"; + } + return "odd"; + } + +} diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml index 009df03721..afc81b6334 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml @@ -11,8 +11,12 @@ http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - - + + + + + + diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteSubscriber.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/Logger.java similarity index 73% rename from org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteSubscriber.java rename to org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/Logger.java index 7c8f577bfa..a3947141d6 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteSubscriber.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/Logger.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. @@ -16,14 +16,16 @@ package org.springframework.integration.samples.quote; -import org.springframework.integration.annotation.Subscriber; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.MessageTarget; /** * @author Mark Fisher */ -public class QuoteSubscriber { +@MessageEndpoint(input="quotes") +public class Logger { - @Subscriber(channel="quotes") + @MessageTarget public void log(Object o) { System.out.println(o); } diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuotePublisher.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java similarity index 74% rename from org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuotePublisher.java rename to org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java index 6b880f189d..840eba411a 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuotePublisher.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java @@ -22,25 +22,15 @@ import java.util.Random; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.Polled; /** * @author Mark Fisher */ -@MessageEndpoint(output="quotes") -public class QuotePublisher { - - @Polled(period=300) - public String generateTicker() { - char[] chars = new char[3]; - for (int i = 0; i < 3; i++) { - chars[i] = (char) (new Random().nextInt(25) + 65); - } - return new String(chars); - } +@MessageEndpoint(input="tickers", output="quotes") +public class QuoteService { @Handler - public Quote getQuote(String ticker) { + public Quote lookupQuote(String ticker) { BigDecimal price = new BigDecimal(new Random().nextDouble() * 100); return new Quote(ticker, price.setScale(2, RoundingMode.HALF_EVEN)); } diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java new file mode 100644 index 0000000000..b6cfa54b49 --- /dev/null +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java @@ -0,0 +1,41 @@ +/* + * 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.samples.quote; + +import java.util.Random; + +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.MessageSource; +import org.springframework.integration.annotation.Polled; + +/** + * @author Mark Fisher + */ +@MessageEndpoint(output="tickers") +@Polled(period=300) +public class TickerStream { + + @MessageSource + public String nextTicker() { + char[] chars = new char[3]; + for (int i = 0; i < 3; i++) { + chars[i] = (char) (new Random().nextInt(25) + 65); + } + return new String(chars); + } + +} diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml index cf66173db8..72147bb9c7 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml @@ -11,10 +11,13 @@ + - + - + + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java index 2a516a2267..49c0652006 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.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. @@ -41,6 +41,4 @@ public @interface MessageEndpoint { String output() default ""; - int pollPeriod() default 0; - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java new file mode 100644 index 0000000000..0ed341fe4c --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java @@ -0,0 +1,41 @@ +/* + * 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.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; + +import org.springframework.integration.message.Message; + +/** + * Indicates that a method is capable of producing messages. The method must + * accept no parameters and return either a {@link Message} or an Object to + * be passed as the message payload. The enclosing class may also be annotated + * with {@link MessageEndpoint @MessageEndpoint}. + * + * @author Mark Fisher + */ +@java.lang.annotation.Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface MessageSource { + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/DefaultOutput.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java similarity index 81% rename from org.springframework.integration/src/main/java/org/springframework/integration/annotation/DefaultOutput.java rename to org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java index 5548a099a5..ad52b342f1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/DefaultOutput.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java @@ -25,9 +25,10 @@ import java.lang.annotation.RetentionPolicy; import org.springframework.integration.message.Message; /** - * Indicates that a method is capable of sending messages. The method must - * accept a single parameter that is either a {@link Message} or an Object to - * be passed as a message payload. The enclosing class should be annotated with + * Indicates that a method is capable of consuming messages. The method must + * accept a single parameter that is either a {@link Message} or an Object of + * the expected message payload type. The method itself should define a void + * return, and the enclosing class may also be annotated with * {@link MessageEndpoint @MessageEndpoint}. * * @author Mark Fisher @@ -36,6 +37,6 @@ import org.springframework.integration.message.Message; @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented -public @interface DefaultOutput { +public @interface MessageTarget { } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java index 49deba602a..520c5af5f3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.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. @@ -29,15 +29,11 @@ import org.springframework.integration.scheduling.PollingSchedule; /** * Annotation that can be specified at class-level alongside a * {@link MessageEndpoint @MessageEndpoint} annotation in order to provide the - * scheduling information for that endpoint. Alternatively, as a method-level - * annotation, this indicates that a method is capable of providing messages. - * The method must not accept any parameters but can return either a single - * object or collection. The enclosing class should be annotated with - * {@link MessageEndpoint @MessageEndpoint}. + * scheduling information for that endpoint. * * @author Mark Fisher */ -@Target({ElementType.TYPE, ElementType.METHOD}) +@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index f31a2adad3..66d912689e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -32,6 +32,7 @@ import org.springframework.integration.channel.config.PriorityChannelParser; import org.springframework.integration.channel.config.QueueChannelParser; import org.springframework.integration.channel.config.RendezvousChannelParser; import org.springframework.integration.channel.config.ThreadLocalChannelParser; +import org.springframework.integration.config.annotation.AnnotationDrivenParser; import org.springframework.integration.gateway.config.GatewayParser; import org.springframework.integration.router.config.RouterParser; import org.springframework.integration.router.config.SplitterParser; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java deleted file mode 100644 index f110d8d836..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ /dev/null @@ -1,298 +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.config; - -import java.lang.annotation.Annotation; -import java.lang.reflect.Method; -import java.util.ArrayList; -import java.util.Collections; -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.aop.support.AopUtils; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.core.OrderComparator; -import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.annotation.Aggregator; -import org.springframework.integration.annotation.CompletionStrategy; -import org.springframework.integration.annotation.Concurrency; -import org.springframework.integration.annotation.DefaultOutput; -import org.springframework.integration.annotation.Handler; -import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.Polled; -import org.springframework.integration.annotation.Router; -import org.springframework.integration.annotation.Splitter; -import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.dispatcher.DirectChannel; -import org.springframework.integration.endpoint.ConcurrencyPolicy; -import org.springframework.integration.endpoint.HandlerEndpoint; -import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.handler.AbstractMessageHandlerAdapter; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.MessageHandlerChain; -import org.springframework.integration.handler.MethodInvokingTarget; -import org.springframework.integration.handler.config.DefaultMessageHandlerCreator; -import org.springframework.integration.handler.config.MessageHandlerCreator; -import org.springframework.integration.message.MethodInvokingSource; -import org.springframework.integration.router.AggregatingMessageHandler; -import org.springframework.integration.router.CompletionStrategyAdapter; -import org.springframework.integration.router.config.AggregatorMessageHandlerCreator; -import org.springframework.integration.router.config.RouterMessageHandlerCreator; -import org.springframework.integration.router.config.SplitterMessageHandlerCreator; -import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.integration.scheduling.Subscription; -import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; -import org.springframework.util.StringUtils; - -/** - * A {@link BeanPostProcessor} implementation that generates endpoints for - * classes annotated with {@link MessageEndpoint @MessageEndpoint}. - * - * @author Mark Fisher - * @author Marius Bogoevici - */ -public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final Map, MessageHandlerCreator> handlerCreators = new ConcurrentHashMap, MessageHandlerCreator>(); - - 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()); - } - } - - public void afterPropertiesSet() { - this.handlerCreators.put(Handler.class, new DefaultMessageHandlerCreator()); - this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator()); - this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator()); - this.handlerCreators.put(Aggregator.class, new AggregatorMessageHandlerCreator(messageBus)); - } - - public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { - return bean; - } - - public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - Class beanClass = this.getBeanClass(bean); - MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class); - if (endpointAnnotation == null) { - return bean; - } - if (bean instanceof ChannelRegistryAware) { - ((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus); - } - String outputChannelName = endpointAnnotation.output(); - MessageHandlerChain handlerChain = this.createHandlerChain(bean, outputChannelName); - if (handlerChain == null) { - throw new ConfigurationException("@MessageEndpoint has no handler method"); - } - HandlerEndpoint endpoint = new HandlerEndpoint(handlerChain); - Polled polledAnnotation = AnnotationUtils.findAnnotation(beanClass, Polled.class); - this.configureInput(bean, beanName, endpointAnnotation, polledAnnotation, endpoint); - if (StringUtils.hasText(outputChannelName)) { - endpoint.setOutputChannelName(outputChannelName); - } - else { - this.configureOutput(bean, beanName, endpoint); - } - Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(beanClass, Concurrency.class); - if (concurrencyAnnotation != null) { - ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(), - concurrencyAnnotation.maxSize()); - concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); - concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); - endpoint.setConcurrencyPolicy(concurrencyPolicy); - } - this.configureCompletionStrategy(bean, endpoint); - this.messageBus.registerEndpoint(beanName + "-endpoint", endpoint); - return bean; - } - - private void configureInput(final Object bean, final String beanName, MessageEndpoint annotation, - Polled polledAnnotation, final HandlerEndpoint endpoint) { - String channelName = annotation.input(); - if (StringUtils.hasText(channelName)) { - PollingSchedule schedule = null; - if (polledAnnotation != null) { - schedule = new PollingSchedule(polledAnnotation.period()); - schedule.setInitialDelay(polledAnnotation.initialDelay()); - schedule.setFixedRate(polledAnnotation.fixedRate()); - schedule.setTimeUnit(polledAnnotation.timeUnit()); - } - Subscription subscription = new Subscription(channelName, schedule); - endpoint.setSubscription(subscription); - } - ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class); - if (annotation != null) { - Polled polledAnnotation = (Polled) annotation; - int period = polledAnnotation.period(); - long initialDelay = polledAnnotation.initialDelay(); - boolean fixedRate = polledAnnotation.fixedRate(); - MethodInvokingSource source = new MethodInvokingSource(); - source.setObject(bean); - source.setMethod(method.getName()); - DirectChannel channel = new DirectChannel(); - PollingSchedule schedule = new PollingSchedule(period); - schedule.setInitialDelay(initialDelay); - schedule.setFixedRate(fixedRate); - SourceEndpoint sourceEndpoint = new SourceEndpoint(source, channel, schedule); - String channelName = beanName + "-inputChannel"; - messageBus.registerChannel(channelName, channel); - messageBus.registerEndpoint(beanName + "-sourceEndpoint", sourceEndpoint); - Subscription subscription = new Subscription(channel); - endpoint.setSubscription(subscription); - } - } - }); - } - - private void configureOutput(final Object bean, final String beanName, final HandlerEndpoint endpoint) { - ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() { - boolean foundOutput = false; - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class); - if (annotation != null) { - if (foundOutput) { - throw new ConfigurationException("only one @DefaultOutput allowed per endpoint"); - } - MethodInvokingTarget target = new MethodInvokingTarget(); - target.setObject(bean); - target.setMethodName(method.getName()); - target.afterPropertiesSet(); - MessageHandler handler = endpoint.getHandler(); - ((MessageHandlerChain) handler).add(target); - foundOutput = true; - return; - } - } - }); - } - - private void configureCompletionStrategy(final Object bean, final HandlerEndpoint endpoint) { - ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class); - if (annotation != null) { - final MessageHandler endpointHandler = endpoint.getHandler(); - AggregatingMessageHandler aggregatingMessageHandler = null; - if (endpointHandler != null) { - if (endpointHandler instanceof MessageHandlerChain) { - for (MessageHandler handlerInChain : ((MessageHandlerChain) endpointHandler).getHandlers()) { - if (handlerInChain instanceof AggregatingMessageHandler) { - aggregatingMessageHandler = (AggregatingMessageHandler) handlerInChain; - break; - } - } - } - else if (endpointHandler instanceof AggregatingMessageHandler) { - aggregatingMessageHandler = (AggregatingMessageHandler) endpointHandler; - } - } - if (aggregatingMessageHandler == null) { - throw new ConfigurationException( - "@CompletionStrategy supported only when @Aggregator is present"); - } - else { - aggregatingMessageHandler.setCompletionStrategy(new CompletionStrategyAdapter(bean, method)); - } - } - } - }); - } - - @SuppressWarnings("unchecked") - private MessageHandlerChain createHandlerChain(final Object bean, final String outputChannelName) { - 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); - attributes.put(AbstractMessageHandlerAdapter.OUTPUT_CHANNEL_NAME_KEY, outputChannelName); - 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() + "'"); - } - } - 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 ConfigurationException("failed to create handler", e); - } - } - if (handler != null) { - handlers.add(handler); - } - } - } - } - } - }); - if (handlers.size() > 0) { - MessageHandlerChain handlerChain = new MessageHandlerChain(); - Collections.sort(handlers, new OrderComparator()); - for (MessageHandler handler : handlers) { - handlerChain.add(handler); - } - return handlerChain; - } - return null; - } - - private Class getBeanClass(Object bean) { - return AopUtils.getTargetClass(bean); - } - - 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/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java new file mode 100644 index 0000000000..6ca7864370 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java @@ -0,0 +1,123 @@ +/* + * 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.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.DelegatingIntroductionInterceptor; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Polled; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Subscription; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; + +/** + * Base class for post-processing annotated methods. + * + * @author Mark Fisher + */ +public abstract class AbstractAnnotationMethodPostProcessor implements AnnotationMethodPostProcessor { + + protected final Log logger = LogFactory.getLog(this.getClass()); + + private final Class annotationType; + + private final MessageBus messageBus; + + private final ClassLoader beanClassLoader; + + + public AbstractAnnotationMethodPostProcessor(Class annotationType, MessageBus messageBus, ClassLoader beanClassLoader) { + Assert.notNull(annotationType, "Annotation type must not be null."); + Assert.notNull(messageBus, "MessageBus must not be null."); + this.annotationType = annotationType; + this.messageBus = messageBus; + this.beanClassLoader = (beanClassLoader != null) ? beanClassLoader : ClassUtils.getDefaultClassLoader(); + } + + + protected MessageBus getMessageBus() { + return this.messageBus; + } + + public Object postProcess(final Object bean, final String beanName, final Class originalBeanClass) { + final List results = new ArrayList(); + ReflectionUtils.doWithMethods(originalBeanClass, new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation annotation = getAnnotation(method); + if (annotation != null) { + T result = processMethod(bean, method, annotation); + if (result != null) { + results.add(result); + } + } + } + }); + T postProcessedBean = (results.size() > 0) ? this.processResults(results) : null; + if (postProcessedBean == null) { + return bean; + } + ProxyFactory proxyFactory = new ProxyFactory(bean); + proxyFactory.addAdvice(new DelegatingIntroductionInterceptor(postProcessedBean)); + return proxyFactory.getProxy(this.beanClassLoader); + } + + private Annotation getAnnotation(Method method) { + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + for (Annotation annotation : annotations) { + if (annotation.annotationType().equals(this.annotationType) + || annotation.annotationType().isAnnotationPresent(this.annotationType)) { + return annotation; + } + } + return null; + } + + protected Subscription createSubscription(final Object bean, final String beanName, MessageEndpoint annotation, Polled polledAnnotation) { + String channelName = annotation.input(); + if (StringUtils.hasText(channelName)) { + PollingSchedule schedule = null; + if (polledAnnotation != null) { + schedule = new PollingSchedule(polledAnnotation.period()); + schedule.setInitialDelay(polledAnnotation.initialDelay()); + schedule.setFixedRate(polledAnnotation.fixedRate()); + schedule.setTimeUnit(polledAnnotation.timeUnit()); + } + Subscription subscription = new Subscription(channelName, schedule); + return subscription; + } + return null; + } + + + protected abstract T processMethod(Object bean, Method method, Annotation annotation); + + protected abstract T processResults(List results); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java similarity index 75% rename from org.springframework.integration/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java rename to org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java index c8cffbe8ad..b80ef9608b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; import org.w3c.dom.Element; @@ -24,6 +24,7 @@ import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.MessageBusParser; /** * Parser for the annotation-driven element of the integration @@ -39,18 +40,18 @@ public class AnnotationDrivenParser implements BeanDefinitionParser { private static final String SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME = "internal.SubscriberAnnotationPostProcessor"; - private static final String MESSAGE_ENDPOINT_ANNOTATION_POST_PROCESSOR_BEAN_NAME = - "internal.MessageEndpointAnnotationPostProcessor"; + private static final String MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME = + "internal.MessagingAnnotationPostProcessor"; public BeanDefinition parse(Element element, ParserContext parserContext) { - this.createPublisherPostProcessor(parserContext); - this.createSubscriberPostProcessor(parserContext); - this.createMessageEndpointPostProcessor(parserContext); + this.registerPublisherPostProcessor(parserContext); + this.registerSubscriberPostProcessor(parserContext); + this.registerMessagingAnnotationPostProcessor(parserContext); return null; } - private void createPublisherPostProcessor(ParserContext parserContext) { + private void registerPublisherPostProcessor(ParserContext parserContext) { BeanDefinition bd = new RootBeanDefinition(PublisherAnnotationPostProcessor.class); bd.getPropertyValues().addPropertyValue("channelRegistry", new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); @@ -59,7 +60,7 @@ public class AnnotationDrivenParser implements BeanDefinitionParser { parserContext.registerBeanComponent(bcd); } - private void createSubscriberPostProcessor(ParserContext parserContext) { + private void registerSubscriberPostProcessor(ParserContext parserContext) { BeanDefinition bd = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class); bd.getPropertyValues().addPropertyValue("messageBus", new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); @@ -68,12 +69,12 @@ public class AnnotationDrivenParser implements BeanDefinitionParser { parserContext.registerBeanComponent(bcd); } - private void createMessageEndpointPostProcessor(ParserContext parserContext) { - BeanDefinition bd = new RootBeanDefinition(MessageEndpointAnnotationPostProcessor.class); + private void registerMessagingAnnotationPostProcessor(ParserContext parserContext) { + BeanDefinition bd = new RootBeanDefinition(MessagingAnnotationPostProcessor.class); bd.getConstructorArgumentValues().addGenericArgumentValue( new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); BeanComponentDefinition bcd = new BeanComponentDefinition( - bd, MESSAGE_ENDPOINT_ANNOTATION_POST_PROCESSOR_BEAN_NAME); + bd, MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME); parserContext.registerBeanComponent(bcd); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java new file mode 100644 index 0000000000..ef842de69b --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java @@ -0,0 +1,33 @@ +/* + * 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.annotation; + +import org.springframework.integration.endpoint.MessageEndpoint; + +/** + * Strategy interface for post-processing annotated methods. + * + * @author Mark Fisher + */ +public interface AnnotationMethodPostProcessor { + + Object postProcess(Object bean, String beanName, Class originalBeanClass); + + MessageEndpoint createEndpoint(Object bean, String beanName, Class originalBeanClass, + org.springframework.integration.annotation.MessageEndpoint endpointAnnotation); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java new file mode 100644 index 0000000000..92d3b47d0e --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java @@ -0,0 +1,139 @@ +/* + * 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.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.Concurrency; +import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.Polled; +import org.springframework.integration.annotation.Router; +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.ChannelRegistryAware; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.handler.MessageHandlerChain; +import org.springframework.integration.handler.config.DefaultMessageHandlerCreator; +import org.springframework.integration.handler.config.MessageHandlerCreator; +import org.springframework.integration.router.config.AggregatorMessageHandlerCreator; +import org.springframework.integration.router.config.RouterMessageHandlerCreator; +import org.springframework.integration.router.config.SplitterMessageHandlerCreator; +import org.springframework.integration.scheduling.Subscription; +import org.springframework.util.StringUtils; + +/** + * Post-processor for the {@link Handler @Handler} annotation. + * + * @author Mark Fisher + */ +public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor { + + private final Map, MessageHandlerCreator> handlerCreators = + new ConcurrentHashMap, MessageHandlerCreator>(); + + private final MessageHandlerCreator defaultHandlerCreator = new DefaultMessageHandlerCreator(); + + + public HandlerAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) { + super(Handler.class, messageBus, beanClassLoader); + this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator()); + this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator()); + this.handlerCreators.put(Aggregator.class, new AggregatorMessageHandlerCreator(messageBus)); + } + + + public void setCustomHandlerCreators(Map, MessageHandlerCreator> customHandlerCreators) { + for (Map.Entry, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) { + this.handlerCreators.put(entry.getKey(), entry.getValue()); + } + } + + protected MessageHandler processMethod(Object bean, Method method, Annotation annotation) { + MessageHandlerCreator handlerCreator = this.handlerCreators.get(annotation.annotationType()); + if (handlerCreator == null) { + handlerCreator = this.defaultHandlerCreator; + if (logger.isDebugEnabled()) { + logger.debug("No handler creator has been registered for handler annotation '" + + annotation.annotationType() + "', using DefaultMessageHandlerCreator."); + } + } + MessageHandler handler = handlerCreator.createHandler(bean, method, AnnotationUtils.getAnnotationAttributes(annotation)); + if (handler != null) { + if (handler instanceof ChannelRegistryAware) { + ((ChannelRegistryAware) handler).setChannelRegistry(this.getMessageBus()); + } + if (handler instanceof InitializingBean) { + try { + ((InitializingBean) handler).afterPropertiesSet(); + } + catch (Exception e) { + throw new ConfigurationException("failed to initialize handler", e); + } + } + } + return handler; + } + + protected MessageHandler processResults(List results) { + MessageHandlerChain handlerChain = new MessageHandlerChain(); + for (MessageHandler handler : results) { + handlerChain.add(handler); + } + if (handlerChain.getHandlers().size() == 0) { + return null; + } + if (handlerChain.getHandlers().size() == 1) { + return handlerChain.getHandlers().get(0); + } + return handlerChain; + } + + public MessageEndpoint createEndpoint(Object bean, String beanName, Class originalBeanClass, + org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) { + HandlerEndpoint endpoint = new HandlerEndpoint((MessageHandler) bean); + String outputChannelName = endpointAnnotation.output(); + if (StringUtils.hasText(outputChannelName)) { + endpoint.setOutputChannelName(outputChannelName); + } + Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class); + Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation); + if (subscription != null) { + endpoint.setSubscription(subscription); + } + Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class); + if (concurrencyAnnotation != null) { + ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy( + concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize()); + concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); + concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); + endpoint.setConcurrencyPolicy(concurrencyPolicy); + } + return endpoint; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java new file mode 100644 index 0000000000..0090d666d8 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -0,0 +1,108 @@ +/* + * 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.annotation; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.aop.support.AopUtils; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.ChannelRegistryAware; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Source; +import org.springframework.integration.message.Target; +import org.springframework.util.Assert; + +/** + * A {@link BeanPostProcessor} implementation that processes method-level + * messaging annotations such as @Handler, @MessageSource, and @MessageTarget. + * It also generates endpoints for classes annotated with the class-level + * {@link MessageEndpoint @MessageEndpoint} annotation. + * + * @author Mark Fisher + * @author Marius Bogoevici + */ +public class MessagingAnnotationPostProcessor implements BeanPostProcessor, InitializingBean, BeanClassLoaderAware { + + private final MessageBus messageBus; + + private volatile ClassLoader beanClassLoader; + + private Map, AnnotationMethodPostProcessor> postProcessors = new HashMap, AnnotationMethodPostProcessor>(); + + + public MessagingAnnotationPostProcessor(MessageBus messageBus) { + Assert.notNull(messageBus, "MessageBus must not be null."); + this.messageBus = messageBus; + } + + + public void setBeanClassLoader(ClassLoader beanClassLoader) { + this.beanClassLoader = beanClassLoader; + } + + protected MessageBus getMessageBus() { + return this.messageBus; + } + + public void afterPropertiesSet() { + this.postProcessors.put(MessageHandler.class, new HandlerAnnotationPostProcessor(this.messageBus, this.beanClassLoader)); + this.postProcessors.put(Source.class, new SourceAnnotationPostProcessor(this.messageBus, this.beanClassLoader)); + this.postProcessors.put(Target.class, new TargetAnnotationPostProcessor(this.messageBus, this.beanClassLoader)); + } + + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + Object originalBean = bean; + Class beanClass = this.getBeanClass(bean); + MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class); + for (Map.Entry, AnnotationMethodPostProcessor> entry : this.postProcessors.entrySet()) { + AnnotationMethodPostProcessor postProcessor = entry.getValue(); + bean = postProcessor.postProcess(bean, beanName, beanClass); + if (endpointAnnotation != null && entry.getKey().isAssignableFrom(bean.getClass())) { + org.springframework.integration.endpoint.MessageEndpoint endpoint = + postProcessor.createEndpoint(bean, beanName, beanClass, endpointAnnotation); + if (endpoint != null) { + this.messageBus.registerEndpoint(beanName + "." + entry.getKey().getSimpleName() + ".endpoint", endpoint); + } + } + } + if (bean instanceof ChannelRegistryAware) { + ((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus); + } + if (endpointAnnotation != null && bean.equals(originalBean)) { + throw new ConfigurationException("Class [" + beanClass.getName() + + "] is annotated with @MessageEndpoint but contains no source, target, or handler method annotations."); + } + return bean; + } + + protected Class getBeanClass(Object bean) { + return AopUtils.getTargetClass(bean); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PublisherAnnotationPostProcessor.java similarity index 92% rename from org.springframework.integration/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java rename to org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PublisherAnnotationPostProcessor.java index e676befc06..168dcaafa9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PublisherAnnotationPostProcessor.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; @@ -38,9 +38,9 @@ import org.springframework.util.Assert; */ public class PublisherAnnotationPostProcessor implements BeanPostProcessor, BeanClassLoaderAware { - private Class publisherAnnotationType = Publisher.class; + private volatile Class publisherAnnotationType = Publisher.class; - private String channelNameAttribute = "channel"; + private volatile String channelNameAttribute = "channel"; private ChannelRegistry channelRegistry; @@ -86,8 +86,8 @@ public class PublisherAnnotationPostProcessor implements BeanPostProcessor, Bean if (targetClass == null) { return bean; } - if (advisor == null) { - createAdvisor(); + if (this.advisor == null) { + this.createAdvisor(); } if (AopUtils.canApply(this.advisor, targetClass)) { if (bean instanceof Advised) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java new file mode 100644 index 0000000000..4227731e27 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java @@ -0,0 +1,82 @@ +/* + * 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.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.List; + +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.annotation.MessageSource; +import org.springframework.integration.annotation.Polled; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.DirectChannel; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.SourceEndpoint; +import org.springframework.integration.message.MethodInvokingSource; +import org.springframework.integration.message.Source; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.util.StringUtils; + +/** + * Post-processor for classes annotated with {@link MessageSource @MessageSource}. + * + * @author Mark Fisher + */ +public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor> { + + public SourceAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) { + super(MessageSource.class, messageBus, beanClassLoader); + } + + + protected Source processMethod(Object bean, Method method, Annotation annotation) { + MethodInvokingSource source = new MethodInvokingSource(); + source.setObject(bean); + source.setMethod(method.getName()); + return source; + } + + protected Source processResults(List> results) { + if (results.size() > 1) { + throw new ConfigurationException("At most one @MessageSource annotation is allowed per class."); + } + return (results.size() == 1) ? results.get(0) : null; + } + + public MessageEndpoint createEndpoint(Object bean, String beanName, Class originalBeanClass, + org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) { + Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class); + int period = polledAnnotation.period(); + long initialDelay = polledAnnotation.initialDelay(); + boolean fixedRate = polledAnnotation.fixedRate(); + PollingSchedule schedule = new PollingSchedule(period); + schedule.setInitialDelay(initialDelay); + schedule.setFixedRate(fixedRate); + String outputChannelName = endpointAnnotation.output(); + MessageChannel outputChannel = (StringUtils.hasText(outputChannelName)) ? + this.getMessageBus().lookupChannel(outputChannelName) : null; + if (outputChannel == null) { + outputChannel = new DirectChannel(); + this.getMessageBus().registerChannel(beanName + ".output", outputChannel); + } + return new SourceEndpoint((Source) bean, outputChannel, schedule); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java similarity index 98% rename from org.springframework.integration/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java rename to org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java index 6338a14627..f0d2bc095c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java new file mode 100644 index 0000000000..13e395dc07 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java @@ -0,0 +1,79 @@ +/* + * 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.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.List; + +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.annotation.Concurrency; +import org.springframework.integration.annotation.MessageTarget; +import org.springframework.integration.annotation.Polled; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.TargetEndpoint; +import org.springframework.integration.handler.MethodInvokingTarget; +import org.springframework.integration.message.Target; +import org.springframework.integration.scheduling.Subscription; + +/** + * Post-processor for classes annotated with {@link MessageTarget @MessageTarget}. + * + * @author Mark Fisher + */ +public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor { + + public TargetAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) { + super(MessageTarget.class, messageBus, beanClassLoader); + } + + + public Target processMethod(Object bean, Method method, Annotation annotation) { + MethodInvokingTarget target = new MethodInvokingTarget(); + target.setObject(bean); + target.setMethod(method); + return target; + } + + protected Target processResults(List results) { + if (results.size() > 1) { + throw new ConfigurationException("At most one @MessageTarget annotation is allowed per class."); + } + return (results.size() == 1) ? results.get(0) : null; + } + + public MessageEndpoint createEndpoint(Object bean, String beanName, Class originalBeanClass, + org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) { + TargetEndpoint endpoint = new TargetEndpoint((Target) bean); + Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class); + Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation); + endpoint.setSubscription(subscription); + Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class); + if (concurrencyAnnotation != null) { + ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(), + concurrencyAnnotation.maxSize()); + concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); + concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); + endpoint.setConcurrencyPolicy(concurrencyPolicy); + } + return endpoint; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java index 0d831c389e..1439fa3c04 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.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. @@ -16,14 +16,19 @@ package org.springframework.integration.router.config; +import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.Map; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.annotation.CompletionStrategy; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.config.AbstractMessageHandlerCreator; import org.springframework.integration.router.AggregatingMessageHandler; import org.springframework.integration.router.AggregatorAdapter; +import org.springframework.integration.router.CompletionStrategyAdapter; +import org.springframework.util.ReflectionUtils; /** * Creates an {@link AggregatorAdapter AggregatorAdapter} for methods that aggregate messages. @@ -82,7 +87,19 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat messageHandler.setTrackedCorrelationIdCapacity( (Integer) attributes.get(TRACKED_CORRELATION_ID_CAPACITY)); } + this.configureCompletionStrategy(object, messageHandler); return messageHandler; } + private void configureCompletionStrategy(final Object object, final AggregatingMessageHandler handler) { + ReflectionUtils.doWithMethods(object.getClass(), new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class); + if (annotation != null) { + handler.setCompletionStrategy(new CompletionStrategyAdapter(object, method)); + } + } + }); + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/config/SplitterMessageHandlerCreator.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/SplitterMessageHandlerCreator.java index 5ee25a83e9..914b054863 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/config/SplitterMessageHandlerCreator.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/SplitterMessageHandlerCreator.java @@ -19,6 +19,9 @@ package org.springframework.integration.router.config; import java.lang.reflect.Method; import java.util.Map; +import org.springframework.aop.support.AopUtils; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.handler.AbstractMessageHandlerAdapter; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.config.AbstractMessageHandlerCreator; @@ -33,6 +36,12 @@ public class SplitterMessageHandlerCreator extends AbstractMessageHandlerCreator public MessageHandler doCreateHandler(Object object, Method method, Map attributes) { String outputChannelName = (String) attributes.get(AbstractMessageHandlerAdapter.OUTPUT_CHANNEL_NAME_KEY); + if (outputChannelName == null) { + MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(AopUtils.getTargetClass(object), MessageEndpoint.class); + if (endpointAnnotation != null) { + outputChannelName = endpointAnnotation.output(); + } + } return new SplitterMessageHandlerAdapter(object, method, outputChannelName); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationPostProcessorTests.java index 5ac01de977..a200325a35 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationPostProcessorTests.java @@ -37,7 +37,7 @@ public class PublisherAnnotationPostProcessorTests { ITestBean testBean = (ITestBean) context.getBean("testBean"); testBean.test(); MessageChannel channel = (MessageChannel) context.getBean("testChannel"); - Message result = channel.receive(); + Message result = channel.receive(); assertEquals("test", result.getPayload()); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml index 3703ba1f64..4499ac92d2 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml @@ -10,7 +10,7 @@ - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index c8bcbdf4f9..30945c5822 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -26,7 +26,7 @@ import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.ThreadLocalChannel; -import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor; +import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor; import org.springframework.integration.dispatcher.DirectChannel; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.handler.MessageHandler; @@ -69,7 +69,7 @@ public class DirectChannelSubscriptionTests { @Test public void testSendAndReceiveForAnnotatedEndpoint() { - MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(bus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(bus); postProcessor.afterPropertiesSet(); TestEndpoint endpoint = new TestEndpoint(); postProcessor.postProcessAfterInitialization(endpoint, "testEndpoint"); @@ -83,7 +83,7 @@ public class DirectChannelSubscriptionTests { @Test(expected=MessagingException.class) public void testExceptionThrownFromRegisteredEndpoint() { QueueChannel errorChannel = new QueueChannel(); - bus.setErrorChannel(errorChannel); + bus.setErrorChannel(errorChannel); HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { public Message handle(Message message) { throw new RuntimeException("intentional test failure"); @@ -100,7 +100,7 @@ public class DirectChannelSubscriptionTests { public void testExceptionThrownFromAnnotatedEndpoint() { QueueChannel errorChannel = new QueueChannel(); bus.setErrorChannel(errorChannel); - MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(bus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(bus); postProcessor.afterPropertiesSet(); FailingTestEndpoint endpoint = new FailingTestEndpoint(); postProcessor.postProcessAfterInitialization(endpoint, "testEndpoint"); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java deleted file mode 100644 index 8eb56a1dc2..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java +++ /dev/null @@ -1,79 +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.config; - -import java.lang.reflect.Method; -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.HandlerEndpoint; -import org.springframework.integration.handler.MessageHandlerChain; -import org.springframework.integration.router.AggregatingMessageHandler; -import org.springframework.integration.router.CompletionStrategyAdapter; - -/** - * @author Marius Bogoevici - */ -public class CompletionStrategyAnnotationTests { - - @Test - public void testAnnotationWithDefaultSettings() { - ApplicationContext context = new ClassPathXmlApplicationContext( - new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" }); - final String endpointName = "endpointWithDefaultAnnotationAndCustomCompletionStrategy"; - DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, - endpointName); - Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof CompletionStrategyAdapter); - DirectFieldAccessor invokerAccessor = new DirectFieldAccessor(new DirectFieldAccessor( - aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy")).getPropertyValue("invoker")); - Assert.assertSame(context.getBean(endpointName), invokerAccessor.getPropertyValue("object")); - Method completionCheckerMethod = (Method) invokerAccessor.getPropertyValue("method"); - Assert.assertEquals("completionChecker", completionCheckerMethod.getName()); - - } - - @Test(expected=BeanCreationException.class) - public void testInvalidAnnotation() { - new ClassPathXmlApplicationContext(new String[] { - "classpath:/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml" }); - } - - @SuppressWarnings("unchecked") - private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context, - final String endpointName) { - MessageBus messageBus = getMessageBus(context); - HandlerEndpoint endpoint = (HandlerEndpoint) messageBus - .lookupEndpoint(endpointName + "-endpoint"); - MessageHandlerChain messageHandlerChain = (MessageHandlerChain) endpoint.getHandler(); - AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor( - messageHandlerChain).getPropertyValue("handlers")).get(0); - DirectFieldAccessor aggregatingMessageHandlerAccessor = new DirectFieldAccessor(aggregatingMessageHandler); - return aggregatingMessageHandlerAccessor; - } - - private MessageBus getMessageBus(ApplicationContext context) { - MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); - return messageBus; - } -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java index 2522ac28cd..103b1dbf94 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java @@ -35,6 +35,7 @@ import org.springframework.integration.annotation.Subscriber; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.annotation.SubscriberAnnotationPostProcessor; import org.springframework.integration.message.StringMessage; /** diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java similarity index 60% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index c6e26b171a..d205702dc6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -14,31 +14,38 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; -import java.util.List; +import java.lang.reflect.Method; import org.junit.Assert; import org.junit.Test; +import org.springframework.aop.framework.Advised; +import org.springframework.aop.support.AopUtils; +import org.springframework.aop.support.DelegatingIntroductionInterceptor; 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.config.MessageBusParser; import org.springframework.integration.endpoint.HandlerEndpoint; -import org.springframework.integration.handler.MessageHandlerChain; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.router.AggregatingMessageHandler; +import org.springframework.integration.router.CompletionStrategyAdapter; import org.springframework.integration.router.SequenceSizeCompletionStrategy; /** * @author Marius Bogoevici + * @author Mark Fisher */ public class AggregatorAnnotationTests { @Test public void testAnnotationWithDefaultSettings() { ApplicationContext context = new ClassPathXmlApplicationContext( - new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" }); + new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithDefaultAnnotation"; DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, endpointName); @@ -59,7 +66,7 @@ public class AggregatorAnnotationTests { @Test public void testAnnotationWithCustomSettings() { ApplicationContext context = new ClassPathXmlApplicationContext( - new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" }); + new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithCustomizedAnnotation"; DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, endpointName); @@ -79,23 +86,49 @@ public class AggregatorAnnotationTests { aggregatingMessageHandlerAccessor.getPropertyValue("trackedCorrelationIdCapacity")); } - @SuppressWarnings("unchecked") - private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context, - final String endpointName) { - MessageBus messageBus = getMessageBus(context); - HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint(endpointName + "-endpoint"); - MessageHandlerChain messageHandlerChain = (MessageHandlerChain) endpoint.getHandler(); - AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor( - messageHandlerChain).getPropertyValue("handlers")).get(0); - DirectFieldAccessor aggregatingMessageHandlerAccessor = new DirectFieldAccessor(aggregatingMessageHandler); - return aggregatingMessageHandlerAccessor; + @Test + public void testAnnotationWithCustomCompletionStrategy() throws Exception { + ApplicationContext context = new ClassPathXmlApplicationContext( + new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); + final String endpointName = "endpointWithDefaultAnnotationAndCustomCompletionStrategy"; + DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, + endpointName); + Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof CompletionStrategyAdapter); + DirectFieldAccessor invokerAccessor = new DirectFieldAccessor(new DirectFieldAccessor( + aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy")).getPropertyValue("invoker")); + Assert.assertSame(((Advised) context.getBean(endpointName)).getTargetSource().getTarget(), invokerAccessor.getPropertyValue("object")); + Method completionCheckerMethod = (Method) invokerAccessor.getPropertyValue("method"); + Assert.assertEquals("completionChecker", completionCheckerMethod.getName()); + } + + @Test(expected=BeanCreationException.class) + public void testInvalidCompletionStrategyAnnotation() { + new ClassPathXmlApplicationContext(new String[] { + "classpath:/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml" }); + } + + + @SuppressWarnings("unchecked") + private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context, final String endpointName) { + MessageBus messageBus = this.getMessageBus(context); + HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint(endpointName + ".MessageHandler.endpoint"); + MessageHandler handler = endpoint.getHandler(); + try { + if (AopUtils.isAopProxy(handler)) { + DelegatingIntroductionInterceptor interceptor = (DelegatingIntroductionInterceptor) + ((Advised) handler).getAdvisors()[0].getAdvice(); + Object delegate = new DirectFieldAccessor(interceptor).getPropertyValue("delegate"); + return new DirectFieldAccessor(delegate); + } + } + catch (Exception e) { + // will return the accessor for the handler + } + return new DirectFieldAccessor(endpoint.getHandler()); } - - private MessageBus getMessageBus(ApplicationContext context) { - MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); - return messageBus; + return (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java similarity index 76% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index f0ec3594ae..f2d636ea1b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -14,25 +14,32 @@ * limitations under the License. */ -package org.springframework.integration.endpoint.annotation; +package org.springframework.integration.config.annotation; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; +import org.springframework.context.ApplicationContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.ConfigurationException; import org.springframework.integration.annotation.Concurrency; -import org.springframework.integration.annotation.DefaultOutput; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.MessageSource; +import org.springframework.integration.annotation.MessageTarget; import org.springframework.integration.annotation.Polled; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.bus.MessageBus; @@ -40,9 +47,9 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollingSchedule; @@ -51,7 +58,47 @@ import org.springframework.integration.scheduling.Schedule; /** * @author Mark Fisher */ -public class MessageEndpointAnnotationPostProcessorTests { +public class MessagingAnnotationPostProcessorTests { + + @Test + public void testHandlerAnnotation() { + MessageBus messageBus = new MessageBus(); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + HandlerAnnotatedBean bean = new HandlerAnnotatedBean(); + Object result = postProcessor.postProcessAfterInitialization(bean, "testBean"); + assertTrue(result instanceof MessageHandler); + } + + @Test + public void testCustomHandlerAnnotation() { + MessageBus messageBus = new MessageBus(); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + CustomHandlerAnnotatedBean bean = new CustomHandlerAnnotatedBean(); + Object result = postProcessor.postProcessAfterInitialization(bean, "testBean"); + assertTrue(result instanceof MessageHandler); + } + + @Test + public void testSimpleHandlerWithContext() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "handlerAnnotationPostProcessorTests.xml", this.getClass()); + MessageHandler handler = (MessageHandler) context.getBean("simpleHandler"); + Message reply = handler.handle(new StringMessage("world")); + assertEquals("hello world", reply.getPayload()); + } + + @Test + public void testSimpleHandlerEndpointWithContext() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "handlerAnnotationPostProcessorTests.xml", this.getClass()); + MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel"); + MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel"); + inputChannel.send(new StringMessage("foo")); + Message reply = outputChannel.receive(1000); + assertEquals("hello foo", reply.getPayload()); + } @Test public void testSimpleHandler() throws InterruptedException { @@ -104,31 +151,14 @@ public class MessageEndpointAnnotationPostProcessorTests { } @Test - public void testPolledAnnotation() throws InterruptedException { + public void testTargetAnnotation() throws InterruptedException { MessageBus messageBus = new MessageBus(); QueueChannel testChannel = new QueueChannel(); messageBus.registerChannel("testChannel", testChannel); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); - postProcessor.afterPropertiesSet(); - PolledAnnotationTestBean testBean = new PolledAnnotationTestBean(); - postProcessor.postProcessAfterInitialization(testBean, "testBean"); - messageBus.start(); - Message message = testChannel.receive(1000); - assertEquals("test", message.getPayload()); - messageBus.stop(); - } - - @Test - public void testDefaultOutputAnnotation() throws InterruptedException { - MessageBus messageBus = new MessageBus(); - QueueChannel testChannel = new QueueChannel(); - messageBus.registerChannel("testChannel", testChannel); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); CountDownLatch latch = new CountDownLatch(1); - DefaultOutputAnnotationTestBean testBean = new DefaultOutputAnnotationTestBean(latch); + TargetAnnotationTestBean testBean = new TargetAnnotationTestBean(latch); postProcessor.postProcessAfterInitialization(testBean, "testBean"); messageBus.start(); testChannel.send(new StringMessage("foo")); @@ -141,12 +171,11 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test public void testConcurrencyAnnotationWithValues() { MessageBus messageBus = new MessageBus(); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); ConcurrencyAnnotationTestBean testBean = new ConcurrencyAnnotationTestBean(); postProcessor.postProcessAfterInitialization(testBean, "testBean"); - HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean-endpoint"); + HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint"); ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy(); assertEquals(17, concurrencyPolicy.getCoreSize()); assertEquals(42, concurrencyPolicy.getMaxSize()); @@ -156,14 +185,13 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test(expected=IllegalArgumentException.class) public void testPostProcessorWithNullMessageBus() { - new MessageEndpointAnnotationPostProcessor(null); + new MessagingAnnotationPostProcessor(null); } @Test public void testChannelRegistryAwareBean() { MessageBus messageBus = new MessageBus(); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); ChannelRegistryAwareTestBean testBean = new ChannelRegistryAwareTestBean(); assertNull(testBean.getChannelRegistry()); @@ -177,8 +205,7 @@ public class MessageEndpointAnnotationPostProcessorTests { public void testProxiedMessageEndpointAnnotation() { MessageBus messageBus = new MessageBus(); messageBus.setAutoCreateChannels(true); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpoint()); Object proxy = proxyFactory.getProxy(); @@ -195,8 +222,7 @@ public class MessageEndpointAnnotationPostProcessorTests { public void testMessageEndpointAnnotationInherited() { MessageBus messageBus = new MessageBus(); messageBus.setAutoCreateChannels(true); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointSubclass(), "subclass"); messageBus.start(); @@ -211,8 +237,7 @@ public class MessageEndpointAnnotationPostProcessorTests { public void testMessageEndpointAnnotationInheritedWithProxy() { MessageBus messageBus = new MessageBus(); messageBus.setAutoCreateChannels(true); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointSubclass()); Object proxy = proxyFactory.getProxy(); @@ -232,8 +257,7 @@ public class MessageEndpointAnnotationPostProcessorTests { MessageChannel outputChannel = new QueueChannel(); messageBus.registerChannel("inputChannel", inputChannel); messageBus.registerChannel("outputChannel", outputChannel); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl"); messageBus.start(); @@ -246,8 +270,7 @@ public class MessageEndpointAnnotationPostProcessorTests { public void testMessageEndpointAnnotationInheritedFromInterfaceWithAutoCreatedChannels() { MessageBus messageBus = new MessageBus(); messageBus.setAutoCreateChannels(true); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl"); messageBus.start(); @@ -265,8 +288,7 @@ public class MessageEndpointAnnotationPostProcessorTests { MessageChannel outputChannel = new QueueChannel(); messageBus.registerChannel("inputChannel", inputChannel); messageBus.registerChannel("outputChannel", outputChannel); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointImplementation()); Object proxy = proxyFactory.getProxy(); @@ -284,8 +306,7 @@ public class MessageEndpointAnnotationPostProcessorTests { QueueChannel output = new QueueChannel(); messageBus.registerChannel("input", input); messageBus.registerChannel("output", output); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); SplitterAnnotationTestEndpoint endpoint = new SplitterAnnotationTestEndpoint(); postProcessor.postProcessAfterInitialization(endpoint, "endpoint"); @@ -311,8 +332,7 @@ public class MessageEndpointAnnotationPostProcessorTests { MessageBus messageBus = new MessageBus(); QueueChannel testChannel = new QueueChannel(); messageBus.registerChannel("testChannel", testChannel); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); AnnotatedEndpointWithNoHandlerMethod endpoint = new AnnotatedEndpointWithNoHandlerMethod(); postProcessor.postProcessAfterInitialization(endpoint, "endpoint"); @@ -323,12 +343,11 @@ public class MessageEndpointAnnotationPostProcessorTests { MessageBus messageBus = new MessageBus(); QueueChannel testChannel = new QueueChannel(); messageBus.registerChannel("testChannel", testChannel); - MessageEndpointAnnotationPostProcessor postProcessor = - new MessageEndpointAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); AnnotatedEndpointWithPolledAnnotation endpoint = new AnnotatedEndpointWithPolledAnnotation(); postProcessor.postProcessAfterInitialization(endpoint, "testBean"); - HandlerEndpoint processedEndpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean-endpoint"); + HandlerEndpoint processedEndpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint"); Schedule schedule = processedEndpoint.getSubscription().getSchedule(); assertEquals(PollingSchedule.class, schedule.getClass()); PollingSchedule pollingSchedule = (PollingSchedule) schedule; @@ -338,31 +357,31 @@ public class MessageEndpointAnnotationPostProcessorTests { assertEquals(TimeUnit.SECONDS, pollingSchedule.getTimeUnit()); } - - @MessageEndpoint(output="testChannel") - private static class PolledAnnotationTestBean { - - @Polled(period=100) - public String poller() { - return "test"; - } - - @Handler - public Message handle(Message message) { - return message; - } + @Test + public void testMessageSourceAnnotation() { + MessageBus messageBus = new MessageBus(); + QueueChannel testChannel = new QueueChannel(); + messageBus.registerChannel("testChannel", testChannel); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + MessageSourceAnnotationTestBean testBean = new MessageSourceAnnotationTestBean(); + postProcessor.postProcessAfterInitialization(testBean, "testBean"); + messageBus.start(); + Message message = testChannel.receive(1000); + assertEquals("test", message.getPayload()); + messageBus.stop(); } @MessageEndpoint(input="testChannel") - private static class DefaultOutputAnnotationTestBean { + private static class TargetAnnotationTestBean { private String messageText; private CountDownLatch latch; - public DefaultOutputAnnotationTestBean(CountDownLatch latch) { + public TargetAnnotationTestBean(CountDownLatch latch) { this.latch = latch; } @@ -370,12 +389,7 @@ public class MessageEndpointAnnotationPostProcessorTests { return this.messageText; } - @Handler - public Message handle(Message message) { - return message; - } - - @DefaultOutput + @MessageTarget public void countdown(String input) { this.messageText = input; latch.countDown(); @@ -418,7 +432,7 @@ public class MessageEndpointAnnotationPostProcessorTests { } - @MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=25) + @MessageEndpoint(input="inputChannel", output="outputChannel") private static interface SimpleAnnotatedEndpointInterface { String test(String input); } @@ -458,4 +472,43 @@ public class MessageEndpointAnnotationPostProcessorTests { } } + + private static class HandlerAnnotatedBean { + + @Handler + public String test(String s) { + return s + s; + } + + } + + + @Target(ElementType.METHOD) + @Retention(RetentionPolicy.RUNTIME) + @Handler + private static @interface CustomHandler { + } + + + private static class CustomHandlerAnnotatedBean { + + @CustomHandler + public String test(String s) { + return s + s; + } + + } + + + @MessageEndpoint(output="testChannel") + @Polled(period=100) + private static class MessageSourceAnnotationTestBean { + + @MessageSource + public String test() { + return "test"; + } + + } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java similarity index 82% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java index 7185ab547f..8a74a4f78b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java @@ -14,15 +14,16 @@ * limitations under the License. */ -package org.springframework.integration.endpoint.annotation; +package org.springframework.integration.config.annotation; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.endpoint.annotation.ITestEndpoint; /** * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=10) +@MessageEndpoint(input="inputChannel", output="outputChannel") public class SimpleAnnotatedEndpoint implements ITestEndpoint { @Handler diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.java similarity index 95% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.java index 28e610013d..082d8ce196 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; import java.util.ArrayList; import java.util.Collections; @@ -60,7 +60,7 @@ public class TestAnnotatedEndpointWithCompletionStrategy { public boolean completionChecker(List> messages) { return true; } - + public ConcurrentMap> getAggregatedMessages() { return aggregatedMessages; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategyOnly.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategyOnly.java similarity index 67% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategyOnly.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategyOnly.java index 3ae863ca46..e0d005c683 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategyOnly.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategyOnly.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. @@ -14,21 +14,13 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; import java.util.List; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.CompletionStrategy; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.message.Message; -import org.springframework.integration.message.StringMessage; -import org.springframework.integration.router.MessageSequenceComparator; import org.springframework.stereotype.Component; /** @@ -40,7 +32,7 @@ public class TestAnnotatedEndpointWithCompletionStrategyOnly { @CompletionStrategy public boolean checkCompleteness(List> messages) { - throw new UnsupportedOperationException("Not intended to being called"); + throw new UnsupportedOperationException("Not intended to be called"); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java similarity index 97% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java index c5bb2a7b45..2a5881869d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; import java.util.ArrayList; import java.util.Collections; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java similarity index 97% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java index c5106b2178..455996812f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.config; +package org.springframework.integration.config.annotation; import java.util.ArrayList; import java.util.Collections; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/TypeConvertingTestEndpoint.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java similarity index 86% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/TypeConvertingTestEndpoint.java rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java index 2a65362218..c07f722bd2 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/TypeConvertingTestEndpoint.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.endpoint.annotation; +package org.springframework.integration.config.annotation; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; @@ -22,7 +22,7 @@ import org.springframework.integration.annotation.MessageEndpoint; /** * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=10) +@MessageEndpoint(input="inputChannel", output="outputChannel") public class TypeConvertingTestEndpoint { @Handler diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/handlerAnnotationPostProcessorTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/handlerAnnotationPostProcessorTests.xml new file mode 100644 index 0000000000..8a17fdc4be --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/handlerAnnotationPostProcessorTests.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/messageParameterAnnotatedEndpointTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/messageParameterAnnotatedEndpointTests.xml similarity index 100% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/messageParameterAnnotatedEndpointTests.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/messageParameterAnnotatedEndpointTests.xml diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointTests.xml similarity index 77% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointTests.xml index 2bb8a3c774..558707113e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointTests.xml @@ -13,9 +13,9 @@ - + - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml similarity index 69% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml index 0b944a260e..f745f4cbfb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml @@ -8,9 +8,9 @@ - + - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml similarity index 83% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml index 7970222d7a..3763fabf42 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml @@ -21,9 +21,9 @@ + expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpoint.*"/> + expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpointWithCompletionStrategyOnly"/> diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml similarity index 89% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml index e280fd8fc0..988f7595b1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml @@ -21,7 +21,7 @@ + expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpointWithCompletionStrategyOnly"/> diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/typeConvertingEndpointTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml similarity index 77% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/typeConvertingEndpointTests.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml index 9be31517a6..57d9a266f4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/typeConvertingEndpointTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml @@ -13,9 +13,9 @@ - + - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java index 66d1d745e5..92999c499b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java @@ -19,14 +19,16 @@ package org.springframework.integration.endpoint.annotation; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Polled; +import org.springframework.integration.annotation.MessageSource; /** * @author Mark Fisher */ @MessageEndpoint(output="outputChannel") +@Polled(period=100) public class InboundChannelAdapterTestBean { - @Polled(period=100) + @MessageSource public String getName() { return "world"; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java index d569f53a6a..56173982d9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java @@ -24,7 +24,7 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=10) +@MessageEndpoint(input="inputChannel", output="outputChannel") public class MessageParameterAnnotatedEndpoint { @Handler diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/annotation/SimpleHandlerTestBean.java similarity index 57% rename from org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java rename to org.springframework.integration/src/test/java/org/springframework/integration/handler/annotation/SimpleHandlerTestBean.java index 8bcd6becd4..9a9c10db84 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/annotation/SimpleHandlerTestBean.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. @@ -14,26 +14,18 @@ * limitations under the License. */ -package org.springframework.integration.endpoint.annotation; +package org.springframework.integration.handler.annotation; -import org.springframework.integration.annotation.DefaultOutput; import org.springframework.integration.annotation.Handler; -import org.springframework.integration.annotation.MessageEndpoint; /** * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel") -public class OutboundChannelAdapterTestBean { +public class SimpleHandlerTestBean { @Handler - public String sayHello(String name) { - return "hello " + name; - } - - @DefaultOutput - public void sendGreeting(String greeting) { - System.out.println(greeting); + public String sayHello(String input) { + return "hello " + input; } }